dae.annotation package

Submodules

dae.annotation.annotatable module

class dae.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(value, names=<not given>, *values, module=None, qualname=None, type=None, start=1, boundary=None)

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

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

class dae.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.

class dae.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.

class dae.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.

class dae.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

dae.annotation.annotate_columns module

class dae.annotation.annotate_columns.AnnotateColumnsTool(raw_args: list[str] | None = None)

Bases: AnnotationTool

Annotation tool for TSV-style text files.

add_tasks_to_graph() → None

Add tasks to annotation tool task graph.

get_argument_parser() → ArgumentParser

Configure argument parser.

prepare_for_annotation() → None

Perform operations required for annotation.

dae.annotation.annotate_columns.cli(raw_args: list[str] | None = None) → None

dae.annotation.annotate_columns.combine(args: Any, pipeline_config: list[dict[str, Any]] | RawFullConfig, pipeline_config_old: str | None, grr_definition: dict | None, partfile_paths: list[str], out_file_path: str, ref_genome_id: str) → None

Combine annotated region parts into a single VCF file.

dae.annotation.annotate_doc module

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

Run command line interface for annotate_vcf tool.

dae.annotation.annotate_doc.configure_argument_parser() → ArgumentParser

Construct and configure argument parser.

dae.annotation.annotate_utils module

class dae.annotation.annotate_utils.AnnotationTool(raw_args: list[str] | None = None)

Bases: object

Base class for annotation tools. Format-agnostic.

add_tasks_to_graph() → None

Add tasks to annotation tool task graph.

static annotate(handler: AbstractFormat, batch_mode: bool) → None

Run annotation.

abstract get_argument_parser() → ArgumentParser

static get_task_dir(region: Region | None) → str

Get dir for batch annotation.

prepare_for_annotation() → None

Perform operations required for annotation.

static produce_annotation_pipeline(pipeline_config: list[dict[str, Any]], pipeline_config_old: str | None, grr_definition: dict | None, *, allow_repeated_attributes: bool, work_dir: Path | None = None, full_reannotation: bool = False) → AnnotationPipeline

Produce an annotation or reannotation pipeline.

run() → None

Construct annotation tasks and execute task graph.

setup_work_dir() → None

dae.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.

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

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

dae.annotation.annotate_utils.produce_tabix_index(filepath: str, args: Any = None) → None

Produce a tabix index file for the given variants file.

dae.annotation.annotate_vcf module

class dae.annotation.annotate_vcf.AnnotateVCFTool(raw_args: list[str] | None = None)

Bases: AnnotationTool

Annotation tool for the VCF file format.

add_tasks_to_graph() → None

Add tasks to annotation tool task graph.

get_argument_parser() → ArgumentParser

Construct and configure argument parser.

prepare_for_annotation() → None

Perform operations required for annotation.

dae.annotation.annotate_vcf.cli(raw_args: list[str] | None = None) → None

dae.annotation.annotate_vcf.combine(args: Namespace, input_file_path: str, pipeline_config: list[dict[str, Any]], pipeline_config_old: str | None, grr_definition: dict | None, partfile_paths: list[str], output_file_path: str) → None

Combine annotated region parts into a single VCF file.

dae.annotation.annotation_config module

class dae.annotation.annotation_config.AnnotationConfigParser

Bases: object

Parser for annotation configuration.

static has_wildcard(string: str) → bool

Ascertain whether a string contains a valid wildcard.

static match_labels_query(query: dict[str, str], 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, wildcard: str, grr: GenomicResourceRepo) → list[str]

Collect resources matching a given query.

exception dae.annotation.annotation_config.AnnotationConfigurationError

Bases: ValueError

class dae.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 dae.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]

type: str

class dae.annotation.annotation_config.AttributeInfo(name: str, source: str, *, internal: bool, parameters: ParamsUsageMonitor | dict[str, Any], _type: str = 'str', description: str = '', documentation: str | None = None)

Bases: object

Defines annotation attribute configuration.

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

Create an AttributeInfo instance.

description: str = ''

property documentation: str

internal: bool

name: str

parameters: ParamsUsageMonitor

source: str

type: str = 'str'

class dae.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 dae.annotation.annotation_config.RawFullConfig

Bases: TypedDict

annotators: list[dict[str, Any]]

preamble: RawPreamble

class dae.annotation.annotation_config.RawPreamble

Bases: TypedDict

description: str

input_reference_genome: str

metadata: dict[str, Any]

summary: str

dae.annotation.annotation_factory module

Factory for creation of annotation pipeline.

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

Build an annotation pipeline.

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

Check for repeated attributes in annotator configuration.

dae.annotation.annotation_factory.check_for_repeated_attributes_in_pipeline(pipeline: AnnotationPipeline, *, allow_repeated_attributes: bool = False) → None

Check for repeated attributes in pipeline configuration.

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

Check annotator configuration for unused parameters.

dae.annotation.annotation_factory.copy_annotation_pipeline(pipeline: AnnotationPipeline) → AnnotationPipeline

Copy an annotation pipeline instance.

dae.annotation.annotation_factory.copy_reannotation_pipeline(pipeline: ReannotationPipeline) → ReannotationPipeline

Copy a reannotation pipeline instance.

dae.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.

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

Return the list of all registered annotator factory types.

dae.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.

dae.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.

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

Register additional annotator factory.

By default all genotype storage factories should be registered at [dae.genotype_storage.factories] extenstion point. All registered factories are loaded automatically. This function should be used if you want to bypass extension point mechanism and register addition genotype storage factory programatically.

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

Resolve repeated attributes in pipeline configuration via renaming.

dae.annotation.annotation_pipeline module

Provides annotation pipeline class.

class dae.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.

build_pipeline_genomic_context() → GenomicContext

Create a genomic context from the pipeline parameters.

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_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 dae.annotation.annotation_pipeline.Annotator(pipeline: AnnotationPipeline | None, info: AnnotatorInfo)

Bases: ABC

Annotator provides a set of attrubutes for a given Annotatable.

abstract 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

get_info() → AnnotatorInfo

is_open() → bool

open() → Annotator

property resource_ids: set[str]

property resources: list[GenomicResource]

property used_context_attributes: tuple[str, ...]

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

Bases: Annotator

Defines annotator decorator base class.

close() → None

is_open() → bool

open() → Annotator

class dae.annotation.annotation_pipeline.FullReannotationPipeline(pipeline_new: AnnotationPipeline, pipeline_old: AnnotationPipeline)

Bases: ReannotationPipeline

Special-case ReannotationPipeline.

Completely removes all old attributes and runs every new annotator, without reusing anything.

class dae.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 dae.annotation.annotation_pipeline.ReannotationPipeline(pipeline_new: AnnotationPipeline, pipeline_old: AnnotationPipeline)

Bases: AnnotationPipeline

Special pipeline that handles reannotation of a previous pipeline.

AnnotationDependencyGraph

alias of dict[AnnotatorInfo, list[tuple[AnnotatorInfo, AttributeInfo]]]

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

Apply all annotators to an annotatable.

static build_dependency_graph(pipeline: AnnotationPipeline) → AnnotationDependencyGraph

Make dependency graph for an annotation pipeline.

get_attributes() → list[AttributeInfo]

get_dependencies_for(info: AnnotatorInfo) → set[AnnotatorInfo]

Get all dependencies for a given annotator.

get_dependents_for(info: AnnotatorInfo) → set[AnnotatorInfo]

Get all dependents for a given annotator.

print() → None

Print the annotation pipeline.

class dae.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.

dae.annotation.annotator_base module

Provides base class for annotators.

class dae.annotation.annotator_base.AnnotatorBase(pipeline: AnnotationPipeline | None, info: AnnotatorInfo, source_type_desc: dict[str, tuple[str, str]])

Bases: Annotator

Base implementation of the Annotator class.

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

dae.annotation.cnv_collection_annotator module

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

Bases: Annotator

Simple effect annotator class.

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

Produce annotation attributes for an annotatable.

close() → None

open() → Annotator

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

dae.annotation.debug_annotator module

class dae.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.

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

Create an example hello world annotator.

dae.annotation.docker_annotator module

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

Bases: AnnotatorBase

Base class for annotators that use docker containers.

open() → Annotator

abstract run(**kwargs: Any) → None

dae.annotation.effect_annotator module

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

Bases: AnnotatorBase

Adapts effect annotator to be used in annotation infrastructure.

close() → None

open() → Annotator

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

dae.annotation.format_handlers module

class dae.annotation.format_handlers.AbstractFormat(pipeline_config: list[dict[str, Any]] | RawFullConfig, pipeline_config_old: str | None, cli_args: dict, grr_definition: dict | None, region: Region | None)

Bases: object

Abstract class of input/output handlers for various formats.

This class and its children are responsible for correctly reading from and writing to formats that can be annotated by our system.

They convert the raw input data to types that can be passed to the annotation pipeline and then convert it back to its native format, as well as handling the reading, updating and writing of metadata the format may possess.

Each child class handles the specific differences of a single format.

close() → None

Close any open files, clean up anything unnecessary.

static get_task_dir(region: Region | None) → str

Get dir for batch annotation.

open() → None

Initialize all member variables and process relevant metadata.

process() → None

Iteratively carry out the annotation of the input.

This method will read, annotate, apply and then write each variant from the input data in an iterative fashion - one by one.

process_batched() → None

Carry out the annotation of the input in batches.

This method performs each step of the read-annotate-apply-write loop in batches.

class dae.annotation.format_handlers.ColumnsFormat(pipeline_config: list[dict[str, Any]] | RawFullConfig, pipeline_config_old: str | None, cli_args: dict, grr_definition: dict | None, region: Region | None, input_path: str, output_path: str, ref_genome_id: str | None)

Bases: AbstractFormat

Handler for delimiter-separated values text files.

close() → None

Close any open files, clean up anything unnecessary.

open() → None

Initialize all member variables and process relevant metadata.

class dae.annotation.format_handlers.ParquetFormat(pipeline_config: list[dict[str, Any]] | RawFullConfig, pipeline_config_old: str | None, cli_args: dict, grr_definition: dict | None, region: Region | None, input_layout: Schema2DatasetLayout, output_dir: str, bucket_idx: int, variants_blob_serializer: str = 'json')

Bases: AbstractFormat

Handler for Schema2 Parquet datasets.

close() → None

Close any open files, clean up anything unnecessary.

open() → None

Initialize all member variables and process relevant metadata.

class dae.annotation.format_handlers.VCFFormat(pipeline_config: list[dict[str, Any]] | RawFullConfig, pipeline_config_old: str | None, cli_args: dict, grr_definition: dict | None, region: Region | None, input_path: str, output_path: str)

Bases: AbstractFormat

Handler for VCF format files.

close() → None

Close any open files, clean up anything unnecessary.

open() → None

Initialize all member variables and process relevant metadata.

dae.annotation.format_handlers.stringify(value: Any, *, vcf: bool = False) → str

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

dae.annotation.gene_score_annotator module

Module containing the gene score annotator.

class dae.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.

property used_context_attributes: tuple[str, ...]

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

Create a gene score annotator.

dae.annotation.gene_set_annotator module

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

Bases: AnnotatorBase

Gene set annotator class.

property used_context_attributes: tuple[str, ...]

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

Create a gene set annotator.

dae.annotation.genomic_context module

class dae.annotation.genomic_context.CLIAnnotationContextProvider

Bases: CLIGenomicContextProvider

Defines annotation pipeline genomics context provider.

static add_argparser_arguments(parser: ArgumentParser) → None

Add command line arguments to the argument parser.

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

Build a CLI genomic context.

dae.annotation.genomic_context.get_context_pipeline(context: GenomicContext) → AnnotationPipeline | None

Construct an annotation pipeline.

dae.annotation.liftover_annotator module

Provides a lift over annotator and helpers.

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

Bases: AnnotatorBase

Liftovver annotator class.

close() → None

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 dae.annotation.liftover_annotator.BasicLiftoverAnnotator(pipeline: AnnotationPipeline | None, info: AnnotatorInfo, chain: LiftoverChain, source_genome: ReferenceGenome, target_genome: ReferenceGenome)

Bases: AbstractLiftoverAnnotator

Basic liftover annotator class.

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

Bases: AbstractLiftoverAnnotator

BCF tools liftover re-implementation annotator class.

dae.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.

dae.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.

dae.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.

dae.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.

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

Create a liftover annotator.

dae.annotation.normalize_allele_annotator module

Provides normalize allele annotator and helpers.

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

Bases: AnnotatorBase

Annotator to normalize VCF alleles.

close() → None

open() → Annotator

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

dae.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

dae.annotation.processing_pipeline module

class dae.annotation.processing_pipeline.AnnotatablesBatchFilter

Bases: AbstractContextManager

A filter that can filter annotatables in batches.

abstract filter_batch(batch: Sequence[Annotation]) → Sequence[Annotation]

Filter annotatables in a single batch.

filter_batch_with_source(batch_with_source: Sequence[AnnotationsWithSource]) → Sequence[AnnotationsWithSource]

Filter a single batch of annotatables with source.

filter_batches(batches: Iterable[Sequence[Annotation]]) → Iterable[Sequence[Annotation]]

Filter annotatables in batches.

filter_batches_with_source(batches_with_source: Iterable[Sequence[AnnotationsWithSource]]) → Iterable[Sequence[AnnotationsWithSource]]

Filter annotatables with source in batches.

class dae.annotation.processing_pipeline.AnnotatablesFilter

Bases: AbstractContextManager

A filter that can filter annotatables.

filter(data: Iterable[Annotation]) → Iterable[Annotation]

Filter a single batch of annotatables.

abstract filter_one(annotation: Annotation) → Annotation

Filter a single annotatable.

filter_one_with_source(data: AnnotationsWithSource) → AnnotationsWithSource

filter_with_source(data: Iterable[AnnotationsWithSource]) → Iterable[AnnotationsWithSource]

Filter annotatables with source.

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

Bases: object

annotatable: Annotatable | None

context: dict[str, Any]

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

Bases: AnnotatablesBatchFilter, AnnotationPipelineContextManager

A filter that can filter annotatables in batches.

filter_batch(batch: Sequence[Annotation]) → Sequence[Annotation]

Filter annotatables in a single batch.

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

Bases: AnnotatablesFilter, AnnotationPipelineContextManager

A filter that can filter annotatables in batches.

filter_one(annotation: Annotation) → Annotation

Filter annotatables.

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

Bases: AbstractContextManager

A context manager for annotation pipelines.

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

Bases: object

annotations: list[Annotation]

source: Any

dae.annotation.reannotate_instance module

class dae.annotation.reannotate_instance.ReannotateInstanceTool(raw_args: list[str] | None = None, gpf_instance: GPFInstance | None = None)

Bases: object

Annotation tool to reannotate the configured GPF instance

get_argument_parser() → ArgumentParser

Construct and configure argument parser.

run() → None

Run the tool.

dae.annotation.reannotate_instance.cli(raw_args: list[str] | None = None, gpf_instance: GPFInstance | None = None) → None

Entry point method for instance reannotation tool.

dae.annotation.record_to_annotatable module

class dae.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

class dae.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

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

Bases: ABC

Base class for record to annotable transformation.

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

class dae.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

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

Bases: RecordToAnnotable

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

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

Bases: RecordToAnnotable

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

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

Bases: RecordToAnnotable

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

class dae.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

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

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

Transform a variant record into an annotatable.

dae.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 dae.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 dae.annotation.score_annotator.GenomicScoreAnnotatorBase(pipeline: AnnotationPipeline, info: AnnotatorInfo, score: GenomicScore)

Bases: Annotator

Genomic score base annotator.

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

Collect score aggregator documentation.

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

Construct score aggregator documentation.

close() → None

is_open() → bool

open() → Annotator

class dae.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.

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

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

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

dae.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.

dae.annotation.simple_effect_annotator module

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

Bases: AnnotatorBase

Simple effect annotator class.

call_region(chrom: str, beg: int, end: int, transcripts: list[TranscriptModel], func_name: str, classification: str) → tuple[str, set[str]] | None

Call a region with a specific classification.

cds_intron_regions(transcript: TranscriptModel) → list[Region]

Return whether region is CDS intron.

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) → tuple[str, set[str]]

Return classification with a set of affected genes.

utr_regions(transcript: TranscriptModel) → Sequence[Region]

Return whether the region is classified as UTR.

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

Module contents