Annotators¶

Driven adapters that enrich VariantAnnotation objects with knowledge from external reference databases. Each annotator takes a list of variants, looks up matching records, and returns a new list with the relevant fields populated. Variants not found in the source are returned unchanged.

Overview¶

How annotators work¶

Every annotator implements annotate(items) → list[VariantAnnotation]. Because domain objects are immutable, each enrichment step produces a new VariantAnnotation via attrs.evolve() — the input list is never modified in place.

The typical pipeline chains multiple annotators sequentially via VariantExtractionPipeline:

from iris.adapters.driven.pipeline import VariantExtractionPipeline
from iris.adapters.driven.stores.zarr import ZarrVariantStore
from iris.adapters.driven.annotators.vep_flight import VepFlightAnnotator
from iris.adapters.driven.annotators.dbnsfp import DbNsfpAnnotator
from iris.adapters.driven.annotators.clinvar import ClinvarAnnotator

with ZarrVariantStore(path=ZARR_PATH) as store:
    pipeline = VariantExtractionPipeline(
        store=store,
        annotators=(
            VepFlightAnnotator(location="grpc://vep-node:8815", profile="epi_default"),
            DbNsfpAnnotator(data_dir=DBNSFP_DIR, workers=8),
            ClinvarAnnotator(path=CLINVAR_TSV),
        ),
        workers=8,
    )
    variants = pipeline.run(regions=regions)

`TabixVariantAnnotator` — shared base¶

Most annotators extend TabixVariantAnnotator, which combines TabixReader I/O with the VariantAnnotator port. The core helper _annotate_rows() encapsulates the full pipeline:

Collect (chrom, pos, ref, alt) keys from the input variants.
Batch-query the tabix file for those keys.
For each matching row, call enrich_fn(variant, row) to produce the enriched variant.

columns restricts which fields are loaded to avoid saturating RAM with wide files (critical for dbNSFP, which has 200+ columns). workers enables chromosome-partitioned parallel fetching via ThreadPoolExecutor; each worker opens its own TabixFile.

Annotator catalog¶

Annotator	Module	What it adds
`TabixVariantAnnotator`	`tabix`	Base class; not used directly
`VcfVariantAnnotator`	`vcf`	Base class for VCF/BCF-backed annotators
`VepFlightAnnotator`	`vep_flight`	Genes, `TranscriptVariant` (HGVS, SO terms), LOEUF/pLI, SpliceAI, LoF (LOFTEE HC/LC) via live Arrow Flight service
`GencodeVariantAnnotator`	`gencode`	Genes and `TranscriptVariant` stubs from a local GENCODE GTF; no HGVS or scores
`DbNsfpAnnotator`	`dbnsfp`	Multiple `VariantScore` (AlphaMissense, REVEL, CADD, PolyPhen2, SIFT, BayesDel, MetaRNN, ESM1b, PrimateAI, ClinPred, …) + `AlleleFrequency` per gnomAD 4.1/TOPMed/ALFA population
`CaddAnnotator`	`cadd`	`VariantScore`: RawScore, PHRED
`McpsAnnotator`	`mcps`	`AlleleFrequency` for each MCPS population (AC, AN, AF)
`ClinvarAnnotator`	`clinvar`	`ClinicalClassification` (significance + review status)
`VepTsvAnnotator`	`vep`	Configurable `VariantScore` set via `ScoreSpec` from a pre-computed VEP TSV
`HgncNomenclatureResolver`	`annotators.hgnc`	`GeneNomenclature` with symbol, aliases, and cross-DB IDs (HGNC, Ensembl, RefSeq, Entrez) from a local HGNC JSON dump

TOML mapping tables¶

Each tabix-backed annotator ships a .toml file alongside its source that maps raw database strings to domain enum values (e.g. dbnsfp.toml maps AlphaMissense prediction letters "B", "A", "LP" to ScorePredictionRank). The [columns] section drives column filtering so only the needed subset of a wide file is loaded into memory.

Memory efficiency¶

Gene and Transcript objects are frozen (attrs frozen=True) and safe to share. VepFlightAnnotator maintains a per-call cache keyed by Ensembl ID so that a single Gene and Transcript object is reused across every variant that maps to the same gene or transcript, instead of allocating one object per variant.

DbNsfpAnnotator loads only the columns listed in dbnsfp.toml [columns] (≈ 48 of 200+ dbNSFP columns) and supports parallel per-chromosome loading via the workers field.

`VepFlightAnnotator`¶

The only non-tabix, non-file annotator. Calls a running vep-arrowd Rust service over Arrow Flight (gRPC). Requires vep_arrow_client to be installed:

uv pip install /path/to/vep-flight/pyclient

Key parameters:

Parameter	Default	Description
`location`	—	gRPC server address, e.g. `"grpc://10.42.42.5:8815"`
`profile`	`"epi_default"`	VEP plugin profile configured on the server
`transcript_mode`	`"all"`	`"picked"` keeps only VEP's `--pick` transcript per variant
`batch_size`	`10_000`	Variants per Arrow Flight batch
`timeout_seconds`	`3600`	Per-call gRPC timeout

Scores produced for the epi_default profile:

Score name	Type	Threshold
`LOEUF`	`CONSTRAINT`	`VERY_HIGH` if ≤ 0.35
`pLI`	`CONSTRAINT`	`VERY_HIGH` if ≥ 0.9
`SpliceAI_DS_{AG,AL,DG,DL}`	`SPLICING`	`VERY_HIGH` if ≥ 0.5
`LoF`	`PATHOGENICITY`	`VERY_HIGH` (HC) / `HIGH` (LC) from LOFTEE

`GencodeVariantAnnotator`¶

Gene and transcript annotation from a local GENCODE GTF, loaded via GencodeGeneRepository. Returns TranscriptVariant stubs (no HGVS, no pathogenicity scores). Useful when the VEP Flight service is unavailable or as a lightweight gene-assignment step. Resolves the GENCODE GTF path via DataRegistry.resolve_path("gencode"), configurable through the IRIS_GENCODE_GTF environment variable.

`HgncNomenclatureResolver`¶

Implements the GeneNomenclatureResolver port using the HGNC complete-set JSON dump (hgnc_complete_set.json), downloadable from genenames.org.

Builds six in-memory indices on first use (lazy load): approved symbol, alias/prev-symbol, HGNC ID, Ensembl gene ID, Entrez ID, and RefSeq accession. All subsequent lookups are O(1).

from iris.adapters.driven.annotators.hgnc import HgncNomenclatureResolver

# From the registry (reads [sources.hgnc] in ~/.iris/config.toml)
resolver = HgncNomenclatureResolver.from_registry()

# Or with an explicit path
resolver = HgncNomenclatureResolver(path="/data/hgnc/hgnc_complete_set.json")

resolver.resolve_symbol("MLL").symbol        # → "KMT2A"  (prev_symbol lookup)
resolver.resolve_symbol("RNF53").symbol      # → "BRCA1"  (alias_symbol lookup)
resolver.canonical_symbol("MLL")            # → "KMT2A"  (fast string-only helper)
resolver.resolve_hgnc_id("HGNC:1100").symbol # → "BRCA1"
resolver.resolve_ensembl_id("ENSG00000012048").symbol  # → "BRCA1"

Register the file path in ~/.iris/config.toml:

[sources.hgnc]
host = "local"
path = "hgnc/hgnc_complete_set.json"

The canonical_symbol(name) convenience method is used by the recgenes_vep_flight_pipeline to remap alias/previous gene symbols from a clinical gene list to the approved HGNC symbol before querying GENCODE.

Tabix base¶

iris.adapters.driven.annotators.tabix ¶

TabixVariantAnnotator base class combining tabix I/O with the VariantAnnotator port.

TabixVariantAnnotator ¶

Bases: TabixReader, VariantAnnotator

Base for variant annotators that query bgzipped, tabix-indexed files via pysam.

Subclasses implement annotate() using self._annotate_rows() as the main pipeline, or call self._query() / self._query_region() directly for custom traversal patterns.

annotate `abstractmethod` ¶

annotate(items)

Return new VariantAnnotation objects enriched with this source's data.

Source code in src/iris/domain/genomics/ports/driven/annotators.py

@abstractmethod
def annotate(self, items: Sequence[VariantAnnotation]) -> list[VariantAnnotation]:
    """Return new VariantAnnotation objects enriched with this source's data."""

VCF base¶

iris.adapters.driven.annotators.vcf ¶

VcfVariantAnnotator base class combining VCF I/O with the VariantAnnotator port.

VcfVariantAnnotator ¶

Bases: VcfReader, VariantAnnotator

Base for variant annotators that query bgzipped, indexed VCF/BCF files via pysam.

Subclasses implement annotate() using self._annotate_records() as the main pipeline. Requires a TBI or CSI index alongside the VCF/BCF file.

annotate `abstractmethod` ¶

annotate(items)

Return new VariantAnnotation objects enriched with this source's data.

Source code in src/iris/domain/genomics/ports/driven/annotators.py

@abstractmethod
def annotate(self, items: Sequence[VariantAnnotation]) -> list[VariantAnnotation]:
    """Return new VariantAnnotation objects enriched with this source's data."""

read ¶

read(path)

Read a VCF/BCF file and return a list of VariantAnnotation objects.

Source code in src/iris/adapters/driven/readers/vcf.py

def read(self, path: str | Path) -> list[VariantAnnotation]:
    """Read a VCF/BCF file and return a list of VariantAnnotation objects."""
    pysam = require_pysam()
    results: list[VariantAnnotation] = []
    with pysam.VariantFile(str(path)) as vcf:
        asm = self.assembly or _detect_assembly(vcf)
        sample_idx = self._resolve_sample(vcf)
        for record in vcf:
            results.extend(self._process_record(record, asm, sample_idx))
    return results

VEP Flight¶

iris.adapters.driven.annotators.vep_flight ¶

Annotator that calls an external VEP-via-Arrow-Flight service (vep_arrow_client).

Unlike VepTsvAnnotator (tabix lookups against a pre-computed file), this adapter calls a live VEP service over Arrow Flight for variants not yet annotated. It is a thin translation layer: vep_arrow_client.VepFlightAnnotator already handles the Flight protocol, batching, and Arrow conversion — this adapter's only job is mapping between VariantAnnotation and the RecordBatch rows the service returns.

VepFlightAnnotator ¶

VepFlightAnnotator(
    location,
    *,
    profile="epi_default",
    transcript_mode="all",
    batch_size=10000,
    timeout_seconds=3600,
    gene_cache=None,
)

Bases: VariantAnnotator

Annotates VariantAnnotation objects via an external VEP Arrow Flight service.

Wraps vep_arrow_client.VepFlightAnnotator (the external SDK — note the name collision is intentional, this class is the iris-lib adapter, the SDK class is the underlying client). Translates between VariantAnnotation/Variant and the SimpleVariant/RecordBatch contract of the external service.

PARAMETER	DESCRIPTION
`location`	Flight server address, e.g. 'grpc://vep-node.internal:8815'. TYPE: `str`
`profile`	VEP plugin profile name configured on the server (e.g. 'epi_default', 'clinical', 'epi_core', 'epi_gwas'). TYPE: `str` DEFAULT: `'epi_default'`
`transcript_mode`	'all' to keep every overlapping transcript consequence, or 'picked' to keep only VEP's --pick selection. TYPE: `str` DEFAULT: `'all'`
`batch_size`	Number of variants per Flight batch sent to the server. TYPE: `int` DEFAULT: `10000`
`timeout_seconds`	Per-call timeout passed to the underlying client. TYPE: `int` DEFAULT: `3600`

Example:

    annotator = VepFlightAnnotator(
        location="grpc://vep-node.internal:8815",
        profile="epi_default",
    )
    annotated = annotator.annotate(variants)

Initialize the annotator.

PARAMETER	DESCRIPTION
`location`	Flight server address. TYPE: `str`
`profile`	VEP plugin profile name. TYPE: `str` DEFAULT: `'epi_default'`
`transcript_mode`	'all' or 'picked'. TYPE: `str` DEFAULT: `'all'`
`batch_size`	Variants per Flight batch. TYPE: `int` DEFAULT: `10000`
`timeout_seconds`	Per-call timeout. TYPE: `int` DEFAULT: `3600`
`gene_cache`	Optional pre-built mapping of gene identifier (versioned/unversioned Ensembl ID, or gene symbol) to canonical Gene objects — typically built from GencodeGeneRepository before running the pipeline. VEP rows matching a key here reuse that exact Gene instance instead of constructing a new one from VEP's bare gene_id/symbol fields, keeping gene identity consistent across the whole pipeline. Persists and grows for the lifetime of this annotator instance across multiple annotate() calls. TYPE: `dict[str, Gene] \| None` DEFAULT: `None`

Source code in src/iris/adapters/driven/annotators/vep_flight.py

def __init__(
    self,
    location: str,
    *,
    profile: str = "epi_default",
    transcript_mode: str = "all",
    batch_size: int = 10_000,
    timeout_seconds: int = 3600,
    gene_cache: dict[str, Gene] | None = None,
) -> None:
    """Initialize the annotator.

    Args:
        location: Flight server address.
        profile: VEP plugin profile name.
        transcript_mode: 'all' or 'picked'.
        batch_size: Variants per Flight batch.
        timeout_seconds: Per-call timeout.
        gene_cache: Optional pre-built mapping of gene identifier
            (versioned/unversioned Ensembl ID, or gene symbol) to
            canonical Gene objects — typically built from
            GencodeGeneRepository before running the pipeline. VEP rows
            matching a key here reuse that exact Gene instance instead
            of constructing a new one from VEP's bare gene_id/symbol
            fields, keeping gene identity consistent across the whole
            pipeline. Persists and grows for the lifetime of this
            annotator instance across multiple annotate() calls.
    """
    try:
        from vep_arrow_client import VepFlightAnnotator as _VepClient  # pyright: ignore[reportMissingImports]
    except ImportError as exc:
        raise ImportError(
            "vep_arrow_client is required for VepFlightAnnotator: "
            "install it from the vep-arrow project."
        ) from exc

    self._client = _VepClient(
        location=location,
        profile=profile,
        transcript_mode=transcript_mode,
        timeout_seconds=timeout_seconds,
    )
    self._batch_size = batch_size

    # Persisted across multiple annotate() calls so genes pre-seeded
    # from an external source (e.g. GencodeGeneRepository) are reused
    # by reference for the lifetime of this instance, instead of being
    # rebuilt from VEP's bare gene_id/symbol fields every time.
    self._gene_cache: dict[str, Gene] = dict(gene_cache) if gene_cache else {}
    self._transcript_cache: dict[str, Transcript] = {}

ready ¶

ready()

Return the underlying service's readiness/health payload.

Source code in src/iris/adapters/driven/annotators/vep_flight.py

def ready(self) -> dict[str, Any]:
    """Return the underlying service's readiness/health payload."""
    return self._client.ready()

annotate ¶

annotate(items)

Return new VariantAnnotation objects enriched with VEP consequences and scores.

Variants without a complete (chrom, pos, ref, alt) key (i.e. Variant.as_key() returns None) are passed through unchanged — the service has nothing to annotate them with.

Source code in src/iris/adapters/driven/annotators/vep_flight.py

def annotate(self, items: Sequence[VariantAnnotation]) -> list[VariantAnnotation]:
    """Return new VariantAnnotation objects enriched with VEP consequences and scores.

    Variants without a complete (chrom, pos, ref, alt) key (i.e.
    Variant.as_key() returns None) are passed through unchanged — the
    service has nothing to annotate them with.
    """
    try:
        from vep_arrow_client.client import SimpleVariant  # pyright: ignore[reportMissingImports]
    except ImportError as exc:
        raise ImportError(
            "vep_arrow_client is required for VepFlightAnnotator: "
            "install it from the vep-arrow project."
        ) from exc

    # Map list-index -> source VariantAnnotation, and build the
    # SimpleVariant batch for every item with a resolvable key.
    # variant_uid is just the index — uniqueness within the call is all
    # that's required, no need to carry meaning across calls.
    indexed_keys: dict[int, tuple[str, int, str, str]] = {}
    simple_variants: list[Any] = []

    for i, va in enumerate(items):
        key = va.variant.as_key()
        if key is None:
            continue
        chrom, pos, ref, alt = key
        indexed_keys[i] = key
        simple_variants.append(
            SimpleVariant(
                variant_uid=i,
                input_order=i,
                chrom=chrom,
                pos=pos,
                ref=ref,
                alt=alt,
                variant_key=f"{chrom}:{pos}:{ref}:{alt}",
            )
        )

    if not simple_variants:
        return list(items)

    # VEP requires input sorted by chromosome then position.
    # variant_uid carries the original index so responses map back correctly.
    simple_variants.sort(key=lambda v: (chromosome_sort_key(v.chrom), v.pos))

    rows_by_uid: dict[int, list[dict[str, Any]]] = {i: [] for i in indexed_keys}

    for batch in self._client.annotate_iter(
        simple_variants, batch_size=self._batch_size, full_duplex=True
    ):
        for row in batch.to_pylist():
            rows_by_uid[row["variant_uid"]].append(row)

    result: list[VariantAnnotation] = []
    for i, va in enumerate(items):
        rows = rows_by_uid.get(i)
        if not rows:
            result.append(va)
            continue
        result.append(self._merge_annotation(va, rows))

    return result

GENCODE¶

iris.adapters.driven.annotators.gencode ¶

GencodeVariantAnnotator — gene and transcript annotation from a GENCODE repository.

GencodeVariantAnnotator ¶

Bases: VariantAnnotator

Annotate VariantAnnotation objects with genes and transcripts from GENCODE.

Builds a per-chromosome interval index from the repository on the first :meth:annotate call. Subsequent calls reuse the cached index.

For each variant the annotator:

Replaces va.genes with the full :class:~iris.domain.genomics.entities.Gene objects from GENCODE (including Ensembl ID, HGNC ID, strand, and position).
When annotate_transcripts is True, prepends :class:~iris.domain.genomics.entities.TranscriptVariant stubs for every transcript that overlaps the variant position. Positional fields (exon, intron, cdna_pos, cds_pos) are derived from the GENCODE structure; sequence-dependent fields are left for downstream annotators.
When primary_only is True, keeps only the single preferred transcript (MANE Select > Ensembl canonical > first) and records the total transcript count in va.metadata under "n_transcripts".

Variants whose position is unknown or that fall outside all annotated genes are passed through unchanged.

Example

gencode = GencodeGeneRepository(GTF_PATH, assembly="GRCh38", tags={"MANE_Select"})
annotator = GencodeVariantAnnotator(repository=gencode, primary_only=True)
annotated = annotator.annotate(variants)

ATTRIBUTE	DESCRIPTION
`repository`	A loaded (or lazily-loaded) :class:`~iris.adapters.driven.readers.providers.gencode.GencodeGeneRepository`. TYPE: `GencodeGeneRepository`
`annotate_transcripts`	When `True` (default), also populate `va.transcripts` with stubs derived from GENCODE transcripts. TYPE: `bool`
`primary_only`	When `True`, keep only the preferred transcript (MANE Select > canonical > first) and store the total count in metadata. TYPE: `bool`

annotate ¶

annotate(items)

Return variants enriched with GENCODE gene and transcript information.

PARAMETER	DESCRIPTION
`items`	Variants to annotate. TYPE: `Sequence[VariantAnnotation]`

RETURNS	DESCRIPTION
`list[VariantAnnotation]`	New list of :class:`~iris.domain.genomics.entities.VariantAnnotation`
`list[VariantAnnotation]`	objects. Variants outside any annotated gene region are passed
`list[VariantAnnotation]`	through unchanged.

Source code in src/iris/adapters/driven/annotators/gencode.py

def annotate(self, items: Sequence[VariantAnnotation]) -> list[VariantAnnotation]:
    """Return variants enriched with GENCODE gene and transcript information.

    Args:
        items: Variants to annotate.

    Returns:
        New list of :class:`~iris.domain.genomics.entities.VariantAnnotation`
        objects.  Variants outside any annotated gene region are passed
        through unchanged.
    """
    self._ensure_loaded()
    result: list[VariantAnnotation] = []

    for va in items:
        pos = va.variant.position
        if pos is None or pos.start is None:
            result.append(va)
            continue

        chrom = HUMAN_CHROMOSOMES.resolve(pos.chromosome)
        gene_anns = _overlapping_genes(self._index.get(chrom, []), pos.start)

        if not gene_anns:
            result.append(va)
            continue

        genes = tuple(ann.gene for ann in gene_anns)

        if not self.annotate_transcripts:
            result.append(attrs.evolve(va, genes=genes))
            continue

        tcs = _transcript_consequences(gene_anns, pos.start)

        # add transcript count to metadata, preserving existing metadata if present
        meta = (
            va.metadata.with_data(n_transcripts=len(tcs))
            if va.metadata is not None
            else Metadata(source="gencode", data={"n_transcripts": len(tcs)})
        )

        if self.primary_only and tcs:
            primary = _select_primary(tcs)
            kept = (primary,) if primary is not None else ()
            result.append(
                attrs.evolve(
                    va, genes=genes, transcripts=(*va.transcripts, *kept), metadata=meta
                )
            )
        else:
            result.append(
                attrs.evolve(
                    va, genes=genes, transcripts=(*va.transcripts, *tcs), metadata=meta
                )
            )

    return result

DbNSFP¶

iris.adapters.driven.annotators.dbnsfp ¶

dbNSFP in-silico scores and population frequency annotator via tabix.

DbNsfpAnnotator ¶

Bases: TabixVariantAnnotator

Annotates variants with in-silico scores and population data from dbNSFP via tabix.

AlphaMissense, REVEL, CADD, PolyPhen2-HDIV, SIFT, BayesDel (addAF/noAF),

GERP++RS, MetaRNN.

Extended scores: ESM1b, MutScore, PHACTboost, gMVP, MPC, PrimateAI, SIFT4G, VEST4, ClinPred, GERP91, phyloP/phastCons (100way/470way/17way-primate), bStatistic. Metadata: ClinVar (id/clnsig/review), gnomAD4.1 (joint + 5 populations), TOPMed, ALFA, Interpro domain.

Expects bgzipped, tabix-indexed per-chromosome files under data_dir matching file_pattern ({chrom} substituted with the bare chromosome, e.g. "1"). For multi-transcript rows, the most damaging value is kept per score.

from_registry `classmethod` ¶

from_registry(source='dbnsfp', registry=None)

Build a DbNsfpAnnotator resolving source from the iris DataRegistry.

Source code in src/iris/adapters/driven/annotators/dbnsfp.py

@classmethod
def from_registry(cls, source: str = "dbnsfp", registry=None) -> "DbNsfpAnnotator":
    """Build a DbNsfpAnnotator resolving *source* from the iris DataRegistry."""
    from iris.config import get_registry

    r = registry or get_registry()
    return cls(data_dir=r.resolve_path(source))

annotate ¶

annotate(items)

Annotate variants with dbNSFP in-silico scores and allele frequencies.

Source code in src/iris/adapters/driven/annotators/dbnsfp.py

def annotate(self, items: Sequence[VariantAnnotation]) -> list[VariantAnnotation]:
    """Annotate variants with dbNSFP in-silico scores and allele frequencies."""
    targets: set[_Key] = set()
    for va in items:
        maybe_key = va.variant.as_key()
        if maybe_key is None:
            continue
        chrom, pos, ref, alt = maybe_key
        targets.add((normalize_chromosome(chrom), pos, ref, alt))
    if not targets:
        return list(items)
    chroms = {key[0] for key in targets}
    hit_map = self._load(chroms, targets, workers=self.workers)
    return [self._enrich(va, hit_map) for va in items]

CADD¶

iris.adapters.driven.annotators.cadd ¶

CADD RawScore and PHRED annotator via tabix random-access.

CaddAnnotator ¶

Bases: TabixVariantAnnotator

Annotates variants with CADD RawScore and PHRED via tabix random-access.

Requires the CADD TSV to be bgzipped and tabix-indexed (.tbi alongside it).

from_registry `classmethod` ¶

from_registry(source='cadd', registry=None)

Build a CaddAnnotator resolving source from the iris DataRegistry.

Source code in src/iris/adapters/driven/annotators/cadd.py

@classmethod
def from_registry(cls, source: str = "cadd", registry=None) -> "CaddAnnotator":
    """Build a CaddAnnotator resolving *source* from the iris DataRegistry."""
    from iris.config import get_registry

    r = registry or get_registry()
    return cls(tsv_path=r.resolve_path(source))

annotate ¶

annotate(items)

Annotate variants with CADD RawScore and PHRED scores.

Source code in src/iris/adapters/driven/annotators/cadd.py

def annotate(self, items: Sequence[VariantAnnotation]) -> list[VariantAnnotation]:
    """Annotate variants with CADD RawScore and PHRED scores."""
    targets: set[_Key] = set()
    for va in items:
        maybe_key = va.variant.as_key()
        if maybe_key is None:
            continue
        targets.add(maybe_key)
    if not targets:
        return list(items)
    scores = self._load(targets)
    return [self._enrich(va, scores) for va in items]

MCPS¶

iris.adapters.driven.annotators.mcps ¶

MCPS allele frequency annotator via tabix random-access.

McpsAFAnnotator ¶

Bases: TabixVariantAnnotator

Annotates variants with MCPS allele frequencies (global + AFR/EUR/MEX ancestries).

Expects bgzipped, tabix-indexed per-chromosome files named chr{N}.freq.cpra.tsv.bgz under freq_dir. Column layout is defined in mcps.toml and matches the file header::

#CHROM  START  END  REF  ALT  ID  CPRA  SOURCE  STON  GHOM  AHOM
AN_RAW  AC_RAW  AF_RAW  AN_AFR  AC_AFR  AF_AFR
AN_EUR  AC_EUR  AF_EUR  AN_MEX  AC_MEX  AF_MEX

from_registry `classmethod` ¶

from_registry(source_name='mcps', registry=None)

Build a McpsAFAnnotator resolving source_name from the iris DataRegistry.

Source code in src/iris/adapters/driven/annotators/mcps.py

@classmethod
def from_registry(cls, source_name: str = "mcps", registry=None) -> "McpsAFAnnotator":
    """Build a McpsAFAnnotator resolving *source_name* from the iris DataRegistry."""
    from iris.config import get_registry

    r = registry or get_registry()
    return cls(freq_dir=r.resolve_path(source_name))

annotate ¶

annotate(items, *, workers=1)

Annotate variants with MCPS allele frequencies.

workers controls how many per-chromosome files are queried in parallel. Each worker opens its own TabixFile; the default of 1 preserves sequential behaviour.

Source code in src/iris/adapters/driven/annotators/mcps.py

def annotate(
    self, items: Sequence[VariantAnnotation], *, workers: int = 1
) -> list[VariantAnnotation]:
    """Annotate variants with MCPS allele frequencies.

    ``workers`` controls how many per-chromosome files are queried in
    parallel.  Each worker opens its own ``TabixFile``; the default of 1
    preserves sequential behaviour.
    """
    targets: set[_Key] = set()
    for va in items:
        maybe_key = va.variant.as_key()
        if maybe_key is None:
            continue
        targets.add(maybe_key)
    if not targets:
        return list(items)
    chroms = {key[0] for key in targets}
    freq_map = self._load(chroms, targets, workers=workers)
    return [self._enrich(va, freq_map) for va in items]

query_region ¶

query_region(regions, *, workers=1)

Return all MCPS variants whose positions overlap the given regions.

Each row in the per-chromosome frequency file becomes a fresh VariantAnnotation containing the parsed allele frequencies. workers controls how many chromosome files are scanned in parallel. Files that do not exist on disk are silently skipped.

Source code in src/iris/adapters/driven/annotators/mcps.py

def query_region(
    self, regions: GenomicPosition | Sequence[GenomicPosition], *, workers: int = 1
) -> list[VariantAnnotation]:
    """Return all MCPS variants whose positions overlap the given regions.

    Each row in the per-chromosome frequency file becomes a fresh
    ``VariantAnnotation`` containing the parsed allele frequencies.
    ``workers`` controls how many chromosome files are scanned in parallel.
    Files that do not exist on disk are silently skipped.
    """
    if isinstance(regions, GenomicPosition):
        regions = (regions,)

    by_chrom: dict[str, list[GenomicPosition]] = {}
    for region in regions:
        by_chrom.setdefault(region.chromosome, []).append(region)

    batches = [
        (Path(self.freq_dir) / self.file_pattern.format(chrom=chrom), chrom_regions)
        for chrom, chrom_regions in by_chrom.items()
        if (Path(self.freq_dir) / self.file_pattern.format(chrom=chrom)).exists()
    ]
    if not batches:
        return []

    n_workers = min(workers, len(batches))
    if n_workers <= 1:
        results: list[VariantAnnotation] = []
        for path, chrom_regions in batches:
            results.extend(self._scan_region(path, chrom_regions))
        return results

    results = []
    with ThreadPoolExecutor(max_workers=n_workers) as pool:
        futures = [pool.submit(self._scan_region, path, cr) for path, cr in batches]
        for future in as_completed(futures):
            results.extend(future.result())
    return results

ClinVar¶

iris.adapters.driven.annotators.clinvar ¶

ClinVar variant classifier via a bgzipped, tabix-indexed TSV.

ClinvarAnnotator ¶

Bases: TabixVariantAnnotator

Annotates variants with ClinVar classifications from a bgzipped, tabix-indexed TSV.

Expects the flat TSV produced from the ClinVar VCF (one row per variant): #CHROM POS REF ALT ALLELEID CLNSIG CLNREVSTAT AF_ESP AF_EXAC AF_TGP …

Frequency column → population label mapping is configured in clinvar.toml under [frequencies]. Classification and review-status maps live in the same file under [clinvar_significance] and [clinvar_review_status].

from_registry `classmethod` ¶

from_registry(source='clinvar', registry=None)

Build a ClinvarAnnotator resolving source from the iris DataRegistry.

Source code in src/iris/adapters/driven/annotators/clinvar.py

@classmethod
def from_registry(cls, source: str = "clinvar", registry=None) -> "ClinvarAnnotator":
    """Build a ClinvarAnnotator resolving *source* from the iris DataRegistry."""
    from iris.config import get_registry

    r = registry or get_registry()
    return cls(path=r.resolve_path(source))

annotate ¶

annotate(items, *, workers=1)

Annotate variants with ClinVar classifications and allele frequencies.

Source code in src/iris/adapters/driven/annotators/clinvar.py

def annotate(
    self, items: Sequence[VariantAnnotation], *, workers: int = 1
) -> list[VariantAnnotation]:
    """Annotate variants with ClinVar classifications and allele frequencies."""
    return self._annotate_rows(
        items,
        Path(self.path),
        key_fn=_key_fn,
        enrich_fn=self._enrich,
        columns=_COLUMNS,
        workers=workers,
    )

VEP TSV¶

iris.adapters.driven.annotators.vep ¶

VEP TSV variant annotator via tabix.

VepTsvAnnotator ¶

Bases: TabixVariantAnnotator

Annotates variants with scores from a bgzipped, tabix-indexed VEP TSV.

The file is expected to have a header line (leading # is stripped automatically) with Uploaded_variation encoded as chrom:pos:ref:alt.

Which scores to extract is controlled by score_specs. By default the specs defined in vep.toml are used; pass a custom tuple to override. Columns not present in the file are silently ignored.

Example — annotate with only CADD and LoF:

from iris.adapters.driven.annotators._specs import ScoreSpec
from iris.domain.genomics.objects import VariantScoreType, ScorePredictionRank

annotator = VepTsvAnnotator(
    path="/data/vep.tsv.gz",
    score_specs=(
        ScoreSpec(column="CADD_PHRED", name="cadd_phred", type=VariantScoreType.PATHOGENICITY),
        ScoreSpec(
            column="LoF",
            name="lof",
            type=VariantScoreType.PATHOGENICITY,
            prediction_map={"HC": ScorePredictionRank.VERY_HIGH, "LC": ScorePredictionRank.HIGH},
        ),
    ),
)

annotate ¶

annotate(items, *, workers=1)

Annotate variants with VEP scores.

Source code in src/iris/adapters/driven/annotators/vep.py

def annotate(
    self, items: Sequence[VariantAnnotation], *, workers: int = 1
) -> list[VariantAnnotation]:
    """Annotate variants with VEP scores."""
    columns = [_KEY_COLUMN, *(s.column for s in self.score_specs)]
    return self._annotate_rows(
        items,
        Path(self.path),
        key_fn=self._key_fn,
        enrich_fn=self._enrich,
        columns=columns,
        workers=workers,
    )

HGNC¶

iris.adapters.driven.annotators.hgnc ¶

HGNC JSON dump adapter for the GeneNomenclatureResolver port.

Reads the HGNC complete set JSON (downloadable from genenames.org) and builds in-memory indices for fast symbol, alias, Ensembl, HGNC-ID, Entrez, and RefSeq lookups. The JSON is loaded lazily on first use.

Download the JSON from

https://www.genenames.org/download/statistics-and-files/ → "Complete HGNC approved gene symbol report" → JSON format

Then register it in ~/.iris/config.toml: [sources.hgnc] host = "local" path = "hgnc/hgnc_complete_set.json"

HgncNomenclatureResolver ¶

Bases: GeneNomenclatureResolver

GeneNomenclatureResolver backed by the HGNC complete-set JSON dump.

Accepts a path to hgnc_complete_set.json (downloadable from genenames.org). Builds in-memory indices (symbol, alias/prev-symbol, HGNC ID, Ensembl ID, Entrez ID, RefSeq accession) on first use; subsequent lookups are O(1).

Example

from iris.adapters.driven.annotators.hgnc import HgncNomenclatureResolver

resolver = HgncNomenclatureResolver.from_registry()
nom = resolver.resolve_symbol("MLL")  # alias → KMT2A
nom.symbol
# 'KMT2A'

from_registry `classmethod` ¶

from_registry(source='hgnc', registry=None)

Instantiate from the iris DataRegistry (reads ~/.iris/config.toml).

PARAMETER	DESCRIPTION
`source`	Registry source name (default: `"hgnc"`). TYPE: `str` DEFAULT: `'hgnc'`
`registry`	Optional pre-built DataRegistry; uses the default if None. TYPE: `Any` DEFAULT: `None`

RETURNS	DESCRIPTION
`'HgncNomenclatureResolver'`	HgncNomenclatureResolver pointed at the configured JSON file.

Source code in src/iris/adapters/driven/annotators/hgnc.py

@classmethod
def from_registry(
    cls, source: str = "hgnc", registry: Any = None
) -> "HgncNomenclatureResolver":
    """Instantiate from the iris DataRegistry (reads ~/.iris/config.toml).

    Args:
        source: Registry source name (default: ``"hgnc"``).
        registry: Optional pre-built DataRegistry; uses the default if None.

    Returns:
        HgncNomenclatureResolver pointed at the configured JSON file.
    """
    from iris.config import get_registry

    r = registry or get_registry()
    return cls(path=r.resolve_path(source))

resolve_symbol ¶

resolve_symbol(symbol)

Resolve a gene symbol, checking approved symbols then aliases/prev-symbols.

PARAMETER	DESCRIPTION
`symbol`	Gene symbol to resolve (e.g. 'BRCA1', 'MLL'). TYPE: `str`

RETURNS	DESCRIPTION
`GeneNomenclature \| None`	Canonical GeneNomenclature if found, None otherwise.

Source code in src/iris/adapters/driven/annotators/hgnc.py

def resolve_symbol(self, symbol: str) -> GeneNomenclature | None:
    """Resolve a gene symbol, checking approved symbols then aliases/prev-symbols.

    Args:
        symbol: Gene symbol to resolve (e.g. 'BRCA1', 'MLL').

    Returns:
        Canonical GeneNomenclature if found, None otherwise.
    """
    self._load()
    doc = self._by_symbol.get(symbol) or self._doc_via_alias(symbol)
    return _nom_from_doc(doc) if doc else None

resolve_hgnc_id ¶

resolve_hgnc_id(hgnc_id)

Resolve by HGNC ID (e.g. 'HGNC:1100' or '1100').

PARAMETER	DESCRIPTION
`hgnc_id`	HGNC identifier string. TYPE: `str`

RETURNS	DESCRIPTION
`GeneNomenclature \| None`	GeneNomenclature if found, None otherwise.

Source code in src/iris/adapters/driven/annotators/hgnc.py

def resolve_hgnc_id(self, hgnc_id: str) -> GeneNomenclature | None:
    """Resolve by HGNC ID (e.g. 'HGNC:1100' or '1100').

    Args:
        hgnc_id: HGNC identifier string.

    Returns:
        GeneNomenclature if found, None otherwise.
    """
    self._load()
    normalized = _hgnc_id_str(hgnc_id)
    if normalized is None:
        return None
    symbol = self._by_hgnc_id.get(normalized)
    doc = self._by_symbol.get(symbol) if symbol else None
    return _nom_from_doc(doc) if doc else None

resolve_ensembl_id ¶

resolve_ensembl_id(ensembl_id)

Resolve by Ensembl gene ID (e.g. 'ENSG00000012048').

PARAMETER	DESCRIPTION
`ensembl_id`	Ensembl gene identifier. TYPE: `str`

RETURNS	DESCRIPTION
`GeneNomenclature \| None`	GeneNomenclature if found, None otherwise.

Source code in src/iris/adapters/driven/annotators/hgnc.py

def resolve_ensembl_id(self, ensembl_id: str) -> GeneNomenclature | None:
    """Resolve by Ensembl gene ID (e.g. 'ENSG00000012048').

    Args:
        ensembl_id: Ensembl gene identifier.

    Returns:
        GeneNomenclature if found, None otherwise.
    """
    self._load()
    symbol = self._by_ensembl.get(ensembl_id)
    doc = self._by_symbol.get(symbol) if symbol else None
    return _nom_from_doc(doc) if doc else None

resolve_refseq_id ¶

resolve_refseq_id(refseq_id)

Resolve by RefSeq accession (e.g. 'NM_007294').

PARAMETER	DESCRIPTION
`refseq_id`	RefSeq accession string. TYPE: `str`

RETURNS	DESCRIPTION
`GeneNomenclature \| None`	GeneNomenclature if found, None otherwise.

Source code in src/iris/adapters/driven/annotators/hgnc.py

def resolve_refseq_id(self, refseq_id: str) -> GeneNomenclature | None:
    """Resolve by RefSeq accession (e.g. 'NM_007294').

    Args:
        refseq_id: RefSeq accession string.

    Returns:
        GeneNomenclature if found, None otherwise.
    """
    self._load()
    symbol = self._by_refseq.get(refseq_id)
    doc = self._by_symbol.get(symbol) if symbol else None
    return _nom_from_doc(doc) if doc else None

resolve_entrez_id ¶

resolve_entrez_id(entrez_id)

Resolve by NCBI Entrez gene ID.

PARAMETER	DESCRIPTION
`entrez_id`	NCBI Entrez gene identifier. TYPE: `int`

RETURNS	DESCRIPTION
`GeneNomenclature \| None`	GeneNomenclature if found, None otherwise.

Source code in src/iris/adapters/driven/annotators/hgnc.py

def resolve_entrez_id(self, entrez_id: int) -> GeneNomenclature | None:
    """Resolve by NCBI Entrez gene ID.

    Args:
        entrez_id: NCBI Entrez gene identifier.

    Returns:
        GeneNomenclature if found, None otherwise.
    """
    self._load()
    symbol = self._by_entrez.get(str(entrez_id))
    doc = self._by_symbol.get(symbol) if symbol else None
    return _nom_from_doc(doc) if doc else None

search_by_name ¶

search_by_name(name)

Search genes by full or partial name (case-insensitive substring match).

PARAMETER	DESCRIPTION
`name`	Full or partial gene name. TYPE: `str`

RETURNS	DESCRIPTION
`list[GeneNomenclature]`	List of matching GeneNomenclature objects, possibly empty.

Source code in src/iris/adapters/driven/annotators/hgnc.py

def search_by_name(self, name: str) -> list[GeneNomenclature]:
    """Search genes by full or partial name (case-insensitive substring match).

    Args:
        name: Full or partial gene name.

    Returns:
        List of matching GeneNomenclature objects, possibly empty.
    """
    self._load()
    lower = name.lower()
    return [
        _nom_from_doc(doc)
        for doc in (self._docs or [])
        if lower in (doc.get("name") or "").lower()
    ]

aliases ¶

aliases(symbol)

Return all known alias and previous symbols for a gene.

PARAMETER	DESCRIPTION
`symbol`	Approved gene symbol. TYPE: `str`

RETURNS	DESCRIPTION
`list[str]`	List of alias and previous symbols. Empty if gene not found.

Source code in src/iris/adapters/driven/annotators/hgnc.py

def aliases(self, symbol: str) -> list[str]:
    """Return all known alias and previous symbols for a gene.

    Args:
        symbol: Approved gene symbol.

    Returns:
        List of alias and previous symbols. Empty if gene not found.
    """
    doc = self._doc(symbol)
    if doc is None:
        return []
    return _coerce_list(doc.get("alias_symbol")) + _coerce_list(doc.get("prev_symbol"))

gene_family ¶

gene_family(symbol)

Return gene family names associated with a gene symbol.

PARAMETER	DESCRIPTION
`symbol`	Approved gene symbol. TYPE: `str`

RETURNS	DESCRIPTION
`list[str]`	List of gene family names.

Source code in src/iris/adapters/driven/annotators/hgnc.py

def gene_family(self, symbol: str) -> list[str]:
    """Return gene family names associated with a gene symbol.

    Args:
        symbol: Approved gene symbol.

    Returns:
        List of gene family names.
    """
    doc = self._doc(symbol)
    if doc is None:
        return []
    return _coerce_list(doc.get("gene_group") or doc.get("gene_family"))

orthologs ¶

orthologs(symbol, *, species=None)

Return orthology data for a gene.

The HGNC JSON does not include orthology predictions; this always returns an empty list. Use a dedicated orthology source if needed.

PARAMETER	DESCRIPTION
`symbol`	Approved gene symbol. TYPE: `str`
`species`	Ignored (no orthology data in the HGNC JSON dump). TYPE: `int \| None` DEFAULT: `None`

RETURNS	DESCRIPTION
`list[dict[str, str]]`	Always an empty list.

Source code in src/iris/adapters/driven/annotators/hgnc.py

def orthologs(self, symbol: str, *, species: int | None = None) -> list[dict[str, str]]:
    """Return orthology data for a gene.

    The HGNC JSON does not include orthology predictions; this always
    returns an empty list.  Use a dedicated orthology source if needed.

    Args:
        symbol: Approved gene symbol.
        species: Ignored (no orthology data in the HGNC JSON dump).

    Returns:
        Always an empty list.
    """
    return []

resolve_many ¶

resolve_many(symbols)

Resolve multiple gene symbols in a single pass.

Loads the index once, then performs O(1) lookups per symbol.

PARAMETER	DESCRIPTION
`symbols`	Sequence of gene symbols to resolve. TYPE: `Sequence[str]`

RETURNS	DESCRIPTION
`dict[str, GeneNomenclature \| None]`	Dict mapping each input symbol to its GeneNomenclature or None.

Source code in src/iris/adapters/driven/annotators/hgnc.py

def resolve_many(self, symbols: Sequence[str]) -> dict[str, GeneNomenclature | None]:
    """Resolve multiple gene symbols in a single pass.

    Loads the index once, then performs O(1) lookups per symbol.

    Args:
        symbols: Sequence of gene symbols to resolve.

    Returns:
        Dict mapping each input symbol to its GeneNomenclature or None.
    """
    self._load()
    return {sym: self.resolve_symbol(sym) for sym in symbols}

canonical_symbol ¶

canonical_symbol(name)

Return the HGNC-approved symbol for a name (alias, prev, or approved).

Convenience method for the pipeline: given any gene name, return the approved symbol that GENCODE is likely to use, or None if not found.

PARAMETER	DESCRIPTION
`name`	Any gene name or alias. TYPE: `str`

RETURNS	DESCRIPTION
`str \| None`	Approved HGNC symbol if found, None otherwise.

Source code in src/iris/adapters/driven/annotators/hgnc.py

def canonical_symbol(self, name: str) -> str | None:
    """Return the HGNC-approved symbol for a name (alias, prev, or approved).

    Convenience method for the pipeline: given any gene name, return the
    approved symbol that GENCODE is likely to use, or None if not found.

    Args:
        name: Any gene name or alias.

    Returns:
        Approved HGNC symbol if found, None otherwise.
    """
    self._load()
    if name in self._by_symbol:
        return name
    return self._alias_to_approved.get(name)

Annotators¶

Overview¶

How annotators work¶

TabixVariantAnnotator — shared base¶

Annotator catalog¶

TOML mapping tables¶

Memory efficiency¶

VepFlightAnnotator¶

GencodeVariantAnnotator¶

HgncNomenclatureResolver¶

Tabix base¶

iris.adapters.driven.annotators.tabix ¶

TabixVariantAnnotator ¶

annotate abstractmethod ¶

VCF base¶

iris.adapters.driven.annotators.vcf ¶

VcfVariantAnnotator ¶

annotate abstractmethod ¶

read ¶

VEP Flight¶

iris.adapters.driven.annotators.vep_flight ¶

VepFlightAnnotator ¶

ready ¶

annotate ¶

GENCODE¶

iris.adapters.driven.annotators.gencode ¶

GencodeVariantAnnotator ¶

annotate ¶

DbNSFP¶

iris.adapters.driven.annotators.dbnsfp ¶

DbNsfpAnnotator ¶

from_registry classmethod ¶

annotate ¶

CADD¶

iris.adapters.driven.annotators.cadd ¶

CaddAnnotator ¶

from_registry classmethod ¶

annotate ¶

MCPS¶

iris.adapters.driven.annotators.mcps ¶

McpsAFAnnotator ¶

from_registry classmethod ¶

annotate ¶

query_region ¶

ClinVar¶

iris.adapters.driven.annotators.clinvar ¶

ClinvarAnnotator ¶

from_registry classmethod ¶

annotate ¶

VEP TSV¶

iris.adapters.driven.annotators.vep ¶

VepTsvAnnotator ¶

annotate ¶

HGNC¶

iris.adapters.driven.annotators.hgnc ¶

HgncNomenclatureResolver ¶

from_registry classmethod ¶

resolve_symbol ¶

resolve_hgnc_id ¶

resolve_ensembl_id ¶

resolve_refseq_id ¶

resolve_entrez_id ¶

search_by_name ¶

aliases ¶

gene_family ¶

orthologs ¶

resolve_many ¶

canonical_symbol ¶

`TabixVariantAnnotator` — shared base¶

`VepFlightAnnotator`¶

`GencodeVariantAnnotator`¶

`HgncNomenclatureResolver`¶

annotate `abstractmethod` ¶

annotate `abstractmethod` ¶

from_registry `classmethod` ¶

from_registry `classmethod` ¶

from_registry `classmethod` ¶

from_registry `classmethod` ¶

from_registry `classmethod` ¶