gain.annotation package

Submodules

gain.annotation.allele_score_annotator module

The allele_score_annotator.

Annotates with scores keyed by allele – frequencies, pathogenicity predictions and the like – read from an allele_score resource, by exact allele match or reduced over a region.

class gain.annotation.allele_score_annotator.AlleleScoreAnnotator(pipeline: AnnotationPipeline, info: AnnotatorInfo)

Bases: GenomicScoreAnnotatorBase

Annotator for allele-level genomic scores (frequencies, pathogenicity…).

Operates in one of two modes, selected by the mode parameter:

• allele (default): performs an exact chrom/pos/ref/alt lookup and returns the single matching line’s scores. The annotatable must be a VCFAllele; other types receive an empty result.

• region: the score reduces all allele lines that overlap the annotatable’s span, in one streaming walk (AlleleScore.get_allele_scores_in_region_agg). Works with any Annotatable (VCFAllele, Region, CNV, …). An aggregator must be defined for every score attribute, either in the attribute config or as the score’s aggregator default in the resource YAML; an attribute with neither – only a bool score can be in that position – is refused when the pipeline loads, in either mode, because a CNV or a region takes the region path whatever the mode.

Virtual allele attribute

All annotators expose a virtual attribute "allele" (is_default=False) that is synthesised rather than read from the data file.

• In allele mode: returns ["chrom:pos:ref:alt"] for the matched line.

• In region mode: returns the distinct "chrom:pos:ref:alt" strings of the lines that pass the optional allele_filter, in the order the lines were first met – the resource’s own genomic order.

Optionally append score values to each allele string with include_attributes. The string’s format is the score’s, allele_key(), so the two modes cannot drift.

An aggregator named on this attribute reduces nothing, in either mode. It is not a score: its value is the keys the annotator synthesised, and region mode has always answered them beside the reductions rather than as one of them. Exact-match mode used to differ – the base folded its one-element list – and stopped in gain#1133, so the two modes now agree.

allele_filter

An optional annotator-level boolean expression evaluated against each record before it is included in the result. The annotator only resolves the parameter; the expression language belongs to the score, so see GenomicScore.compile_filter() for the operators it admits and what a name may contain.

ACCEPTED_RESOURCE_TYPES: ClassVar[tuple[str, ...]] = ('allele_score',)

The resource types this annotator’s resource_id may name.

An annotator that consumes a typed genomic resource states them here, and resolves its resource through resolve_resource(). Before gain#1329 the same fact was written once per annotator in whatever shape that annotator happened to use – a literal at a call site, a constant, or nothing at all with the check left to whichever constructor met the resource first – and the refusal a reader got for the wrong resource type differed accordingly.

This is the ANNOTATOR’s copy, not the only one: the wildcard expansion in annotation_config keys the same fact on annotator NAME rather than class (gain#1266), and the web editor states it again per configuration field. What is gone is the five different shapes it took inside the annotators. The wildcard map stays a separate statement on purpose – whether a name expands a wildcard is the annotation layer’s policy, not a property of the annotator (docs/adr/0029-wildcard-expandability-is-parser-policy.md, gain#1334) – and a test pins the two against each other.

The first element is the preferred spelling. A tuple rather than a set for that reason, as FRAGMENT_SCORE_TYPES is one: the order is rendered into the refusal, and AnnotationConfigParser.WILDCARD_RESOURCE_TYPES is pinned against element zero rather than against membership (why, in ADR 0029). So an annotator that comes to accept a further spelling APPENDS it.

Two annotators accept two spellings; each warns from the constructor that opens the resource, which still runs after this check passes the spelling through.

Empty means the annotator does not constrain its resource type – the default, because most annotators (effect_annotator, liftover_annotator, chrom_mapping, …) have no single typed resource to constrain. Those never call resolve_resource(), which refuses an empty declaration rather than rejecting every type in turn.

build_score_aggregator_documentation(attr: Attribute) → list[str]

Collect score aggregator documentation.

get_attribute_defaults(spec: AttributeSpec) → dict[str, Any]

Defaults for spec: an aggregator and parameters.

Empty by default. The constructor consults it for every attribute: the aggregator key becomes the aggregator when the configuration names none, and every other key becomes a parameter that the configuration’s own parameters override. Override it when defaults live somewhere other than the spec – a score resource declares its own, for instance.

get_attribute_specs() → dict[str, AttributeSpec]

Return score attribute specs plus the virtual allele.

gain.annotation.allele_score_annotator.build_allele_score_annotator(pipeline: AnnotationPipeline, info: AnnotatorInfo) → Annotator

gain.annotation.annotatable module

class gain.annotation.annotatable.Annotatable(chrom: str, pos: int, pos_end: int, annotatable_type: Type)

Bases: object

Base class for annotatables used in annotation pipeline.

An annotatable is the thing a pipeline annotates: a position, a region or an allele on one chromosome. Every annotatable spans a closed, 1-based interval [pos, pos_end] – both ends inclusive, so a Position has pos_end == pos and len(annotatable) is pos_end - pos + 1. VCFAllele says how it derives pos_end from its alleles.

The canonical spellings are chrom, pos and pos_end – the constructor’s names and the keys to_dict() writes. chromosome, position and end_position are aliases kept for the callers that use them; each reads the same value as its twin. Equality compares the type, the chromosome and both ends.

class Type(*values)

Bases: Enum

Defines annotatable types.

COMPLEX = 5

LARGE_DELETION = 7

LARGE_DUPLICATION = 6

POSITION = 0

REGION = 1

SMALL_DELETION = 4

SMALL_INSERTION = 3

SUBSTITUTION = 2

static from_string(variant: str) → Type

Construct annotatable type from string argument.

property chrom: str

The chromosome name, as given at construction.

property chromosome: str

Alias of chrom.

property end_position: int

Alias of pos_end.

static from_string(value: str) → Annotatable

Deserialize an Annotatable instance from a string value.

property pos: int

The 1-based start of the interval, inclusive.

property pos_end: int

The 1-based end of the interval, inclusive.

Equal to pos for a single position; see the class docstring for the convention and VCFAllele for how an allele’s end is derived.

property position: int

Alias of pos.

abstractmethod to_dict() → dict

Serialize the annotatable to a dictionary.

static tokenize(value: str) → tuple[str, list[str]]

Split the serialized form TYPE(arg1, arg2, ...) into its parts.

Returns the type token and the list of argument tokens, with whitespace stripped from the arguments. Raises ValueError for a value that is not exactly one call-like expression. The inverse of __repr__; from_string() dispatches on the type token and the concrete classes parse the arguments.

class gain.annotation.annotatable.CNVAllele(chrom: str, pos_begin: int, pos_end: int, cnv_type: Type)

Bases: Annotatable

Defines copy number variants annotatable.

static from_string(value: str) → CNVAllele

Deserialize an Annotatable instance from a string value.

to_dict() → dict

Serialize the annotatable to a dictionary.

class gain.annotation.annotatable.Position(chrom: str, pos: int)

Bases: Annotatable

Annotatable class representing a single position in a chromosome.

static from_string(value: str) → Position

Deserialize an Annotatable instance from a string value.

to_dict() → dict

Serialize the annotatable to a dictionary.

class gain.annotation.annotatable.Region(chrom: str, pos_begin: int, pos_end: int)

Bases: Annotatable

Annotatable class representing a region in a chromosome.

static from_string(value: str) → Region

Deserialize an Annotatable instance from a string value.

to_dict() → dict

Serialize the annotatable to a dictionary.

class gain.annotation.annotatable.VCFAllele(chrom: str, pos: int, ref: str, alt: str)

Bases: Annotatable

A small variant in VCF terms: chrom, pos, ref and alt.

The alleles decide both the Annotatable.Type and the interval:

• one base to one base is a SUBSTITUTION, spanning pos alone;

• a one-base reference that the alternative extends (same first base) is a SMALL_INSERTION, spanning pos to pos + 1 – the two bases the insertion falls between;

• a reference longer than one base collapsed to its first base is a SMALL_DELETION, and any other pair is COMPLEX; both span pos to pos + len(ref).

So for a deletion or a complex allele pos_end reaches one base past the last reference base, which sits at pos + len(ref) - 1. Annotators query exactly this span.

The canonical spellings are ref and alt; reference and alternative are aliases. Equality also compares both alleles.

property alt: str

The alternative allele as written in VCF, anchor base included.

property alternative: str

Alias of alt.

static from_string(value: str) → VCFAllele

Deserialize a VCFAllele from its __repr__ form.

Accepts VCFAllele(chrom, pos, ref, alt), and the same four arguments under any small-variant type name (SUBSTITUTION, SMALL_INSERTION, SMALL_DELETION, COMPLEX). Raises ValueError for another type token or argument count. The type is re-derived from the alleles, not taken from the token.

property ref: str

The reference allele as written in VCF, anchor base included.

property reference: str

Alias of ref.

to_dict() → dict

Serialize to type, chrom, pos, ref and alt.

type is the type’s name. pos_end is not written: it is re-derived from the alleles.

gain.annotation.annotate_columns module

Deprecated alias for gain.annotation.annotate_tabular.

gain.annotation.annotate_columns.cli(argv: list[str] | None = None) → None

Entry point for the deprecated annotate_columns CLI.

gain.annotation.annotate_doc module

gain.annotation.annotate_doc.cli(raw_args: list[str] | None = None) → None

Run command line interface for annotate_vcf tool.

gain.annotation.annotate_doc.configure_argument_parser() → ArgumentParser

Construct and configure argument parser.

gain.annotation.annotate_tabular module

gain.annotation.annotate_tabular.annotate_tabular(input_path: str, pipeline: AnnotationPipeline, output_path: str, args: dict[str, Any], *, reference_genome: ReferenceGenome | None = None, region: Region | None = None, attributes_to_delete: Sequence[str] | None = None) → None

Annotate a tabular file using a processing pipeline.

gain.annotation.annotate_tabular.cli(argv: list[str] | None = None) → None

Entry point for running the tabular annotation tool.

gain.annotation.annotate_utils module

gain.annotation.annotate_utils.add_common_annotation_arguments(parser: ArgumentParser) → None

Add common arguments to an annotation command line parser.

gain.annotation.annotate_utils.add_input_files_to_task_graph(args: dict, task_graph: TaskGraph) → None

gain.annotation.annotate_utils.build_output_path(raw_input_path: str, output_path: str | None) → str

Build an output filepath for an annotation tool’s output.

An explicit compression suffix (.gz/.bgz) on the output is preserved. An output named without one inherits (“mirrors”) the input’s compression suffix, so a .bgz input yields a .bgz output and a .gz input a .gz output.

gain.annotation.annotate_utils.cache_pipeline_resources(grr: GenomicResourceRepo, pipeline: AnnotationPipeline, *, workers: int | None = None, progress: bool = True) → None

Cache resources that the given pipeline will use.

gain.annotation.annotate_utils.check_resource_locality(pipeline: AnnotationPipeline, count_rows: Callable[[int], int], *, allow_remote: bool = False) → None

Guard against annotating many variants over non-local resources.

count_rows(limit) returns the number of input rows, capped at limit (short-circuiting so a huge input is never read in full).

Below LOCALITY_WARNING_THRESHOLD rows the guard is silent; between the warning and error thresholds it logs a warning and proceeds; above LOCALITY_ERROR_THRESHOLD it raises ValueError. Passing allow_remote disables the guard entirely.

gain.annotation.annotate_utils.emit_annotation_plan(args: dict[str, Any], pipeline: AnnotationPipeline, grr: GenomicResourceRepo) → None

Print the (re)annotation plan to stderr.

With --reannotate the previous pipeline is loaded and a ReannotationPipeline plan is rendered; otherwise the plain all-ADDED annotation plan is rendered. Printed with print (not a logger) so it is visible at the default WARNING log level.

gain.annotation.annotate_utils.find_nonlocal_resources(pipeline: AnnotationPipeline) → list[tuple[str, str]]

Return (resource_id, scheme) for each non-local pipeline resource.

A resource is local when it is served by a caching protocol (its files are mirrored to disk) or by an fsspec protocol with a file or memory scheme. Everything else (http/https/s3) is non-local and would be queried over the network per variant.

gain.annotation.annotate_utils.get_pipeline_from_context(context: GenomicContext) → AnnotationPipeline

Get the annotation pipeline from the genomic context.

gain.annotation.annotate_utils.handle_default_args(args: dict[str, Any]) → dict[str, Any]

Handle default arguments for annotation command line tools.

gain.annotation.annotate_utils.maybe_wrap_reannotation(pipeline: AnnotationPipeline, args: dict[str, Any], grr: GenomicResourceRepo) → AnnotationPipeline

Wrap pipeline in a ReannotationPipeline if reannotating.

When --reannotate is not given the pipeline is returned unchanged. Otherwise the previous pipeline is loaded, the new pipeline is wrapped in a ReannotationPipeline, and the previous pipeline is closed – the wrapper reuses the live new-pipeline annotators and never touches the previous pipeline after construction.

gain.annotation.annotate_utils.produce_partfile_paths(input_file_path: str, regions: list[Region], work_dir: str) → list[str]

Produce a list of file paths for output region part files.

gain.annotation.annotate_utils.produce_regions(pysam_file: TabixFile, region_size: int) → list[Region]

Given a region size, produce contig regions to annotate by.

gain.annotation.annotate_vcf module

gain.annotation.annotate_vcf.annotate_vcf(input_path: str, pipeline: AnnotationPipeline, output_path: str, args: dict[str, Any], *, region: Region | None = None, attributes_to_delete: Sequence[str] | None = None) → None

Annotate a columns file using a processing pipeline.

gain.annotation.annotate_vcf.cli(argv: list[str] | None = None) → None

Entry point for running the VCF annotation tool.

gain.annotation.annotation_config module

class gain.annotation.annotation_config.AnnotationConfigParser

Bases: object

Parser for annotation configuration.

WILDCARD_EXEMPT_ANNOTATORS: ClassVar[frozenset[str]] = frozenset({'gene_set_annotator'})

The annotators that declare ACCEPTED_RESOURCE_TYPES and still take no wildcard.

Named one at a time, with a reason, rather than by weakening the pin above to “mapped, or not”: an annotator missing from both is a mistake, and a pin that cannot tell the two apart catches neither.

gene_set_annotator is here because the two spellings it accepts are not related by equivalent_resource_types() (see GENE_SET_TYPES), so a wildcard keyed on either one would answer only the gene sets declaring that spelling. Whether search should relate them, and this annotator then take a wildcard, is gain#1365.

WILDCARD_LIMIT = 500

WILDCARD_RESOURCE_TYPES: ClassVar[Mapping[str, str]] = mappingproxy({'position_score': 'position_score', 'position_score_annotator': 'position_score', 'allele_score': 'allele_score', 'allele_score_annotator': 'allele_score', 'fragment_score': 'fragment_score', 'fragment_score_annotator': 'fragment_score', 'cnv_collection': 'fragment_score', 'cnv_collection_annotator': 'fragment_score', 'gene_score_annotator': 'gene_score'})

The annotator names a wildcard resource_id is accepted for, each mapped to the ONE canonical resource type it selects.

Written here rather than derived from what the annotators declare, and kept honest by test_wildcard_annotator_map, which pins every value against the annotator’s own ACCEPTED_RESOURCE_TYPES[0]. Why not derived: docs/adr/0029-wildcard-expandability-is-parser-policy.md.

Canonical, not every accepted spelling: a fragment score has two, and either annotator name must find either of them, but which spellings denote the same kind of resource is a fact about the repository vocabulary. search_resources expands the type it is given through equivalent_resource_types, so tabulating the expansion here too would be a second copy of that rule – one that a type acquiring a second spelling updates in the repository and silently misses here, leaving a wildcard that matches nothing in a repository that does hold the resources (gain#1266).

The legacy keys are deprecated (gain#538) but warn nowhere near here: a wildcard resolves against every resource in the repository, so a warning would fire per candidate rather than per pipeline. FragmentScoreAnnotator.__init__ owns that.

Read-only, like its sibling below. It used to be rebuilt on every call, so an in-place edit could not outlive one; as a class attribute it would, process-wide.

static has_wildcard(string: str) → bool

Ascertain whether a string contains a valid wildcard.

static parse_complete(raw: dict[str, Any], idx: int, grr: GenomicResourceRepo | None = None) → list[AnnotatorInfo]

Parse a full-form annotation config.

static parse_minimal(raw: str, idx: int) → AnnotatorInfo

Parse a minimal-form annotation config.

static parse_raw(pipeline_raw_config: list[dict[str, Any]] | RawFullConfig | None, grr: GenomicResourceRepo | None = None) → tuple[AnnotationPreamble | None, list[AnnotatorInfo]]

Parse raw dictionary annotation pipeline configuration.

static parse_raw_attribute_config(raw_attribute_config: dict[str, Any]) → AttributeConfig

Parse annotation attribute raw configuration.

static parse_raw_attributes(raw_attributes_config: Any) → list[AttributeConfig]

Parse annotator pipeline attribute configuration.

static parse_short(raw: dict[str, Any], idx: int, grr: GenomicResourceRepo | None = None) → list[AnnotatorInfo]

Parse a short-form annotation config.

static parse_str(content: str, source_file_name: str | None = None, grr: GenomicResourceRepo | None = None) → tuple[AnnotationPreamble | None, list[AnnotatorInfo]]

Parse annotation pipeline configuration string.

static query_resources(annotator_type: str, resource_query: str, grr: GenomicResourceRepo) → list[str]

Collect the ids of the resources matching resource_query.

resource_query is an id glob plus an optional label filter, not a resource id – the config key it is read from is spelled resource_id, but by the time it reaches here it has been recognised as a wildcard. It is spelled the same as the search_resources parameter that takes the same language.

Both filters are answered by search_resources: the query language and the resource-type vocabulary live in genomic_resources, and asking it means a wildcard here selects exactly what the same query selects anywhere else.

What this adds is the annotation layer’s policy about the result – an annotator name selects the one resource type it can consume, an annotator that names its resource directly is refused a wildcard outright, a wildcard that selects nothing is a configuration error, a resource two repositories both carry is expanded once, and a wildcard selecting more than WILDCARD_LIMIT resources is refused rather than silently expanded into a pipeline of that size.

exception gain.annotation.annotation_config.AnnotationConfigurationError(message: str | None, other_error: Exception | None = None, error_mark: ErrorMark | None = None)

Bases: Exception

Exception raised for errors in the annotation configuration.

error_mark: ErrorMark | None

message: str | None

class gain.annotation.annotation_config.AnnotationPreamble(summary: 'str', description: 'str', input_reference_genome: 'str | None', input_reference_genome_res: 'GenomicResource | None', metadata: 'dict[str, Any]')

Bases: object

description: str

input_reference_genome: str | None

None when the preamble declares no genome; never "".

input_reference_genome_res: GenomicResource | None

metadata: dict[str, Any]

summary: str

class gain.annotation.annotation_config.AnnotatorInfo(_type: str, attributes: list[AttributeConfig], parameters: ParamsUsageMonitor | dict[str, Any], documentation: str = '', resources: list[GenomicResource] | None = None, annotator_id: str = 'N/A')

Bases: object

Defines annotator configuration.

annotator_id: str

attributes: list[AttributeConfig]

documentation: str = ''

parameters: ParamsUsageMonitor

resources: list[GenomicResource]

to_dict() → dict[str, Any]

Convert annotator info to a configuration dictionary.

type: str

class gain.annotation.annotation_config.Attribute(name: str, source: str, internal: bool | None = None, aggregator: AggregatorSource | None = None, parameters: ParamsUsageMonitor = <factory>, spec: AttributeSpec | None = None, _documentation: str | None = None)

Bases: object

Runtime attribute instance produced by an annotator.

aggregator: AggregatorSource | None = None

property description: str

property documentation: str

fold(values: list[Any]) → Any

Reduce values with the aggregator this attribute NAMES.

The one statement of HOW an attribute reduces, kept beside the name it reduces by. The caller decides WHETHER there is anything to reduce, because the container differs by annotator – a list of a score’s values, a mapping of per-gene values – and must hold an aggregator before asking.

A fresh accumulator per call, never a held one: an aggregator is mutable state, and one built here cannot outlive the fold it was built for. Building costs ~0.26 us against the ~0.06 us of clearing a held instance (measured, gain#1133) – a fifth of a microsecond per folded attribute per variant, paid only where a fold actually happens. The name resolution itself is memoised (gain#1157), so what is left is the object, not the parsing.

get_value_type(*, aggregated: bool = True) → str

Value type produced by this attribute.

Pass aggregated=True (default) when the aggregator is known to have run; the aggregator’s output_value_type then takes precedence over the spec’s declared type. Pass aggregated=False when aggregation was skipped (e.g. a scalar value that bypassed a list aggregator) so that the spec type is returned instead. The raw spec type is always accessible via self.spec.value_type.

The type is read off the aggregator’s NAME – the only thing the attribute holds since gain#1133 – through Aggregator.resolve_class(), which is class-level and builds no accumulator.

internal: bool | None = None

name: str

parameters: ParamsUsageMonitor

source: str

spec: AttributeSpec | None = None

class gain.annotation.annotation_config.AttributeConfig(name: str, source: str, internal: bool | None = None, aggregator: AggregatorDefinition | str | dict[str, ~typing.Any] | None=None, parameters: dict[str, ~typing.Any]=<factory>)

Bases: object

Configuration for an annotator attribute (from pipeline YAML).

aggregator: AggregatorDefinition | str | dict[str, Any] | None = None

as_dict() → dict[str, Any]

Serialize to a config dict, omitting fields that are unset.

internal: bool | None = None

name: str

parameters: dict[str, Any]

source: str

class gain.annotation.annotation_config.ErrorMark(row: int, column: int)

Bases: object

Marks an error position in a file.

column: int

row: int

class gain.annotation.annotation_config.ParamsUsageMonitor(data: dict[str, Any], owner: str | None = None)

Bases: Mapping

Class to monitor usage of annotator parameters.

as_dict() → dict[str, Any]

Return a plain copy of all parameters without tracking.

get_integer(key: str, *, default: int, minimum: float | None = None, maximum: float | None = None, not_a_number_explanation: str | None = None, out_of_range_explanation: str | None = None) → int

get_integer(key: str, *, default: None = None, minimum: float | None = None, maximum: float | None = None, not_a_number_explanation: str | None = None, out_of_range_explanation: str | None = None) → int | None

Read a parameter that has to be a whole number.

A length counted in bases is one: the effect annotator does index arithmetic with what it reads here, so a fractional value is a typo to refuse rather than something to truncate. Otherwise as get_number().

get_number(key: str, *, default: float, minimum: float | None = None, maximum: float | None = None, not_a_number_explanation: str | None = None, out_of_range_explanation: str | None = None) → float

get_number(key: str, *, default: None = None, minimum: float | None = None, maximum: float | None = None, not_a_number_explanation: str | None = None, out_of_range_explanation: str | None = None) → float | None

Read a parameter that has to be a number.

Absent – unwritten, or written with no value – means default, and reading is what DECLARES the key: the lookup goes through item access, so a parameter read here is not an unused one. Everything a number cannot be is refused with an AnnotationConfigurationError naming the key as the user spelled it, because a pipeline is wrong the moment it is written and whoever has to fix it is reading YAML (gain#477).

The answer is a number, not necessarily a float: a whole one stays an int, so a caller needing that type can ask for it with get_integer() and be sure of it.

The two explanations say what THIS parameter is, in the caller’s own words, and are appended to the refusal they are named for. A generic sentence stating the bounds stands in for a missing out_of_range_explanation.

get_unused_keys() → set[str]

Return the set of keys that have not been accessed.

get_used_keys() → set[str]

Return the set of keys that have been accessed.

inject(key: str, value: Any) → None

Add a parameter and mark it as used (for framework injection).

owner

Whose parameters these are – the annotator type, set by the AnnotatorInfo that holds them. It names the annotator in a refusal and does nothing else: parameters are compared and hashed by their data alone, so two monitors holding the same parameters stay equal whatever their owners are.

class gain.annotation.annotation_config.RawFullConfig

Bases: TypedDict

annotators: list[dict[str, Any]]

preamble: RawPreamble

class gain.annotation.annotation_config.RawPreamble

Bases: TypedDict

description: str

input_reference_genome: str

metadata: dict[str, Any]

summary: str

gain.annotation.annotation_factory module

Factory for creation of annotation pipeline.

gain.annotation.annotation_factory.build_annotation_pipeline(config: list[dict[str, Any]] | RawFullConfig, grr: GenomicResourceRepo, *, allow_repeated_attributes: bool = False, work_dir: Path | None = None) → AnnotationPipeline

Build an annotation pipeline.

gain.annotation.annotation_factory.build_pipeline_annotator(pipeline: AnnotationPipeline, annotator_config: AnnotatorInfo, work_dir: Path) → Annotator

Build an annotator for the pipeline.

gain.annotation.annotation_factory.check_for_repeated_attributes_in_annotator(annotator_config: AnnotatorInfo) → None

Check for repeated attributes in annotator configuration.

gain.annotation.annotation_factory.check_for_repeated_attributes_in_pipeline(pipeline: AnnotationPipeline, *, allow_repeated_attributes: bool = False, annotator_config: AnnotatorInfo | None = None) → None

Check for repeated attributes in pipeline configuration.

gain.annotation.annotation_factory.check_for_unused_attribute_parameters(annotator: Annotator) → None

Check each attribute’s parameters for unused keys.

gain.annotation.annotation_factory.check_for_unused_parameters(info: AnnotatorInfo) → None

Check annotator configuration for unused parameters.

gain.annotation.annotation_factory.get_annotator_factory(annotator_type: str) → Callable[[AnnotationPipeline, AnnotatorInfo], Annotator]

Find and return a factory function for creation of an annotator type.

If the specified annotator type is not found, this function raises ValueError exception.

Returns:

the annotator factory for the specified annotator type.

Raises:

ValueError – when can’t find an annotator factory for the specified annotator type.

gain.annotation.annotation_factory.get_available_annotator_types() → list[str]

Return the list of all registered annotator factory types.

gain.annotation.annotation_factory.load_pipeline_from_file(raw_path: str, grr: GenomicResourceRepo, *, allow_repeated_attributes: bool = False, work_dir: Path | None = None) → AnnotationPipeline

Load an annotation pipeline from a configuration file.

gain.annotation.annotation_factory.load_pipeline_from_file_or_resource(arg: str, grr: GenomicResourceRepo, *, allow_repeated_attributes: bool = False, work_dir: Path | None = None) → AnnotationPipeline

Load a pipeline from a file path or a GRR resource id.

Tries to interpret arg as a filesystem path first; on miss, falls back to looking it up as a GRR resource of type annotation_pipeline.

gain.annotation.annotation_factory.load_pipeline_from_yaml(raw: str, grr: GenomicResourceRepo, *, allow_repeated_attributes: bool = False, work_dir: Path | None = None) → AnnotationPipeline

Load an annotation pipeline from a YAML-formatted string.

gain.annotation.annotation_factory.register_annotator_factory(annotator_type: str, factory: Callable[[AnnotationPipeline, AnnotatorInfo], Annotator]) → None

Register additional annotator factory.

By default all annotator factories should be registered at the [gain.annotation.annotators] entry point. All registered factories are loaded automatically. This function should be used if you want to bypass the entry point mechanism and register an additional annotator factory programmatically.

gain.annotation.annotation_factory.resolve_repeated_attributes(pipeline: AnnotationPipeline, repeated_attributes: set[str]) → None

Resolve repeated attributes in pipeline configuration via renaming.

gain.annotation.annotation_genomic_context_cli module

Command line helpers for constructing annotation pipelines.

The utilities in this module complement the generic genomic context providers by supplying annotation pipeline objects. They enable CLI tools to load pipeline definitions from the file system or from genomic resource repositories, and to make the resulting AnnotationPipeline instances available through the shared genomic context mechanism.

class gain.annotation.annotation_genomic_context_cli.CLIAnnotationContextProvider

Bases: GenomicContextProvider

Expose annotation pipeline configuration through CLI options.

The provider allows users to point to an annotation pipeline definition (either as a file path or a genomic resource identifier) and optionally tweak pipeline behaviour via command-line flags. When invoked without a pipeline argument the provider abstains from creating a context so that other providers can supply their default pipelines.

add_argparser_arguments(parser: ArgumentParser, **kwargs: Any) → None

Register arguments that describe the annotation pipeline source.

Parameters:

parser – The parser that should receive the provider specific CLI options.

init(**kwargs: Any) → GenomicContext | None

Materialise a genomic context containing an annotation pipeline.

Parameters:

**kwargs – Keyword arguments parsed from the command line. The provider looks at pipeline, allow_repeated_attributes, and work_dir.

Returns:

A context containing the annotation pipeline, or None when no pipeline could be created (for example when the pipeline argument is omitted).

Return type:

GenomicContext | None

gain.annotation.annotation_genomic_context_cli.get_context_pipeline(context: GenomicContext) → AnnotationPipeline | None

Extract a validated AnnotationPipeline from context.

Parameters:

context – The genomic context from which to retrieve the pipeline object.

Returns:

The pipeline instance or None when the context does not expose a pipeline.

Return type:

AnnotationPipeline | None

Raises:

TypeError – If the context entry is present but does not contain the expected AnnotationPipeline type.

gain.annotation.annotation_pipeline module

Provides annotation pipeline class.

class gain.annotation.annotation_pipeline.AnnotationPipeline(repository: GenomicResourceRepo)

Bases: object

Provides annotation pipeline abstraction.

add_annotator(annotator: Annotator) → None

Append an annotator; it runs after every annotator already added.

Adding to an open pipeline does not open the annotator: the pipeline opens its annotators only in open().

annotate(annotatable: Annotatable | None, context: dict | None = None) → dict

Apply all annotators to an annotatable.

annotators: list[Annotator]

batch_annotate(annotatables: Sequence[Annotatable | None], contexts: list[dict] | None = None, batch_work_dir: str | None = None) → list[dict]

Apply all annotators to a list of annotatables.

close() → None

Close the annotation pipeline.

get_annotator_by_attribute_info(attribute_info: Attribute) → Annotator | None

The annotator producing attribute_info, or None.

Matched by attribute equality, so pass an attribute obtained from this pipeline – get_attribute_info()’s answer, say.

get_attribute_info(attribute_name: str) → Attribute | None

The attribute named attribute_name, or None.

The first match in pipeline order, so a later annotator that reuses a name is shadowed here.

get_attributes() → list[Attribute]

Every attribute every annotator produces, in pipeline order.

get_attributes_by_type(attribute_type: str) → list[Attribute]

The attributes of one attribute_type, in pipeline order.

Attributes without a spec are skipped.

get_info() → list[AnnotatorInfo]

The AnnotatorInfo of every annotator, in pipeline order.

get_resource_ids() → set[str]

The ids of every resource any annotator uses, as one set.

open() → AnnotationPipeline

Open all annotators in the pipeline and mark it as open.

preamble: AnnotationPreamble | None

print() → None

Print the annotation pipeline.

raw: list[dict[str, Any]] | RawFullConfig

repository: GenomicResourceRepo

resolve_attribute_parameter(info: AnnotatorInfo, parameter: str, *, expected_attribute_type: str) → str

Resolve info’s parameter to the name of one of my attributes.

The parameter names an upstream attribute the annotator reads. Refused, as a ValueError, when info has no such parameter, when no annotator in the pipeline produces an attribute of that name, or when the attribute’s AttributeSpec.attribute_type is not expected_attribute_type.

One implementation rather than one per caller because the copies it replaces had drifted apart – a typo fixed twice (gain#1170), a misspelling fixed in one copy and left in the other (gain#1280), and the listing of available attributes in the refusal present in one copy and absent from the others (gain#1490).

class gain.annotation.annotation_pipeline.Annotator(pipeline: AnnotationPipeline | None, info: AnnotatorInfo)

Bases: ABC

An annotator produces a set of attributes for a given annotatable.

The pipeline drives the lifecycle: open() before the first annotate(), close() once at the end. An annotator may assume it is open when asked to annotate, and does not open itself on demand. Implementations extend AnnotatorBase, which handles configuration and the None annotatable, rather than this class directly.

BASE_DOC_URL = 'https://iossifovlab.com/gaindocs/annotation_infrastructure.html'

abstractmethod annotate(annotatable: Annotatable | None, context: dict[str, Any]) → dict[str, Any]

Produce this annotator’s attributes for one annotatable.

Returns a mapping from attribute name (not source) to value, with every attribute in attributes present. annotatable is None when the input row has none – an unparsable variant, a liftover that found nothing – and the answer is then every attribute set to None, never an exception. context holds the attributes of the annotators before this one: read what used_context_attributes declares and do not write to it – the pipeline merges the returned mapping into it. May assume open() has run. An annotator that only works in batches raises NotImplementedError here and overrides batch_annotate().

abstract property attributes: list[Attribute]

The attributes this annotator produces, in output order.

Configured attributes: names, sources, aggregators and parameters already resolved against get_attribute_specs().

batch_annotate(annotatables: Sequence[Annotatable | None], contexts: list[dict[str, Any]], batch_work_dir: str | None = None) → Iterable[dict[str, Any]]

Annotate many annotatables: one result per input, in order.

The default calls annotate() once per pair, lazily, and is correct for every annotator. Override it only when the backend has a genuinely batched path – an external tool run once over a file, say – and keep the same contract: exactly one result per annotatable, in input order; the empty result for a None annotatable; contexts read, not written. batch_work_dir is a scratch directory the caller may offer, None when it does not; the default ignores it.

close() → None

Release what open() acquired and mark the annotator closed.

Safe on an annotator never opened, and safe twice; overrides keep it so and call the base. The pipeline calls it once per annotator and logs, rather than propagates, what it raises.

abstractmethod get_attribute_specs() → dict[str, AttributeSpec]

Every attribute this annotator can produce, keyed by source.

The catalogue the configuration is checked against: a configured attribute whose source is not a key here is refused. Independent of the configuration and of open(). AnnotatorBase calls it from its constructor, so it may use only what the subclass set before delegating there.

get_info() → AnnotatorInfo

The AnnotatorInfo this annotator was built from.

Its type, id, configured attributes, parameters and resources.

is_open() → bool

Whether open() has run and close() has not since.

open() → Annotator

Acquire resources and mark the annotator open; returns self.

The base only flips the flag. Overrides open the resources they query, call the base and return self. Opening an already-open annotator must be harmless.

property resource_ids: set[str]

The ids of resources, as a set.

property resources: list[GenomicResource]

The genomic resources this annotator was configured with.

property used_context_attributes: tuple[str, ...]

Names of upstream attributes this annotator reads from context.

Empty by default. An annotator that reads an attribute another annotator produced – a gene list, say – names it here: the pipeline builds its dependency graph from this tuple, and a reannotation reruns this annotator when a named attribute’s producer changes. Every name must be an attribute of an earlier annotator in the same pipeline.

class gain.annotation.annotation_pipeline.AnnotatorDecorator(child: Annotator)

Bases: Annotator

Defines annotator decorator base class.

property attributes: list[Attribute]

The attributes this annotator produces, in output order.

Configured attributes: names, sources, aggregators and parameters already resolved against get_attribute_specs().

close() → None

Release what open() acquired and mark the annotator closed.

Safe on an annotator never opened, and safe twice; overrides keep it so and call the base. The pipeline calls it once per annotator and logs, rather than propagates, what it raises.

get_attribute_specs() → dict[str, AttributeSpec]

Every attribute this annotator can produce, keyed by source.

The catalogue the configuration is checked against: a configured attribute whose source is not a key here is refused. Independent of the configuration and of open(). AnnotatorBase calls it from its constructor, so it may use only what the subclass set before delegating there.

is_open() → bool

Whether open() has run and close() has not since.

open() → Annotator

Acquire resources and mark the annotator open; returns self.

The base only flips the flag. Overrides open the resources they query, call the base and return self. Opening an already-open annotator must be harmless.

class gain.annotation.annotation_pipeline.AttributeSpec(source: str, value_type: str, description: str, is_default: bool = True, internal_default: bool = False, supports_aggregation: bool = True, attribute_type: str = 'attribute')

Bases: object

Describes a single attribute an annotator can produce.

as_dict() → dict[str, Any]

Serialize to a response dict.

attribute_type: str = 'attribute'

description: str

internal_default: bool = False

is_default: bool = True

source: str

supports_aggregation: bool = True

value_type: str

class gain.annotation.annotation_pipeline.InputAnnotableAnnotatorDecorator(child: Annotator)

Bases: AnnotatorDecorator

Defines annotator decorator to use input annotatable if defined.

annotate(annotatable: Annotatable | None, context: dict[str, Any]) → dict[str, Any]

Produce this annotator’s attributes for one annotatable.

Returns a mapping from attribute name (not source) to value, with every attribute in attributes present. annotatable is None when the input row has none – an unparsable variant, a liftover that found nothing – and the answer is then every attribute set to None, never an exception. context holds the attributes of the annotators before this one: read what used_context_attributes declares and do not write to it – the pipeline merges the returned mapping into it. May assume open() has run. An annotator that only works in batches raises NotImplementedError here and overrides batch_annotate().

static decorate(child: Annotator) → Annotator

property used_context_attributes: tuple[str, ...]

Names of upstream attributes this annotator reads from context.

Empty by default. An annotator that reads an attribute another annotator produced – a gene list, say – names it here: the pipeline builds its dependency graph from this tuple, and a reannotation reruns this annotator when a named attribute’s producer changes. Every name must be an attribute of an earlier annotator in the same pipeline.

class gain.annotation.annotation_pipeline.PlanEntry(name: str, internal: bool, annotator_id: str, reason: str | None = None)

Bases: object

A single attribute entry in a reannotation/annotation plan.

annotator_id: str

internal: bool

name: str

reason: str | None = None

class gain.annotation.annotation_pipeline.ReannotationPipeline(pipeline_new: AnnotationPipeline, pipeline_previous: AnnotationPipeline, *, full_reannotation: bool = False)

Bases: AnnotationPipeline

Provides functionality for reannotation.

annotators: list[Annotator]

format_plan(reference: str | None = None) → str

Render the reannotation plan as human-readable text.

get_attributes() → list[Attribute]

Every attribute every annotator produces, in pipeline order.

infos_new: set[AnnotatorInfo]

infos_rerun: set[AnnotatorInfo]

print_plan(reference: str | None = None, file: IO[str] | None = None) → None

Print the reannotation plan.

rerun_triggers: dict[AnnotatorInfo, tuple[AnnotatorInfo, Attribute]]

class gain.annotation.annotation_pipeline.ReannotationPlan(copied: list[PlanEntry] = <factory>, added: list[PlanEntry] = <factory>, computed: list[PlanEntry] = <factory>, deleted: list[PlanEntry] = <factory>)

Bases: object

Structured description of how a reannotation reuses/recomputes data.

Each bucket is a list of PlanEntry:

• copied: attributes reused unchanged from the input;

• added: attributes of annotators new to the pipeline;

• computed: attributes of unchanged annotators forced to recompute (reason records the triggering dependency);

• deleted: attributes present in the previous pipeline but no longer produced.

added: list[PlanEntry]

computed: list[PlanEntry]

copied: list[PlanEntry]

deleted: list[PlanEntry]

class gain.annotation.annotation_pipeline.ValueTransformAnnotatorDecorator(child: Annotator, value_transformers: dict[str, Callable[[Any], Any]])

Bases: AnnotatorDecorator

Define value transformer annotator decorator.

annotate(annotatable: Annotatable | None, context: dict[str, Any]) → dict[str, Any]

Produce this annotator’s attributes for one annotatable.

Returns a mapping from attribute name (not source) to value, with every attribute in attributes present. annotatable is None when the input row has none – an unparsable variant, a liftover that found nothing – and the answer is then every attribute set to None, never an exception. context holds the attributes of the annotators before this one: read what used_context_attributes declares and do not write to it – the pipeline merges the returned mapping into it. May assume open() has run. An annotator that only works in batches raises NotImplementedError here and overrides batch_annotate().

static decorate(child: Annotator) → Annotator

Apply value transform decorator to an annotator.

gain.annotation.annotation_pipeline.format_annotation_plan(pipeline: AnnotationPipeline) → str

Render a plain annotation pipeline as an all-ADDED plan.

gain.annotation.annotation_pipeline.print_annotation_plan(pipeline: AnnotationPipeline, file: IO[str] | None = None) → None

Print a plain annotation pipeline plan.

gain.annotation.annotator_base module

Provides base class for annotators.

class gain.annotation.annotator_base.AnnotatedValues

Bases: dict[str, Any]

The finished answer of a _do_annotate, keyed by ATTRIBUTE NAME.

The seam’s one shape (gain#1130, gain#1134). The keys are attribute names rather than sources because a source exposed twice with two aggregators has two different finished values, which a source-keyed mapping has nowhere to put. Nothing checks the type at run time; it is a contract stated on _do_annotate’s signature and held by the type checker.

Read the names when you ANSWER, never in ``__init__``. The one statement of the rule every annotator building one of these has to follow, kept here rather than in each of them. A pipeline naming one attribute twice renames the later ones – annotation_factory.resolve_repeated_attributes – and it does so AFTER every annotator has been constructed. So an annotator that captured attr.name while building its queries would key its answers by names the pipeline has since moved away from, and the attributes it renamed would come back empty. Whatever a query list caches, it must not cache names; self._attributes is walked again at annotate time and the names read off it then.

class gain.annotation.annotator_base.AnnotatorBase(pipeline: AnnotationPipeline | None, info: AnnotatorInfo)

Bases: Annotator

Base implementation of the Annotator class.

The class every in-tree annotator extends. Its constructor checks the configured attributes against get_attribute_specs(), resolves each one’s name, aggregator and parameters (consulting get_attribute_defaults()), and requires a work_dir parameter. A subclass implements get_attribute_specs() and _do_annotate; overrides _do_batch_annotate when it has a batched path; and overrides get_attribute_defaults(), open() and close() when it has defaults or resources. annotate() and batch_annotate() are left alone, except by a batch-only annotator, which makes annotate() refuse.

ACCEPTED_RESOURCE_TYPES: ClassVar[tuple[str, ...]] = ()

The resource types this annotator’s resource_id may name.

An annotator that consumes a typed genomic resource states them here, and resolves its resource through resolve_resource(). Before gain#1329 the same fact was written once per annotator in whatever shape that annotator happened to use – a literal at a call site, a constant, or nothing at all with the check left to whichever constructor met the resource first – and the refusal a reader got for the wrong resource type differed accordingly.

This is the ANNOTATOR’s copy, not the only one: the wildcard expansion in annotation_config keys the same fact on annotator NAME rather than class (gain#1266), and the web editor states it again per configuration field. What is gone is the five different shapes it took inside the annotators. The wildcard map stays a separate statement on purpose – whether a name expands a wildcard is the annotation layer’s policy, not a property of the annotator (docs/adr/0029-wildcard-expandability-is-parser-policy.md, gain#1334) – and a test pins the two against each other.

The first element is the preferred spelling. A tuple rather than a set for that reason, as FRAGMENT_SCORE_TYPES is one: the order is rendered into the refusal, and AnnotationConfigParser.WILDCARD_RESOURCE_TYPES is pinned against element zero rather than against membership (why, in ADR 0029). So an annotator that comes to accept a further spelling APPENDS it.

Two annotators accept two spellings; each warns from the constructor that opens the resource, which still runs after this check passes the spelling through.

Empty means the annotator does not constrain its resource type – the default, because most annotators (effect_annotator, liftover_annotator, chrom_mapping, …) have no single typed resource to constrain. Those never call resolve_resource(), which refuses an empty declaration rather than rejecting every type in turn.

annotate(annotatable: Annotatable | None, context: dict[str, Any]) → dict[str, Any]

Answer through _do_annotate; the empty result for None.

Subclasses implement _do_annotate instead of overriding this: the None annotatable is handled here, so _do_annotate never sees one. A batch-only annotator overrides it to raise NotImplementedError.

attribute_specs: dict[str, AttributeSpec]

property attributes: list[Attribute]

The configured attributes, in configuration order.

With no attributes configured, every spec marked is_default stands in, under its source name.

batch_annotate(annotatables: Sequence[Annotatable | None], contexts: list[dict[str, Any]], batch_work_dir: str | None = None) → Sequence[dict[str, Any]]

Answer through _do_batch_annotate: one result per annotatable.

Subclasses with a batched backend override _do_batch_annotate instead, whose default loops _do_annotate and handles the None annotatables itself.

get_attribute_defaults(spec: AttributeSpec) → dict[str, Any]

Defaults for spec: an aggregator and parameters.

Empty by default. The constructor consults it for every attribute: the aggregator key becomes the aggregator when the configuration names none, and every other key becomes a parameter that the configuration’s own parameters override. Override it when defaults live somewhere other than the spec – a score resource declares its own, for instance.

open() → Annotator

Create work_dir and mark the annotator open; returns self.

Overrides that open resources call this and return self.

static resolve_input_gene_list(pipeline: AnnotationPipeline, info: AnnotatorInfo) → str

Resolve this annotator’s input_gene_list to an attribute name.

The name of an upstream attribute holding the genes the annotator reads, checked to exist in pipeline and to be marked gene_list – the attribute type every gene list the effect annotators produce carries, and the one the web editor offers for this parameter. A bare object attribute is not enough: object is what any structured attribute is stored as.

Beside resolve_resource() for the same reason it is a method here at all: its two callers, the gene-score and gene-set builders, resolve the name to hand to a constructor that has not run yet. The lookup itself is AnnotationPipeline.resolve_attribute_parameter(), shared with the input_annotatable decorator – see there for why.

classmethod resolve_resource(pipeline: AnnotationPipeline, info: AnnotatorInfo) → GenomicResource

Resolve this annotator’s resource_id to a resource it takes.

A classmethod because two of the five callers – the gene-score and gene-set builders – resolve the resource to hand to a constructor that has not run yet; the other three call it as self.resolve_resource(...) from __init__.

Lives on AnnotatorBase rather than on the genomic SCORE base the free function it replaces used to sit beside: two of those callers are not score annotators, and importing the score machinery to reach a type check would be the wrong dependency.

work_dir: Path

gain.annotation.annotator_base.fold_own_values(attributes: Sequence[Attribute], values: Mapping[str, Any]) → AnnotatedValues

Answer an annotator’s OWN values by attribute, each one folded.

For the annotators whose values are their own rather than a score’s record stream – a gene list, a set intersection, one entry per prediction request – so there is no folding read to move the reduction into. Each attribute takes its source’s value, reduced by the aggregator it names (Attribute.fold()), under the attribute’s NAME.

Only a list is folded. A scalar, a None, an absent source pass through, as does any attribute naming no aggregator: an aggregator says how to reduce MANY values and there is nothing to reduce. That is what the base’s own fold did before gain#1133 retired it.

This is a function rather than a method for the reason gain#1133 exists: the BASE does not aggregate; an annotator that reduces calls this itself before it answers.

gain.annotation.chrom_mapping_annotator module

class gain.annotation.chrom_mapping_annotator.ChromMappingAnnotator(pipeline: AnnotationPipeline, info: AnnotatorInfo)

Bases: AnnotatorBase

Annotator for adjusting chromosome values.

get_attribute_specs() → dict[str, AttributeSpec]

Every attribute this annotator can produce, keyed by source.

The catalogue the configuration is checked against: a configured attribute whose source is not a key here is refused. Independent of the configuration and of open(). AnnotatorBase calls it from its constructor, so it may use only what the subclass set before delegating there.

gain.annotation.chrom_mapping_annotator.build_chrom_mapping_annotator(pipeline: AnnotationPipeline, info: AnnotatorInfo) → Annotator

gain.annotation.debug_annotator module

class gain.annotation.debug_annotator.HelloWorldAnnotator(pipeline: AnnotationPipeline | None, info: AnnotatorInfo)

Bases: AnnotatorBase

Defines example annotator.

get_attribute_specs() → dict[str, AttributeSpec]

Every attribute this annotator can produce, keyed by source.

The catalogue the configuration is checked against: a configured attribute whose source is not a key here is refused. Independent of the configuration and of open(). AnnotatorBase calls it from its constructor, so it may use only what the subclass set before delegating there.

gain.annotation.debug_annotator.build_annotator(pipeline: AnnotationPipeline, info: AnnotatorInfo) → Annotator

Create an example hello world annotator.

gain.annotation.docker_annotator module

class gain.annotation.docker_annotator.DockerAnnotator(pipeline: AnnotationPipeline | None, info: AnnotatorInfo)

Bases: AnnotatorBase

Base class for annotators that use docker containers.

open() → Annotator

Create work_dir and mark the annotator open; returns self.

Overrides that open resources call this and return self.

abstractmethod run(**kwargs: Any) → None

gain.annotation.effect_annotator module

class gain.annotation.effect_annotator.EffectAnnotatorAdapter(pipeline: AnnotationPipeline, info: AnnotatorInfo)

Bases: AnnotatorBase

Adapts effect annotator to be used in annotation infrastructure.

close() → None

Release what open() acquired and mark the annotator closed.

Safe on an annotator never opened, and safe twice; overrides keep it so and call the base. The pipeline calls it once per annotator and logs, rather than propagates, what it raises.

get_attribute_defaults(spec: AttributeSpec) → dict[str, Any]

Defaults for spec: an aggregator and parameters.

Empty by default. The constructor consults it for every attribute: the aggregator key becomes the aggregator when the configuration names none, and every other key becomes a parameter that the configuration’s own parameters override. Override it when defaults live somewhere other than the spec – a score resource declares its own, for instance.

get_attribute_specs() → dict[str, AttributeSpec]

Every attribute this annotator can produce, keyed by source.

The catalogue the configuration is checked against: a configured attribute whose source is not a key here is refused. Independent of the configuration and of open(). AnnotatorBase calls it from its constructor, so it may use only what the subclass set before delegating there.

open() → Annotator

Create work_dir and mark the annotator open; returns self.

Overrides that open resources call this and return self.

gain.annotation.effect_annotator.build_effect_annotator(pipeline: AnnotationPipeline, info: AnnotatorInfo) → Annotator

gain.annotation.fragment_score_annotator module

gain.annotation.fragment_score_annotator.FRAGMENT_FILTER_PARAMETER = 'fragment_filter'

Preferred spelling of the fragment-filter parameter.

class gain.annotation.fragment_score_annotator.FragmentScoreAnnotator(pipeline: AnnotationPipeline, info: AnnotatorInfo)

Bases: AnnotatorBase

Annotator over a fragment score.

Configured as fragment_score / fragment_score_annotator, with fragment_filter: selecting which fragments count. The older cnv_collection / cnv_collection_annotator / cnv_filter spellings resolve here too, deprecated: each one logs a warning naming the pipeline’s annotator and the release it stops being accepted in. See docs/adr/0011-deprecate-cnv-collection-vocabulary.md.

ACCEPTED_RESOURCE_TYPES: ClassVar[tuple[str, ...]] = ('fragment_score', 'cnv_collection')

The resource types this annotator’s resource_id may name.

An annotator that consumes a typed genomic resource states them here, and resolves its resource through resolve_resource(). Before gain#1329 the same fact was written once per annotator in whatever shape that annotator happened to use – a literal at a call site, a constant, or nothing at all with the check left to whichever constructor met the resource first – and the refusal a reader got for the wrong resource type differed accordingly.

This is the ANNOTATOR’s copy, not the only one: the wildcard expansion in annotation_config keys the same fact on annotator NAME rather than class (gain#1266), and the web editor states it again per configuration field. What is gone is the five different shapes it took inside the annotators. The wildcard map stays a separate statement on purpose – whether a name expands a wildcard is the annotation layer’s policy, not a property of the annotator (docs/adr/0029-wildcard-expandability-is-parser-policy.md, gain#1334) – and a test pins the two against each other.

The first element is the preferred spelling. A tuple rather than a set for that reason, as FRAGMENT_SCORE_TYPES is one: the order is rendered into the refusal, and AnnotationConfigParser.WILDCARD_RESOURCE_TYPES is pinned against element zero rather than against membership (why, in ADR 0029). So an annotator that comes to accept a further spelling APPENDS it.

Two annotators accept two spellings; each warns from the constructor that opens the resource, which still runs after this check passes the spelling through.

Empty means the annotator does not constrain its resource type – the default, because most annotators (effect_annotator, liftover_annotator, chrom_mapping, …) have no single typed resource to constrain. Those never call resolve_resource(), which refuses an empty declaration rather than rejecting every type in turn.

close() → None

Release what open() acquired and mark the annotator closed.

Safe on an annotator never opened, and safe twice; overrides keep it so and call the base. The pipeline calls it once per annotator and logs, rather than propagates, what it raises.

get_attribute_defaults(spec: AttributeSpec) → dict[str, Any]

Defaults for spec: an aggregator and parameters.

Empty by default. The constructor consults it for every attribute: the aggregator key becomes the aggregator when the configuration names none, and every other key becomes a parameter that the configuration’s own parameters override. Override it when defaults live somewhere other than the spec – a score resource declares its own, for instance.

get_attribute_specs() → dict[str, AttributeSpec]

Every attribute this annotator can produce, keyed by source.

The catalogue the configuration is checked against: a configured attribute whose source is not a key here is refused. Independent of the configuration and of open(). AnnotatorBase calls it from its constructor, so it may use only what the subclass set before delegating there.

open() → Annotator

Create work_dir and mark the annotator open; returns self.

Overrides that open resources call this and return self.

gain.annotation.fragment_score_annotator.LEGACY_ANNOTATOR_NAMES = {'cnv_collection': 'fragment_score', 'cnv_collection_annotator': 'fragment_score_annotator'}

The annotator names that mean this annotator, deprecated spelling to the preferred one it should be rewritten as.

Re-exported rather than declared here: annotation_config needs the same set while parsing and imports this module’s dependencies rather than this module, so it lives beside RETIRED_ANNOTATOR_NAMES in resource_types (gain#1266). Kept as a name here because this is where a reader of the annotator looks for it.

gain.annotation.fragment_score_annotator.LEGACY_FILTER_PARAMETER = 'cnv_filter'

Deprecated spelling, still honoured – pipelines we do not control write it. Stops being accepted in LEGACY_VOCABULARY_REMOVAL_RELEASE.

gain.annotation.fragment_score_annotator.build_fragment_score_annotator(pipeline: AnnotationPipeline, info: AnnotatorInfo) → Annotator

gain.annotation.gene_score_annotator module

Module containing the gene score annotator.

class gain.annotation.gene_score_annotator.GeneScoreAnnotator(pipeline: AnnotationPipeline | None, info: AnnotatorInfo, gene_score_resource: GenomicResource, input_gene_list: str)

Bases: AnnotatorBase

Gene score annotator class.

ACCEPTED_RESOURCE_TYPES: ClassVar[tuple[str, ...]] = ('gene_score',)

The resource types this annotator’s resource_id may name.

An annotator that consumes a typed genomic resource states them here, and resolves its resource through resolve_resource(). Before gain#1329 the same fact was written once per annotator in whatever shape that annotator happened to use – a literal at a call site, a constant, or nothing at all with the check left to whichever constructor met the resource first – and the refusal a reader got for the wrong resource type differed accordingly.

This is the ANNOTATOR’s copy, not the only one: the wildcard expansion in annotation_config keys the same fact on annotator NAME rather than class (gain#1266), and the web editor states it again per configuration field. What is gone is the five different shapes it took inside the annotators. The wildcard map stays a separate statement on purpose – whether a name expands a wildcard is the annotation layer’s policy, not a property of the annotator (docs/adr/0029-wildcard-expandability-is-parser-policy.md, gain#1334) – and a test pins the two against each other.

The first element is the preferred spelling. A tuple rather than a set for that reason, as FRAGMENT_SCORE_TYPES is one: the order is rendered into the refusal, and AnnotationConfigParser.WILDCARD_RESOURCE_TYPES is pinned against element zero rather than against membership (why, in ADR 0029). So an annotator that comes to accept a further spelling APPENDS it.

Two annotators accept two spellings; each warns from the constructor that opens the resource, which still runs after this check passes the spelling through.

Empty means the annotator does not constrain its resource type – the default, because most annotators (effect_annotator, liftover_annotator, chrom_mapping, …) have no single typed resource to constrain. Those never call resolve_resource(), which refuses an empty declaration rather than rejecting every type in turn.

get_attribute_specs() → dict[str, AttributeSpec]

Every attribute this annotator can produce, keyed by source.

The catalogue the configuration is checked against: a configured attribute whose source is not a key here is refused. Independent of the configuration and of open(). AnnotatorBase calls it from its constructor, so it may use only what the subclass set before delegating there.

property used_context_attributes: tuple[str, ...]

Names of upstream attributes this annotator reads from context.

Empty by default. An annotator that reads an attribute another annotator produced – a gene list, say – names it here: the pipeline builds its dependency graph from this tuple, and a reannotation reruns this annotator when a named attribute’s producer changes. Every name must be an attribute of an earlier annotator in the same pipeline.

gain.annotation.gene_score_annotator.build_gene_score_annotator(pipeline: AnnotationPipeline, info: AnnotatorInfo) → Annotator

Create a gene score annotator.

gain.annotation.gene_set_annotator module

class gain.annotation.gene_set_annotator.GeneSetAnnotator(pipeline: AnnotationPipeline | None, info: AnnotatorInfo, gene_set_resource: GenomicResource, input_gene_list: str)

Bases: AnnotatorBase

Gene set annotator class.

ACCEPTED_RESOURCE_TYPES: ClassVar[tuple[str, ...]] = ('gene_set_collection', 'gene_set')

Shared with the collection that opens the resource, rather than spelled again here: a third spelling added there would otherwise be refused by this annotator before the collection could accept it – the “stated in N places” fault gain#1329 is about.

DEFAULT_AGGREGATOR_TYPE = 'list'

get_attribute_defaults(spec: AttributeSpec) → dict[str, Any]

Defaults for spec: an aggregator and parameters.

Empty by default. The constructor consults it for every attribute: the aggregator key becomes the aggregator when the configuration names none, and every other key becomes a parameter that the configuration’s own parameters override. Override it when defaults live somewhere other than the spec – a score resource declares its own, for instance.

get_attribute_specs() → dict[str, AttributeSpec]

Every attribute this annotator can produce, keyed by source.

The catalogue the configuration is checked against: a configured attribute whose source is not a key here is refused. Independent of the configuration and of open(). AnnotatorBase calls it from its constructor, so it may use only what the subclass set before delegating there.

open() → Annotator

Create work_dir and mark the annotator open; returns self.

Overrides that open resources call this and return self.

property used_context_attributes: tuple[str, ...]

Names of upstream attributes this annotator reads from context.

Empty by default. An annotator that reads an attribute another annotator produced – a gene list, say – names it here: the pipeline builds its dependency graph from this tuple, and a reannotation reruns this annotator when a named attribute’s producer changes. Every name must be an attribute of an earlier annotator in the same pipeline.

gain.annotation.gene_set_annotator.build_gene_set_annotator(pipeline: AnnotationPipeline, info: AnnotatorInfo) → Annotator

Create a gene set annotator.

gain.annotation.genomic_score_annotator_base module

The base every genomic score annotator extends.

GenomicScoreAnnotatorBase binds an annotator to one GenomicScore and answers the questions the pipeline asks of any annotator – attribute specs, defaults, help text – from the score’s own definitions. The kinds, position_score_annotator and allele_score_annotator, live one per module beside this one and add the read.

Both kinds resolve their resource_id through resolve_resource(), which reads the types they accept off ACCEPTED_RESOURCE_TYPES.

class gain.annotation.genomic_score_annotator_base.GenomicScoreAnnotatorBase(pipeline: AnnotationPipeline, info: AnnotatorInfo, score: GenomicScore)

Bases: AnnotatorBase

Genomic score base annotator.

add_score_aggregator_documentation(attr: Attribute, aggregator: str, attribute_conf_agg: AggregatorDefinition | str | dict[str, Any] | None) → None

Collect score aggregator documentation.

build_attribute_help(attr: Attribute) → str

Build attribute help.

abstractmethod build_score_aggregator_documentation(attr: Attribute) → list[str]

Construct score aggregator documentation.

close() → None

Release what open() acquired and mark the annotator closed.

Safe on an annotator never opened, and safe twice; overrides keep it so and call the base. The pipeline calls it once per annotator and logs, rather than propagates, what it raises.

get_attribute_defaults(spec: AttributeSpec) → dict[str, Any]

Defaults for spec: an aggregator and parameters.

Empty by default. The constructor consults it for every attribute: the aggregator key becomes the aggregator when the configuration names none, and every other key becomes a parameter that the configuration’s own parameters override. Override it when defaults live somewhere other than the spec – a score resource declares its own, for instance.

get_attribute_specs() → dict[str, AttributeSpec]

Every attribute this annotator can produce, keyed by source.

The catalogue the configuration is checked against: a configured attribute whose source is not a key here is refused. Independent of the configuration and of open(). AnnotatorBase calls it from its constructor, so it may use only what the subclass set before delegating there.

is_open() → bool

Whether open() has run and close() has not since.

open() → Annotator

Create work_dir and mark the annotator open; returns self.

Overrides that open resources call this and return self.

simple_score_queries: list[str]

gain.annotation.liftover_annotator module

Provides a lift over annotator and helpers.

class gain.annotation.liftover_annotator.AbstractLiftoverAnnotator(pipeline: AnnotationPipeline, info: AnnotatorInfo, chain: LiftoverChain, source_genome: ReferenceGenome, target_genome: ReferenceGenome)

Bases: AnnotatorBase

Liftovver annotator class.

close() → None

Release what open() acquired and mark the annotator closed.

Safe on an annotator never opened, and safe twice; overrides keep it so and call the base. The pipeline calls it once per annotator and logs, rather than propagates, what it raises.

get_attribute_specs() → dict[str, AttributeSpec]

Every attribute this annotator can produce, keyed by source.

The catalogue the configuration is checked against: a configured attribute whose source is not a key here is refused. Independent of the configuration and of open(). AnnotatorBase calls it from its constructor, so it may use only what the subclass set before delegating there.

liftover_allele(allele: VCFAllele) → VCFAllele | None

Liftover an allele.

liftover_cnv(cnv_allele: Annotatable) → Annotatable | None

Liftover CNV allele annotatable.

liftover_position(position: Annotatable) → Annotatable | None

Liftover position annotatable.

liftover_region(region: Annotatable) → Annotatable | None

Liftover region annotatable.

open() → Annotator

Create work_dir and mark the annotator open; returns self.

Overrides that open resources call this and return self.

class gain.annotation.liftover_annotator.BasicLiftoverAnnotator(pipeline: AnnotationPipeline, info: AnnotatorInfo, chain: LiftoverChain, source_genome: ReferenceGenome, target_genome: ReferenceGenome)

Bases: AbstractLiftoverAnnotator

Basic liftover annotator class.

class gain.annotation.liftover_annotator.BcfLiftoverAnnotator(pipeline: AnnotationPipeline, info: AnnotatorInfo, chain: LiftoverChain, source_genome: ReferenceGenome, target_genome: ReferenceGenome)

Bases: AbstractLiftoverAnnotator

BCF tools liftover re-implementation annotator class.

class gain.annotation.liftover_annotator.LiftoverFunction(*args, **kwargs)

Bases: Protocol

Protocol for liftover function.

gain.annotation.liftover_annotator.basic_liftover_allele(chrom: str, pos: int, ref: str, alt: str, liftover_chain: LiftoverChain, *, source_genome: ReferenceGenome, target_genome: ReferenceGenome) → tuple[str, int, str, str] | None

Basic liftover an allele.

gain.annotation.liftover_annotator.basic_liftover_variant(chrom: str, pos: int, ref: str, alts: list[str], liftover_chain: LiftoverChain, *, source_genome: ReferenceGenome, target_genome: ReferenceGenome) → tuple[str, int, str, list[str]] | None

Basic liftover variant utility function.

gain.annotation.liftover_annotator.bcf_liftover_allele(chrom: str, pos: int, ref: str, alt: str, liftover_chain: LiftoverChain, *, source_genome: ReferenceGenome, target_genome: ReferenceGenome) → tuple[str, int, str, str] | None

Liftover a variant.

gain.annotation.liftover_annotator.bcf_liftover_variant(chrom: str, pos: int, ref: str, alts: list[str], liftover_chain: LiftoverChain, *, source_genome: ReferenceGenome, target_genome: ReferenceGenome) → tuple[str, int, str, list[str]] | None

BCF liftover variant utility function.

gain.annotation.liftover_annotator.build_liftover_annotator(pipeline: AnnotationPipeline, info: AnnotatorInfo) → Annotator

Create a liftover annotator.

gain.annotation.normalize_allele_annotator module

Provides normalize allele annotator and helpers.

class gain.annotation.normalize_allele_annotator.NormalizeAlleleAnnotator(pipeline: AnnotationPipeline, info: AnnotatorInfo)

Bases: AnnotatorBase

Annotator to normalize VCF alleles.

close() → None

Release what open() acquired and mark the annotator closed.

Safe on an annotator never opened, and safe twice; overrides keep it so and call the base. The pipeline calls it once per annotator and logs, rather than propagates, what it raises.

get_attribute_specs() → dict[str, AttributeSpec]

Every attribute this annotator can produce, keyed by source.

The catalogue the configuration is checked against: a configured attribute whose source is not a key here is refused. Independent of the configuration and of open(). AnnotatorBase calls it from its constructor, so it may use only what the subclass set before delegating there.

open() → Annotator

Create work_dir and mark the annotator open; returns self.

Overrides that open resources call this and return self.

gain.annotation.normalize_allele_annotator.build_normalize_allele_annotator(pipeline: AnnotationPipeline, info: AnnotatorInfo) → Annotator

gain.annotation.normalize_allele_annotator.normalize_allele(allele: VCFAllele, genome: ReferenceGenome) → VCFAllele

Normalize an allele.

Using algorithm defined in following https://genome.sph.umich.edu/wiki/Variant_Normalization

gain.annotation.pipeline_doc module

The one renderer of the pipeline documentation page.

Three callers render annotate_doc_pipeline_template.jinja: the annotate_doc CLI, the web API’s download endpoint, and the annotation_pipeline resource implementation. Each used to bind the template and build its own res_url/hist_url pair, and the copies drifted – d8624b787 moved the CLI’s addresses onto the GRR’s public mirror and left the endpoint’s on the repository’s own url, where they stayed wrong for two months (#841, #952).

The addresses are a policy, not a constant, so they are injected rather than hardcoded. Two callers want the public-mirror policy and get it by default; the resource implementation publishes its pages from inside the GRR tree and passes repository-relative addresses instead.

That policy is one object, not one callable per address (#970). #952 injected the pair as two independent arguments, which left a public resource address beside a relative histogram address representable – the same drift as #841, merely moved up a level and into a single call. Both policies live here, beside the renderer that chooses between them, and the repository-relative one needs nothing but the page’s own resource.

gain.annotation.pipeline_doc.PUBLIC_MIRROR_ADDRESSES = <gain.annotation.pipeline_doc.PublicMirrorAddresses object>

The policy every caller gets unless it says otherwise. Stateless, so one shared instance rather than one per render.

class gain.annotation.pipeline_doc.PipelineDocAddresses(*args, **kwargs)

Bases: Protocol

Where a rendered page points, for every kind of thing it points at.

One object rather than a callable per address: the two are a single policy, and a page that mixes them – a resource named on the public mirror beside a histogram named relatively – is incoherent. Passed separately that pairing was merely unlikely; passed together it is unrepresentable (#970).

histogram_url(score: GenomicScore, score_id: str) → str | None

Address of the score’s histogram image for score_id.

None when the score has no histogram to show.

resource_url(resource: GenomicResource) → str

Address of the resource’s documentation page.

class gain.annotation.pipeline_doc.PublicMirrorAddresses

Bases: object

Address everything on the GRR’s public mirror.

What a reader who is not browsing the GRR tree needs: the page may be downloaded, or served from somewhere else entirely, so nothing on it may be relative to where it happens to sit.

histogram_url(score: GenomicScore, score_id: str) → str | None

resource_url(resource: GenomicResource) → str

class gain.annotation.pipeline_doc.RepositoryRelativeAddresses(resource: GenomicResource)

Bases: object

Address managed targets relative to the repository root.

The policy the annotation_pipeline resource implementation wants: its pages are published inside the GRR tree, under the pipeline’s own resource id, so a reader browsing that tree resolves the links without ever leaving it. Anything outside the managed GRR cannot be reached that way and falls back to the mirror, with a warning.

Built from the pipeline’s own resource, and needs nothing else – no implementation, no repository handle.

histogram_url(score: GenomicScore, score_id: str) → str | None

Address the score’s histogram image, if it has one.

Probed differently from resource_url(), and deliberately: what has to be under the managed GRR is the image, not the score’s own page.

resource_url(resource: GenomicResource) → str

gain.annotation.pipeline_doc.render_pipeline_doc(pipeline: AnnotationPipeline, *, pipeline_path: str | None = None, addresses: PipelineDocAddresses = <gain.annotation.pipeline_doc.PublicMirrorAddresses object>) → str

Render the documentation page for pipeline.

pipeline_path is shown on the page when given; the callers that have no file to name leave it None, which renders the same page as omitting it entirely.

gain.annotation.position_score_annotator module

The position_score_annotator.

Annotates with scores keyed by genomic position – phastCons, phyloP, FitCons2 and the like – read from a position_score resource.

class gain.annotation.position_score_annotator.PositionScoreAnnotator(pipeline: AnnotationPipeline, info: AnnotatorInfo)

Bases: GenomicScoreAnnotatorBase

This class implements the position_score_annotator.

The position_score_annotator requires the resource_id parameter, whose value must be an id of a genomic resource of type position_score.

The position_score resource provides a set of scores (see …) that the position_score_annotator uses as attributes to assign to the annotatable.

The position_score_annotator recognizes two attribute level parameters, both of which apply to annotatables that refer to a region of the reference genome:

• aggregator controls how the position scores are aggregated. The deprecated name position_aggregator is still accepted.

• none_value_replacement stands in for every null of the region’s per-position expansion – a position no record covers, and a covered position whose value is NA – before the aggregator sees it. Unset, nulls stay inert and every aggregator skips them, so a region’s mean is the mean over its covered positions alone.

Neither applies to an annotatable that never reaches the region fold: a substitution, which reads a single position; one on a chromosome the resource does not carry; or one longer than region_length_cutoff, which is declined before it is read.

ACCEPTED_RESOURCE_TYPES: ClassVar[tuple[str, ...]] = ('position_score',)

The resource types this annotator’s resource_id may name.

An annotator that consumes a typed genomic resource states them here, and resolves its resource through resolve_resource(). Before gain#1329 the same fact was written once per annotator in whatever shape that annotator happened to use – a literal at a call site, a constant, or nothing at all with the check left to whichever constructor met the resource first – and the refusal a reader got for the wrong resource type differed accordingly.

This is the ANNOTATOR’s copy, not the only one: the wildcard expansion in annotation_config keys the same fact on annotator NAME rather than class (gain#1266), and the web editor states it again per configuration field. What is gone is the five different shapes it took inside the annotators. The wildcard map stays a separate statement on purpose – whether a name expands a wildcard is the annotation layer’s policy, not a property of the annotator (docs/adr/0029-wildcard-expandability-is-parser-policy.md, gain#1334) – and a test pins the two against each other.

The first element is the preferred spelling. A tuple rather than a set for that reason, as FRAGMENT_SCORE_TYPES is one: the order is rendered into the refusal, and AnnotationConfigParser.WILDCARD_RESOURCE_TYPES is pinned against element zero rather than against membership (why, in ADR 0029). So an annotator that comes to accept a further spelling APPENDS it.

Two annotators accept two spellings; each warns from the constructor that opens the resource, which still runs after this check passes the spelling through.

Empty means the annotator does not constrain its resource type – the default, because most annotators (effect_annotator, liftover_annotator, chrom_mapping, …) have no single typed resource to constrain. Those never call resolve_resource(), which refuses an empty declaration rather than rejecting every type in turn.

build_score_aggregator_documentation(attr: Attribute) → list[str]

Collect score aggregator documentation.

get_attribute_defaults(spec: AttributeSpec) → dict[str, Any]

Defaults for spec: an aggregator and parameters.

Empty by default. The constructor consults it for every attribute: the aggregator key becomes the aggregator when the configuration names none, and every other key becomes a parameter that the configuration’s own parameters override. Override it when defaults live somewhere other than the spec – a score resource declares its own, for instance.

gain.annotation.position_score_annotator.build_position_score_annotator(pipeline: AnnotationPipeline, info: AnnotatorInfo) → Annotator

gain.annotation.prepare_tabular module

Prepare a tabular file for parallel annotation.

Sorts a (possibly gzip-compressed) columnar file by genomic coordinates and produces a bgzip-compressed, tabix-indexed output that annotate_tabular can fan out across regions.

The same --col-* options as annotate_tabular select which input columns carry chromosome / position / etc., and the same RecordToAnnotable lookup is reused to derive the sort and tabix keys.

gain.annotation.prepare_tabular.cli(argv: list[str] | None = None) → None

Entry point for the prepare_tabular tool.

gain.annotation.processing_pipeline module

class gain.annotation.processing_pipeline.Annotation(annotatable: Annotatable | None, context: dict[str, ~typing.Any]=<factory>)

Bases: object

A pair of an annotatable and its relevant context.

The context can hold any key/value pair relevant to the annotatable and is typically used to store the results of annotators.

annotatable: Annotatable | None

context: dict[str, Any]

class gain.annotation.processing_pipeline.AnnotationPipelineAnnotatablesBatchFilter(annotation_pipeline: AnnotationPipeline)

Bases: AnnotationsWithSourceBatchFilter, AnnotationPipelineContextManager

Filter that annotates an AnnotationWithSource batch using a pipeline.

class gain.annotation.processing_pipeline.AnnotationPipelineAnnotatablesFilter(annotation_pipeline: AnnotationPipeline)

Bases: AnnotationsWithSourceFilter, AnnotationPipelineContextManager

Filter that annotates an AnnotationWithSource object using a pipeline.

class gain.annotation.processing_pipeline.AnnotationPipelineContextManager(annotation_pipeline: AnnotationPipeline)

Bases: AbstractContextManager

A context manager for annotation pipelines.

class gain.annotation.processing_pipeline.AnnotationsWithSource(source: Any, annotations: list[Annotation])

Bases: object

A pair of a list of Annotation instances and their source.

The source is typically a variant read from some format, with the ‘annotations’ attribute corresponding to its alleles.

annotations: list[Annotation]

source: Any

class gain.annotation.processing_pipeline.AnnotationsWithSourceBatchFilter

Bases: Filter

Base class for filters that work on AnnotationsWithSource batches.

filter(data: Sequence[AnnotationsWithSource]) → Sequence[AnnotationsWithSource]

Filter a batch of AnnotationsWithSource objects.

class gain.annotation.processing_pipeline.AnnotationsWithSourceFilter

Bases: Filter

Base class for filters that work on AnnotationsWithSource objects.

filter(data: AnnotationsWithSource) → AnnotationsWithSource

Filter a single AnnotationsWithSource object.

class gain.annotation.processing_pipeline.DeleteAttributesFromAWSBatchFilter(attributes_to_remove: Sequence[str])

Bases: Filter

Filter to remove items from AWS batches. Works in-place.

filter(data: Sequence[AnnotationsWithSource]) → Sequence[AnnotationsWithSource]

class gain.annotation.processing_pipeline.DeleteAttributesFromAWSFilter(attributes_to_remove: Sequence[str])

Bases: Filter

Filter to remove items from AWSs. Works in-place.

filter(data: AnnotationsWithSource) → AnnotationsWithSource

gain.annotation.record_to_annotatable module

class gain.annotation.record_to_annotatable.CSHLAlleleRecordToAnnotatable(columns: tuple, ref_genome: ReferenceGenome | None)

Bases: RecordToAnnotable

Transform a CSHL variant record into a VCF allele annotatable.

build(record: dict[str, str]) → Annotatable

Constructs an annotatable from a record.

class gain.annotation.record_to_annotatable.DaeAlleleRecordToAnnotatable(columns: tuple, ref_genome: ReferenceGenome | None)

Bases: RecordToAnnotable

Transform a CSHL variant record into a VCF allele annotatable.

build(record: dict[str, str]) → Annotatable

Constructs an annotatable from a record.

class gain.annotation.record_to_annotatable.RecordToAnnotable(columns: tuple, ref_genome: ReferenceGenome | None)

Bases: ABC

Base class for record to annotable transformation.

abstractmethod build(record: dict[str, str]) → Annotatable

Constructs an annotatable from a record.

class gain.annotation.record_to_annotatable.RecordToCNVAllele(columns: tuple, ref_genome: ReferenceGenome | None)

Bases: RecordToAnnotable

Transform a columns record into a CNV allele annotatable.

build(record: dict[str, str]) → Annotatable

Constructs an annotatable from a record.

class gain.annotation.record_to_annotatable.RecordToPosition(columns: tuple, ref_genome: ReferenceGenome | None)

Bases: RecordToAnnotable

build(record: dict[str, str]) → Annotatable

Constructs an annotatable from a record.

class gain.annotation.record_to_annotatable.RecordToRegion(columns: tuple, ref_genome: ReferenceGenome | None)

Bases: RecordToAnnotable

build(record: dict[str, str]) → Annotatable

Constructs an annotatable from a record.

class gain.annotation.record_to_annotatable.RecordToVcfAllele(columns: tuple, ref_genome: ReferenceGenome | None)

Bases: RecordToAnnotable

build(record: dict[str, str]) → Annotatable

Constructs an annotatable from a record.

class gain.annotation.record_to_annotatable.VcfLikeRecordToVcfAllele(columns: tuple, ref_genome: ReferenceGenome | None)

Bases: RecordToAnnotable

Transform a columns record into VCF allele annotatable.

build(record: dict[str, str]) → Annotatable

Constructs an annotatable from a record.

gain.annotation.record_to_annotatable.add_record_to_annotable_arguments(parser: ArgumentParser) → None

gain.annotation.record_to_annotatable.build_annotatable_from_dict(obj: dict[str, str], ref_genome: ReferenceGenome | None = None) → Annotatable

Build an annotatable from a dictionary of string values.

gain.annotation.record_to_annotatable.build_record_to_annotatable(renamed_columns: dict[str, str], available_columns: set[str], ref_genome: ReferenceGenome | None = None) → RecordToAnnotable

Transform a variant record into an annotatable.

Parameters:

• renamed_columns (dict[str, str]) –

  Mapping from expected internal column identifiers (e.g. “col_<field>”) to the actual column names present in the input source. A column can be excluded from usage if an identifier is mapped to “-“. Example rename:

  "col_<field>": "<input source column name for the field>"
  

  Example exclude:

  "col_<field>": "-"
  

• available_columns (set[str]) – The set of column names available in the input records.

• ref_genome (ReferenceGenome | None, optional) – Optional reference genome context used for creating annotatables. Not all annotatables require it.

gain.annotation.simple_effect_annotator module

class gain.annotation.simple_effect_annotator.SimpleEffect(effect_type: str, transcript_id: str, gene: str)

Bases: object

effect_type: str

gene: str

transcript_id: str

class gain.annotation.simple_effect_annotator.SimpleEffectAnnotator(pipeline: AnnotationPipeline, info: AnnotatorInfo)

Bases: AnnotatorBase

Simple effect annotator class.

call_region(chrom: str, beg: int, end: int, tx: TranscriptModel, *, func_name: str, classification: str) → SimpleEffect | None

Call a region with a specific classification.

cds_intron_regions(transcript: TranscriptModel) → list[Region]

Return whether region is CDS intron.

cds_regions(transcript: TranscriptModel) → Sequence[Region]

Return whether the region is classified as coding.

static effect_types() → list[str]

get_attribute_defaults(spec: AttributeSpec) → dict[str, Any]

Defaults for spec: an aggregator and parameters.

Empty by default. The constructor consults it for every attribute: the aggregator key becomes the aggregator when the configuration names none, and every other key becomes a parameter that the configuration’s own parameters override. Override it when defaults live somewhere other than the spec – a score resource declares its own, for instance.

get_attribute_specs() → dict[str, AttributeSpec]

Every attribute this annotator can produce, keyed by source.

The catalogue the configuration is checked against: a configured attribute whose source is not a key here is refused. Independent of the configuration and of open(). AnnotatorBase calls it from its constructor, so it may use only what the subclass set before delegating there.

noncoding_regions(transcript: TranscriptModel) → list[Region]

Return whether the region is noncoding.

open() → Annotator

Create work_dir and mark the annotator open; returns self.

Overrides that open resources call this and return self.

peripheral_regions(transcript: TranscriptModel) → list[Region]

Return whether the region is peripheral.

run_annotate(chrom: str, beg: int, end: int) → dict[str, set[SimpleEffect]]

Return classification with a set of affected genes.

gain.annotation.simple_effect_annotator.build_simple_effect_annotator(pipeline: AnnotationPipeline, info: AnnotatorInfo) → Annotator

gain.annotation.utils module

gain.annotation.utils.find_annotator_gene_models(info: AnnotatorInfo, grr: GenomicResourceRepo) → GeneModels

Get gene models from the annotator info or genomic context.

gain.annotation.utils.find_annotator_reference_genome(info: AnnotatorInfo, gene_models: GeneModels, pipeline: AnnotationPipeline, grr: GenomicResourceRepo) → ReferenceGenome

Get reference genome from the annotator info or genomic context.

gain.annotation.utils.preamble_reference_genome_id(pipeline: AnnotationPipeline) → str | None

The genome id the pipeline’s preamble declares, if any.

None when there is no preamble, or when it declares no genome (see AnnotationPreamble.input_reference_genome).

gain.annotation.utils.resolve_reference_genome(info: AnnotatorInfo, genome_resource_id: str | None, grr: GenomicResourceRepo, *, searched: str) → ReferenceGenome

Build the genome genome_resource_id names, else use the context.

The caller resolves its own precedence chain – which operands it has differs per annotator – and passes the winning id here. Everything downstream of that chain is the same for every annotator and lives only in this function, so a fix to it cannot miss a call site the way gain#1055 had to be fixed at three of them.

searched names the sources the caller consulted, for the error raised when nothing resolves; it is the one part of that error that cannot be stated here, since the chain is the caller’s.

gain.annotation.value_transform_eval module

Restricted evaluator for pipeline value_transform expressions.

A value_transform is an expression supplied through pipeline configuration, and that configuration can arrive verbatim in an anonymous HTTP request body. It must therefore never reach an unrestricted eval.

compile_value_transform validates the expression against a small whitelist of AST nodes, operators, names and calls, rejects everything else, and returns a callable that evaluates the expression over a single bound name, value, with no access to builtins. Static bounds on numeric and string literals cap the cheapest resource-exhaustion inputs, and a size-propagation pass (_check_result_size) rejects operator-driven blow-ups whose individual literals stay under those bounds – chained sequence repetition ('ab' * 99 * 99 * 99 * 99) and %-format width ('%99999999d' % value) (gain#767).

gain.annotation.value_transform_eval.compile_value_transform(expr: str) → Callable[[Any], Any]

Validate expr and return a callable applying it to value.

Raise ValueError if expr is not valid Python or uses any construct outside the restricted whitelist.

Module contents