gain.annotation package

Submodules

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.

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

property chromosome: str

property end_position: int

static from_string(value: str) → Annotatable

Deserialize an Annotatable instance from a string value.

property pos: int

property pos_end: int

property position: int

abstractmethod to_dict() → dict

Serialize the annotatable to a dictionary.

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

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

Defines small variants annotatable.

property alt: str

property alternative: str

static from_string(value: str) → VCFAllele

Deserialize an Annotatable instance from a string value.

property ref: str

property reference: str

to_dict() → dict

Serialize the annotatable to a dictionary.

gain.annotation.annotate_columns module

gain.annotation.annotate_columns.annotate_columns(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 columns file using a processing pipeline.

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

Entry point for running the columns annotation tool.

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_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_cli_genomic_context(cli_args: dict[str, Any]) → GenomicContext

Helper method to collect necessary objects from the genomic context.

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.

gain.annotation.annotate_utils.cache_pipeline_resources(grr: GenomicResourceRepo, pipeline: AnnotationPipeline) → None

Cache resources that the given pipeline will use.

gain.annotation.annotate_utils.get_grr_from_context(context: GenomicContext) → GenomicResourceRepo

Get the genomic resource repository from the genomic context.

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.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_utils.stringify(value: Any, *, vcf: bool = False) → str

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

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.

ANNOTATION_CONFIG_GRAMMAR = '\n?start: resource_id [filter]\n\n?resource_id: (resource_name | wildcard)\n\nwildcard: /[\\w\\d\\/_*]+/\n\nfilter: "[" (equals | and_)+ "]"\n\nand_: operation "and" operation\n\nequals: (name"=\\""value"\\"") | (name"=\'"value"\'")\n\nin: ("\\""value"\\"" " in " name) | ("\'"value"\'" " in " name)\n\nresource_name: /[\\w\\d\\/_\\-!@#$%^<>+]+/\n\n?name: /[\\w\\d\\/_\\-!@#$%^<>+*]+/\n\n?value: /[\\w\\d\\/ _\\-!@#$%^<>+*]+/\n\n?operation: equals | in | and_\n\n%ignore " "\n'

WILDCARD_LIMIT = 500

static build_labels_query(node: Any, labels_query: dict[str, Any] | None = None) → dict[str, Callable[[str], bool]]

Build labels query from parsed tree node.

static has_wildcard(string: str) → bool

Ascertain whether a string contains a valid wildcard.

static match_labels_query(query: dict[str, Callable[[str], bool]], resource_labels: dict[str, str]) → bool

Check if the labels query for a wildcard matches.

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]) → AttributeInfo

Parse annotation attribute raw configuration.

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

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_id: str, grr: GenomicResourceRepo) → list[str]

Collect resources matching a given query.

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', input_reference_genome_res: 'GenomicResource | None', metadata: 'dict[str, Any]')

Bases: object

description: str

input_reference_genome: str

input_reference_genome_res: GenomicResource | None

metadata: dict[str, Any]

summary: str

class gain.annotation.annotation_config.AnnotatorInfo(_type: str, attributes: list[AttributeInfo], 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[AttributeInfo]

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.AttributeInfo(name: str, source: str, *, internal: bool | None, parameters: ParamsUsageMonitor | dict[str, Any], _type: str = 'str', description: str = '', documentation: str | None = None, attribute_type: str = 'attribute', default: bool = True)

Bases: object

Defines annotation attribute configuration.

attribute_type: str = 'attribute'

static create(source: str, name: str | None = None, *, internal: bool = False) → AttributeInfo

Create an AttributeInfo instance.

default: bool = True

description: str = ''

property documentation: str

internal: bool | None

name: str

parameters: ParamsUsageMonitor

source: str

value_type: str = '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])

Bases: Mapping

Class to monitor usage of annotator parameters.

get_unused_keys() → set[str]

get_used_keys() → set[str]

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_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_grr(grr: GenomicResourceRepo, resource: GenomicResource) → AnnotationPipeline

Load a pipeline from a grr and a resource.

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

GenomicContext | None

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

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

AnnotationPipeline | None

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

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

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

Apply all annotators to an annotatable.

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: AttributeInfo) → Annotator | None

get_attribute_info(attribute_name: str) → AttributeInfo | None

get_attributes() → list[AttributeInfo]

get_attributes_by_type(attribute_type: str) → list[AttributeInfo]

get_info() → list[AnnotatorInfo]

get_resource_ids() → set[str]

open() → AnnotationPipeline

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

print() → None

Print the annotation pipeline.

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

Bases: ABC

Annotator provides a set of attrubutes for a given Annotatable.

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

Produce annotation attributes for an annotatable.

property attributes: list[AttributeInfo]

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

close() → None

abstractmethod get_all_attribute_descriptions() → dict[str, AttributeDesc]

Get descriptions of all attributes provided by the annotator.

get_info() → AnnotatorInfo

is_open() → bool

open() → Annotator

property resource_ids: set[str]

property resources: list[GenomicResource]

property used_context_attributes: tuple[str, ...]

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

Bases: Annotator

Defines annotator decorator base class.

close() → None

get_all_attribute_descriptions() → dict[str, AttributeDesc]

Get descriptions of all attributes provided by the annotator.

is_open() → bool

open() → Annotator

class gain.annotation.annotation_pipeline.AttributeDesc(source: str, type: str, description: str, name: str | None = None, default: bool = True, internal: bool = False, params: dict[str, ~typing.Any] = <factory>, attribute_type: str = 'attribute')

Bases: object

Holds default attribute configuration for annotators.

attribute_type: str = 'attribute'

default: bool = True

description: str

internal: bool = False

name: str | None = None

params: dict[str, Any]

source: str

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 annotation attributes for an annotatable.

static decorate(child: Annotator) → Annotator

property used_context_attributes: tuple[str, ...]

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

Bases: AnnotationPipeline

Provides functionality for reannotation.

get_attributes() → list[AttributeInfo]

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 annotation attributes for an annotatable.

static decorate(child: Annotator) → Annotator

Apply value transform decorator to an annotator.

gain.annotation.annotator_base module

Provides base class for annotators.

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

Bases: Annotator

Base implementation of the Annotator class.

BASE_DOC_URL = 'https://iossifovlab.com/gpfuserdocs/administration/annotation.html'

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

Produce annotation attributes for an annotatable.

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

open() → Annotator

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_all_attribute_descriptions() → dict[str, AttributeDesc]

Get descriptions of all attributes provided by the annotator.

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

gain.annotation.cnv_collection_annotator module

class gain.annotation.cnv_collection_annotator.CnvCollectionAnnotator(pipeline: AnnotationPipeline, info: AnnotatorInfo)

Bases: Annotator

Simple effect annotator class.

CNV_FILTER_GRAMMAR = '\n?start: filter | and_ | or\n\nand_: filter "and" filter\n\nor: filter "or" filter\n\n?filter: subject operator subject | or | and_\n\n?subject: variable | value\n\nvalue: "\\"" word "\\"" | number\n\nvariable: word\n\noperator: equals | greater_than | less_than | in\n\nequals: "=="\n\ngreater_than: ">"\n\nless_than: "<"\n\nin: "in"\n\nword: /[a-zA-Z!@#$%^&*()_+]+/\n\nnumber: /[0-9\\.]+/\n\n%ignore " "\n'

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

Produce annotation attributes for an annotatable.

close() → None

get_all_attribute_descriptions() → dict[str, AttributeDesc]

Get descriptions of all attributes provided by the annotator.

open() → Annotator

gain.annotation.cnv_collection_annotator.build_cnv_collection_annotator(pipeline: AnnotationPipeline, info: AnnotatorInfo) → Annotator

gain.annotation.debug_annotator module

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

Bases: Annotator

Defines example annotator.

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

Produce annotation attributes for an annotatable.

get_all_attribute_descriptions() → dict[str, AttributeDesc]

Get descriptions of all attributes provided by the annotator.

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

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.

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

Produce annotation attributes for an annotatable.

close() → None

get_all_attribute_descriptions() → dict[str, AttributeDesc]

Get descriptions of all attributes provided by the annotator.

open() → Annotator

gain.annotation.effect_annotator.build_effect_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: Annotator

Gene score annotator class.

DEFAULT_AGGREGATOR_TYPE = 'dict'

aggregate_gene_values(score_id: str, gene_symbols: list[str], aggregator_type: str) → Any

Aggregate gene score values.

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

Produce annotation attributes for an annotatable.

get_all_attribute_descriptions() → dict[str, AttributeDesc]

Get descriptions of all attributes provided by the annotator.

property used_context_attributes: tuple[str, ...]

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.

get_all_attribute_descriptions() → dict[str, AttributeDesc]

Get descriptions of all attributes provided by the annotator.

open() → Annotator

property used_context_attributes: tuple[str, ...]

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

Create a gene set annotator.

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

get_all_attribute_descriptions() → dict[str, AttributeDesc]

Get descriptions of all attributes provided by the annotator.

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

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

get_all_attribute_descriptions() → dict[str, AttributeDesc]

Get descriptions of all attributes provided by the annotator.

open() → Annotator

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

class gain.annotation.processing_pipeline.Annotation(annotatable: ~gain.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_columnsdict[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_columnsset[str]

The set of column names available in the input records.

ref_genomeReferenceGenome | None, optional

Optional reference genome context used for creating annotatables. Not all annotatables require it.

gain.annotation.score_annotator module

This contains the implementation of the three score annotators.

Genomic score annotators defined are positions_score, np_score, and allele_score.

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

Bases: GenomicScoreAnnotatorBase

This class implements allele_score annotator.

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

Produce annotation attributes for an annotatable.

build_score_aggregator_documentation(attr_info: AttributeInfo) → list[str]

Collect score aggregator documentation.

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

Bases: Annotator

Genomic score base annotator.

GENOMIC_SCORE_HELP = '\n\n<div class="score-description">\n\n## {{ data.name }}\n\n{{ data.description}}\n\n{{ data.resource_summary }}\n\n{{ data.histogram }}\n\nGenomic resource:\n<a href={{data.resource_url}} target="_blank">{{ data.resource_id }}</a>\n\n<details>\n\n<summary class="details">\n\n#### Details\n\n</summary>\n\n<div class="details-body">\n\n##### Attribute properties:\n\n* **source**: {{ data.source }}\n{% for aggregator in data.aggregators %}\n\n* {{ aggregator }}\n\n{% endfor %}\n\n\n##### Resource properties:\n\n* **resource_type**: `{{ data.resource_type }}`\n\n\n##### Annotator documentation:\n\n* **annotator_type**: `{{ data.annotator_type }}`\n\n{{ data.annotator_doc }}\n\n</div>\n\n</details>\n\n</div>\n\n'

SCORE_HISTOGRAM = '\n<div class="modal-histogram">\n\n<div class="histogram-image">\n\n\n\n</div>\n\n</div>\n'

add_score_aggregator_documentation(attribute_info: AttributeInfo, aggregator: str, attribute_conf_agg: str | None) → None

Collect score aggregator documentation.

build_attribute_help(attr_info: AttributeInfo) → str

Build attribute help.

abstractmethod build_score_aggregator_documentation(attr_info: AttributeInfo) → list[str]

Construct score aggregator documentation.

close() → None

get_all_attribute_descriptions() → dict[str, AttributeDesc]

Get descriptions of all attributes provided by the annotator.

is_open() → bool

open() → Annotator

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

Bases: GenomicScoreAnnotatorBase

This class implements the position_score annotator.

The position_score annotator requires the resrouce_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 recognized one attribute level parameter called position_aggregator that controls how the position scores are aggregator for annotates that ref to a region of the reference genome.

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

Produce annotation attributes for an annotatable.

build_score_aggregator_documentation(attr_info: AttributeInfo) → list[str]

Collect score aggregator documentation.

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

gain.annotation.score_annotator.build_np_score_annotator(pipeline: AnnotationPipeline, info: AnnotatorInfo) → Annotator

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

gain.annotation.score_annotator.get_genomic_resource(pipeline: AnnotationPipeline, info: AnnotatorInfo, resource_types: set[str]) → GenomicResource

Return genomic score resource used for given genomic score annotator.

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.

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

Produce annotation attributes for an annotatable.

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_all_attribute_descriptions() → dict[str, AttributeDesc]

Get descriptions of all attributes provided by the annotator.

noncoding_regions(transcript: TranscriptModel) → list[Region]

Return whether the region is noncoding.

open() → Annotator

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.

Module contents