gain.utils package

Submodules

gain.utils.cnv_utils module

gain.utils.cnv_utils.cnv_variant_type(variant_type: str) → str | None

gain.utils.cnv_utils.cshl2cnv_variant(location: str, variant: str) → tuple[str, int, int, str]

Parse location and variant into CNV variant.

gain.utils.dae_utils module

gain.utils.dae_utils.cshl2vcf_variant(location: str, variant: str, genome: ReferenceGenome | None) → tuple[str, int, str, str]

gain.utils.dae_utils.dae2vcf_variant(chrom: str, position: int, variant: str, genome: ReferenceGenome | None) → tuple[int, str, str]

Convert a given CSHL-style variant to the VCF format.

gain.utils.dae_utils.join_line(line: list[Any | list[Any]], sep: str = '\t') → str

Join an iterable representing a line into a string.

gain.utils.dae_utils.split_iterable(iterable: Iterable, max_chunk_length: int = 50) → Generator[list, None, None]

Split an iterable into chunks of a list type.

gain.utils.debug_closing module

class gain.utils.debug_closing.HasClose(*args, **kwargs)

Bases: Protocol

Protocol for objects that have a close method.

close() → None

Close the object.

class gain.utils.debug_closing.closing(thing: T)

Bases: AbstractContextManager, Generic[T]

Context to automatically close something at the end of a block.

Code like this:

with closing(<module>.open(<arguments>)) as f:

<block>

is equivalent to this:

f = <module>.open(<arguments>) try:

<block>

finally:

f.close()

close() → None

gain.utils.dict_utils module

gain.utils.dict_utils.recursive_dict_update(input_dict: dict[str, Any], updater_dict: dict[str, Any]) → dict[str, Any]

Recursively update a dictionary with another dictionary.

gain.utils.dict_utils.recursive_dict_update_inplace(input_dict: dict[str, Any], updater_dict: dict[str, Any]) → None

Recursively update a dictionary with another dictionary.

gain.utils.fs_utils module

gain.utils.fs_utils.abspath(filename: str) → str

gain.utils.fs_utils.containing_path(path: str | PathLike) → str

Return url to the resource that contains path.

For file paths this is equivalent to the containing directory. For urls this is equivalent to the containing resource.

gain.utils.fs_utils.copy(dest: str, src: str) → None

Copy a file or directory.

gain.utils.fs_utils.exists(filename: str) → bool

gain.utils.fs_utils.find_directory_with_a_file(filename: str, cwd: str | Path | None = None) → Path | None

Find a directory containing a file.

Starts from current working directory or from a directory passed.

gain.utils.fs_utils.find_subdirectories_with_a_file(filename: str, cwd: str | Path | None = None) → Sequence[Path]

Find a list of subdirectories containing a file.

Starts from current working directory or from a directory passed.

gain.utils.fs_utils.glob(path: str) → list[str]

Find files by glob-matching.

gain.utils.fs_utils.is_compressed_filename(filename: str) → bool

Check if a file is compressed by its extension.

gain.utils.fs_utils.is_s3url(path: str) → bool

gain.utils.fs_utils.join(path: str, *paths: str) → str

gain.utils.fs_utils.modified(filename: str) → datetime

Return the modified timestamp of a file.

gain.utils.fs_utils.rm_file(path: str) → None

Remove a file.

gain.utils.fs_utils.sign(filename: str) → str

Create a signed URL representing the given path.

If the coresponding filesystem doesn’t support signing then the filename is returned as is.

gain.utils.fs_utils.tabix_index_filename(tabix_filename: str) → str | None

Given a Tabix/VCF filename returns a tabix index filename if exists.

gain.utils.helpers module

gain.utils.helpers.camelize_string(data: str) → str

gain.utils.helpers.convert_size(size_bytes: int) → str

Convert an integer representing size in bytes to a human-readable string.

Copied from: https://stackoverflow.com/questions/5194057/better-way-to-convert-file-sizes-in-python

gain.utils.helpers.isnan(val: float | None) → bool

gain.utils.helpers.str2bool(value: str | None) → bool

gain.utils.helpers.study_id_from_path(filepath: str) → str

gain.utils.helpers.to_response_json(data: dict) → dict

Convert a dict or Box to an acceptable response JSON.

gain.utils.processing_pipeline module

class gain.utils.processing_pipeline.Filter

Bases: AbstractContextManager

Base class for all processing pipeline filters.

abstractmethod filter(data: Any) → Any

class gain.utils.processing_pipeline.PipelineProcessor(source: Source, filters: Sequence[Filter])

Bases: AbstractContextManager

A processor that can be used to process variants in a pipeline.

process(regions: Iterable[Region] | None = None) → None

Process a pipeline in batches for the given regions.

process_region(region: Region | None = None) → None

class gain.utils.processing_pipeline.Source

Bases: AbstractContextManager

Base class for all processing pipeline sources.

abstractmethod fetch(region: Region | None = None) → Iterable[Any]

gain.utils.regions module

class gain.utils.regions.BedRegion(chrom: str, start: int, stop: int)

Bases: Region

Represents proper bed regions.

property begin: int

property end: int

static from_str(region: str) → BedRegion

Parse string representation of a region.

property start: int

property stop: int

class gain.utils.regions.Region(chrom: str, start: int | None = None, stop: int | None = None)

Bases: object

Class representing a genomic region.

property begin: int | None

contains(other: Region) → bool

Check if the region contains other region.

property end: int | None

static from_str(region: str) → Region

Parse string representation of a region.

intersection(other: Region) → Region | None

Return intersection of the region with other region.

intersects(other: Region) → bool

Check if the region intersects another.

isin(chrom: str, pos: int) → bool

Check if a genomic position is insde of the region.

property start: int | None

property stop: int | None

to_bed_region() → BedRegion

Convert to BedRegion if possible.

gain.utils.regions.all_regions_from_chrom(regions: list[Region], chrom: str) → list[Region]

Subset of regions in R that are from chr.

gain.utils.regions.bedfile2regions(bed_filename: str) → list[BedRegion]

Transform BED file into list of regions.

gain.utils.regions.calc_bin_begin(bin_len: int, bin_idx: int) → int

Calculates the 1-based start position of the <bin_idx>-th bin of length <bin_len>.

n 2n 3n 4n |_______|_______|_______|

bin_len bin_begin

gain.utils.regions.calc_bin_end(bin_len: int, bin_idx: int) → int

Calculates the 1-based end position of the <bin_idx>-th bin of length <bin_len>.

n 2n 3n 4n |_______|_______|_______|

bin_len bin_end

gain.utils.regions.calc_bin_index(bin_len: int, pos: int) → int

Calculates the index of the <bin_len>-long bin the given 1-based position <pos> falls into.

n 2n 3n 4n |_______|_______|_______|

(bin 0) (bin 1) (bin 2)

gain.utils.regions.coalesce(v1: int | None, v2: int) → int

Return first non-None value.

gain.utils.regions.collapse(source: Sequence[Region], *, is_sorted: bool = False) → list[Region]

Collapse list of regions.

gain.utils.regions.collapse_no_chrom(source: list[BedRegion], *, is_sorted: bool = False) → list[BedRegion]

Collapse by ignoring the chromosome.

Useful when the caller knows that all the regions are from the same chromosome.

gain.utils.regions.connected_component(regions: list[BedRegion]) → Any

Return connected component of regions.

This might be the same as collapse.

gain.utils.regions.difference(regions1: list[Region], regions2: list[Region], *, symmetric: bool = False) → list[Region]

Compute difference between two list of regions.

gain.utils.regions.get_chromosome_length_tabix(tabix_file: TabixFile | VariantFile, chrom: str, step: int = 50000000, precision: int = 500000) → int | None

Return the length of a chromosome (or contig).

Returned value is guarnteed to be larger than the actual contig length.

gain.utils.regions.intersection(regions1: list[Region], regions2: list[Region]) → list[Region]

Compute intersection of two list of regions.

First collapses each for lists of regions s1 and s2 and then find the intersection.

gain.utils.regions.regions2bedfile(regions: list[BedRegion], bed_filename: str) → None

Save list of regions into a BED file.

gain.utils.regions.split_into_regions(chrom: str, chrom_length: int, region_size: int, start: int = 1) → list[Region]

Return a list of regions for a chrom with a given length.

gain.utils.regions.total_length(regions: list[BedRegion]) → int

gain.utils.regions.union(*r: list[Region]) → list[Region]

Collapse many lists of regions.

gain.utils.regions.unique_regions(regions: list[Region]) → list[Region]

Remove duplicated regions.

gain.utils.sql_utils module

gain.utils.sql_utils.fill_query_parameters(query: Any, params: list[Any]) → None

Filll query parameters.

gain.utils.sql_utils.glot_and(left_expr: Any, right_expr: Any) → Any

gain.utils.sql_utils.to_duckdb_transpile(query: Any) → str

gain.utils.stats_collection module

class gain.utils.stats_collection.StatsCollection

Bases: MutableMapping[tuple[str, …], Any]

Helper class for collection of variuos statistics.

This class would be used in the project in places where collection of statistics data about how components of the system work seems appropriate.

It provides a dict-like interface.

The keys are tuples of strings. The values could be anything, but usually they are numbers.

>>> stats = StatsCollection()
>>> stats[("a",)] = 1
>>> stats[("a",)]
1
>>> stats.get(("a", 1))

The keys a treated as a hierarchy. You can get all values whose key’s start match the passed key. For example if you add following: >>> stats[(“b”, “1”)] = 42 >>> stats[(“b”, “2”)] = 43

you can get all values whose keys start with (“b”,…) using: >>> stats[(“b”,)] {(‘b’, ‘1’): 42, (‘b’, ‘2’): 43}

inc(key: tuple[str, ...]) → None

Increment stats value for the specified key.

save(filename: str) → None

Save stats to a file.

gain.utils.variant_utils module

Pure string utilities for variant manipulation.

gain.utils.variant_utils.complement(nucleotides: str) → str

gain.utils.variant_utils.reverse_complement(nucleotides: str) → str

gain.utils.variant_utils.trim_parsimonious(pos: int, ref: str, alt: str) → tuple[int, str, str]

Trim identical nucleotides on both ends and adjust position.

gain.utils.variant_utils.trim_str_left(pos: int, ref: str, alt: str) → tuple[int, str, str]

Trim identical nucleotides prefixes and adjust position accordingly.

gain.utils.variant_utils.trim_str_left_right(pos: int, ref: str, alt: str) → tuple[int, str, str]

gain.utils.variant_utils.trim_str_right(pos: int, ref: str, alt: str) → tuple[int, str, str]

Trim identical nucleotides suffixes and adjust position accordingly.

gain.utils.variant_utils.trim_str_right_left(pos: int, ref: str, alt: str) → tuple[int, str, str]

gain.utils.verbosity_configuration module

Provides common configuration for loggers verbosity.

class gain.utils.verbosity_configuration.VerbosityConfiguration

Bases: object

Defines common configuration of verbosity for loggers.

static adjust_verbosity(loglevel: int) → None

Set logging level according to the verbosity specified.

static set(args: Namespace | dict[str, str]) → None

Read verbosity settings from parsed arguments and sets logger.

static set_arguments(parser: ArgumentParser) → None

Add verbosity arguments to argument parser.

static verbosity(verbosity: int) → int

Get verbosity level from loglevel.

Module contents