apcamargo

#!/usr/bin/env python

import shutil
from pathlib import Path
from typing import Literal

import pyarrow as pa
import pyarrow.parquet as pq

apcamargo

from typing import Sequence
import math


def find_cutoff(values: Sequence[float]) -> float:
    """
    Determine the cutoff point in a biphasic distribution curve by identifying
    the "bending point" where the curve transitions from slowly growing values 
    to rapidly growing values, using the maximum perpendicular distance method.

apcamargo

#!/usr/bin/env python

import json
import re
import sys
from typing import Generator, Dict, Any, Optional

import requests
from tqdm import tqdm

apcamargo

from pathlib import Path
from typing import Iterator, Optional, Union

import polars as pl
from needletail import parse_fastx_file
from polars.io.plugins import register_io_source


def scan_fastx(fastx_file: Union[str, Path]) -> pl.LazyFrame:
    schema = pl.Schema(

apcamargo

#!/usr/bin/env python

"""
This script processes SAM (Sequence Alignment/Map format) inputs from standard
input and extracts alignment information that is then provided in a tab-separated
table. The following fields are produced: query, target, query_length, query_start,
query_end, target_start, target_end, alignment_length, alignment_identity.

This script was designed for use with SAM files produced by minimap2. However,
it will work with any SAM data that:

apcamargo

#!/usr/bin/env python

import math
import re

import click
from scipy.cluster.hierarchy import fcluster, linkage
from skbio import DistanceMatrix, Protein, TabularMSA, io
from skbio.sequence.distance import hamming

apcamargo

apcamargo

from coloraide import Color


def lighten(
    color: Color,
    amount: float,
) -> Color:
    """
    Lighten a color by a given amount.
    """

apcamargo

import subprocess

def get_assembly_accession(genbank_accession):
    p1 = subprocess.Popen(
        ["elink", "-db", "nuccore", "-target", "assembly", "-id", genbank_accession],
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
    )
    p2 = subprocess.Popen(
        ["efetch", "-format", "docsum"],