gain.utils package

Submodules

gain.utils.chromosome_order module

Natural ordering of contig names, for the info pages’ tables.

A genomic score’s statistics are keyed by whatever the scanned table called its contigs, and a plain string sort puts chr10 twenty rows above chr2. This module turns a contig name into a sort key that orders its digit runs numerically, so a per-chromosome table reads the way a human expects with no interaction.

The key is deliberately genome-agnostic: no hardcoded human contig table, nothing read from a reference genome. It is therefore NOT karyotypic – chrM sorts before chrX, and alt/random contigs interleave with the primary ones by their embedded numbers rather than grouping at the end. See iossifovlab/gain#982 for why that is the accepted trade.

It lives beside the other contig helpers rather than under genomic_resources/statistics/ because it is genome-naming knowledge, not statistics: the reference genome and gene models pages carry per-chromosome tables of their own, and iossifovlab/gain#984 wants this key at the template layer, which imports nothing from genomic_resources today.

Unrelated to CategoricalHistogramConfig.natural_order, which orders a histogram’s categories – ints before strings, then lexicographic – and knows nothing of digit runs.

gain.utils.chromosome_order.natural_chromosome_key(chrom: str) → str

Return a plain string ordering chrom by its digit runs.

Orderable by < alone – no tuple and no comparator – so the same key can be emitted as a template sort attribute, which is what iossifovlab/gain#984 will want it for.

Digit runs are keyed by _numeric_run_key() and everything else is lowercased, so case never decides an order. A digit-count prefix rather than the fixed-width zero padding the issue sketched, for two reasons: padding silently mis-orders any run wider than the width it was given, and the separator that sketch used would sort chrM ahead of chr22, failing the issue’s own criterion.

Dropping leading zeros leaves the key non-injective. chr01 and chr1 share one deliberately; chrUn_GL000195v1 would share one with a hypothetical chrUn_GL195v1 less deliberately. Ties fall through to the sorter, which is stable.

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

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.S3_PRESIGN_EXPIRATION_SECONDS = 604800

How long a presigned s3 url stays valid, in seconds – and therefore how long a pysam or pyBigWig handle opened on it does, since the library re-requests that url on every seek. Every presign in gain passes this: the GRR’s _get_file_url and sign() below, whose callers may hold the handle for the length of an import. The most SigV4 allows; see ADR 0023 (gain#1398) for why the maximum and what a handle older than this does.

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

gain.utils.fs_utils.compression_suffix(filename: str) → str | None

Return the compression suffix (.gz/.bgz) of a filename, or None.

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.endswith_ci(filename: str, suffixes: str | tuple[str, ...]) → bool

Case-insensitively test a filename’s suffix.

The one place the “a suffix means the same thing whatever its case” rule is written down. It exists because the alternative – each site lowering the name itself – is what gain#348 was: two places deciding the same thing about the same file from suffix vocabularies that had silently drifted apart, one of them case-sensitive.

suffixes must be given in lower case; they are matched against the lowered filename.

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

gain.utils.fs_utils.find_ci(filename: str, substring: str) → int

Case-insensitively locate substring, or return -1.

The position-returning companion to endswith_ci(), for a caller that has to splice the original filename at the match rather than just test it. Implemented with a regex rather than lower().find() so the index is an offset into the filename as GIVEN: lower-casing is not guaranteed to preserve length, and the caller rebuilds a name from this offset.

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.

On s3 the url is presigned for S3_PRESIGN_EXPIRATION_SECONDS. If the corresponding filesystem doesn’t support signing then the filename is returned as is.

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

Return the filename without its compression suffix, if any.

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.log_levels module

Custom logging levels for the GAIn package.

TRACE (5): below DEBUG, for the finest-grained diagnostic output. USER_INFO (25): between INFO and WARNING, for messages directed at end users.

gain.utils.log_safety module

Escaping that keeps untrusted text on one log line.

Anything caller-supplied that a message interpolates – a resource name read out of remote GRR content, a resource query, a piece of an annotation config – can carry a line break, and a line break in a rendered message emits a second, fully-formed-looking record that can assert the opposite of what the run found (gain#642, gain#655). This module owns the character set and the escaping; the policy about refusing such text belongs to the boundary that receives it.

gain.utils.log_safety.escape_unsafe_characters(text: str) → str

Render untrusted text safe to interpolate into ONE log line.

\xNN/\uNNNN rather than repr: it leaves every other character untouched, so a reader still sees the text that was written, with only the invisible part made visible. The two widths matter – \x2028 for U+2028 would read as \x20 followed by the literal text 28, which is different (and legitimate) text.

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.bundle_regions(regions: list[BedRegion], budget: int) → list[list[BedRegion]]

Pack consecutive regions into bundles of at most budget bases.

The other way round from split_into_regions(): many small regions become one unit of work. Order is kept, so a bundle is a run of the input; a region is never split, so one longer than the budget is a bundle on its own.

A budget of 0 or less is no budget at all – an unbounded one, which nothing exceeds, so every region goes into a single bundle. Saying it as the limit rather than as a case of its own keeps one packing rule: whatever the loop learns to do later, it does at every budget. A task per region is a budget of 1, since a region is never split.

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.resource_id_order module

The order the repository index page lists resource ids in.

The page’s own controls order ids with JavaScript’s localeCompare: the ID column’s sorter, and the tree view. The rows the page is published with have to be in the same order, or what a reader sees on arrival and what one click on the ID header hands them disagree with nothing having changed (iossifovlab/gain#1351) – and the published order is all a reader has until the search index loads, or for good if it never does.

localeCompare cannot be called from here, and it is not one thing: with no locale named it follows the browser’s, and a Danish one sorts aa after z, a Czech one chr1 after hg38. What this key reproduces is the root collation on the alphabet a resource id is scanned from, [a-zA-Z0-9/._-] – what a browser in English and most Western locales gives: punctuation ahead of digits ahead of letters, letters compared without regard to case first, and lowercase ahead of uppercase only where nothing else separates two ids. A browser in a locale that tailors the alphabet will still see the published order and the clicked order disagree, exactly as the tree and the sorter already do between themselves there. It is deliberately not locale.strxfrm, which would make the published page depend on the build host’s locale, and a rebuild of an unchanged repository is meant to be byte-identical.

gain.utils.resource_id_order.resource_id_collation_key(resource_id: str) → tuple[str, str]

Sort key ordering ids the way the index page’s controls do.

Punctuation before digits before letters, letters without regard to case; where two ids differ only in case, the one with the lowercase letter at the first difference sorts first, as localeCompare has it (alpha before Alpha) – which is what swapping case and comparing does. Total over str: a character outside the alphabet keeps its code point.

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.stringify module

One value rendered as annotation output renders it.

Here rather than in gain.annotation.annotate_utils, where it grew up and is still re-exported from, because the allele score builds its allele keys with it (allele_key()) and a score module cannot import the annotation package without a cycle – annotate_utils pulls in the pipeline factory, which pulls in the annotators, which pull in the scores.

gain.utils.stringify.stringify(value: Any, *, vcf: bool = False) → str

Format the value to a string for human-readable output.

A bool spells yes/no in both sinks; only None takes the sink’s missing-value marker (. in a VCF, "" in a table), so a false flag never reads as an absent one. See ADR 0026.

gain.utils.url_redaction module

Url credential redaction, and the log-record seam that applies it.

The redactors live here – below the GRR, next to log_levels – because gain/__init__ has to import the seam before any module that logs, and the GRR is one of those modules.

Two redactors, and one rule for choosing between them (ADR 0023): a display url takes ``strip_url_userinfo``; a message takes ``strip_url_credentials``. A display url – a protocol’s public url, a cache-hit log line’s path – keeps its query string, which on a stored url is part of the address; a message – an exception’s text, whatever url a library embedded in it – loses userinfo and query both, because a presigned url carries its signature in the query. An architecture test holds the line.

redact_url_userinfo_in_log_records is the fourth remedy that ADR’s gain#1363 amendment records: it wraps logging.LogRecord.getMessage process-wide, so a credential that reaches any log line – gain’s own or fsspec’s – is stripped when the record is formatted, and the emitting site needs to know nothing about the rule. Like log_levels, this module installs its patch as an import side effect, so importing it is the whole of what a bootstrap has to do.

gain.utils.url_redaction.redact_url_userinfo_in_log_records() → None

Make every LogRecord render its message with userinfo stripped.

Wraps logging.LogRecord.getMessage – the one method every stdlib and third-party Formatter asks for the message text – so the redaction runs exactly when a handler formats the record and never at emission. Idempotent: installing over an installed seam changes nothing.

gain.utils.url_redaction.strip_url_credentials(text: str) → str

Strip every url credential this module recognises from text.

The union of the two redactors: a url can carry userinfo AND a query-string signature at the same time, and dropping only one of them still leaks. Each half returns text unchanged, at the cost of one substring test, when its literal is absent – the common case, since a GRR that is neither url-authed nor s3 carries no credential at all.

The order is load-bearing. Userinfo goes first because a password may itself contain ?; strip the query first and https://alice:p?w@host/f.gz becomes https://alice:p – half the password kept and the host, which is what says which GRR failed, gone. Userinfo-first yields https://host/f.gz.

gain.utils.url_redaction.strip_url_userinfo(text: str) → str

Strip user:pass@ userinfo from every scheme://user:pass@host.

The host/port/path and any query string are preserved; only the userinfo is removed. A string with no userinfo is returned unchanged.

The @ test up front is what keeps the log-record seam below cheap: it runs on every formatted log line in the process, almost none of which carry an @, and the regex re-anchors at every letter of a line that has none – a few microseconds a line, against a few nanoseconds for the test.

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