Annotators API

Ordinary annotation uses the default without selecting an implementation. See experimental annotators for alternatives and Writing an annotator for extensions.

Annotators

varcode.EffectAnnotator

Bases: Protocol

Protocol for an object that annotates variant effects on transcripts.

Conforming objects expose:

• name — short identifier (e.g. "fast") used in the registry and in serialized provenance.
• :meth:annotate_on_transcript — the per-transcript entry point, returning a MutationEffect or NotImplemented.

Return Python's NotImplemented singleton when this particular input is unsupported. Public prediction APIs expose it as an Unresolved effect with a reason, retaining this annotator's provenance. They never silently replace an experimental result with the default's prediction. None is invalid; exceptions retain normal error-handling semantics.

Optionally exposes version (string) — used in CSV provenance headers so readers can detect when a serialized collection came from a different annotator version. Built-in annotators track varcode's version; third-party annotators expose their own.

Optionally implement annotate_with_context(variant, transcript, germline_ctx, phase_resolver=None) with the same return contract. Without it, nonempty germline context is unsupported. Empty context calls annotate_on_transcript as usual.

The contract is duck-typed (@runtime_checkable) so third-party annotators don't need to inherit from varcode just to register.

varcode.FastEffectAnnotator

Annotate point edits and structural variants through one interface.

version = _varcode_version class-attribute instance-attribute

Built-in annotators track varcode's own version. Third-party annotators (isovar's plugin, exacto's plugin) expose their own version string here; CSV provenance headers and round-trip warnings read from this field. See #271.

annotate_on_transcript(variant, transcript)

Delegate to the existing per-transcript prediction.

Returns the raw effect class (ExonicSpliceSite / SpliceDonor / etc. for splice disruptions), not wrapped in SpliceOutcomeSet. The wrap is applied at the collection boundary in :func:predict_variant_effects so internal consumers (notably the protein_diff annotator's dual dispatch) can still pattern-match on the raw class.

Source code in varcode/annotators/fast.py

def annotate_on_transcript(self, variant, transcript):
    """Delegate to the existing per-transcript prediction.

    Returns the raw effect class (``ExonicSpliceSite`` /
    ``SpliceDonor`` / etc. for splice disruptions), **not** wrapped
    in ``SpliceOutcomeSet``. The wrap is applied at the collection
    boundary in :func:`predict_variant_effects` so internal
    consumers (notably the ``protein_diff`` annotator's dual
    dispatch) can still pattern-match on the raw class.
    """
    if getattr(variant, "is_structural", False):
        from ..effects.structural import predict_structural_variant_effect
        return predict_structural_variant_effect(variant, transcript)
    # Lazy import avoids a circular dep at package import time.
    from ..effects.effect_prediction import (
        _predict_variant_effect_on_transcript_raw,
    )
    return _predict_variant_effect_on_transcript_raw(variant, transcript)

annotate_with_context(variant, transcript, germline_ctx, phase_resolver=None)

Use the established patient-baseline path for point edits.

Structural haplotype composition remains experimental in transcript_model. Do not send an SV's placeholder alleles to the point-edit builder.

Source code in varcode/annotators/fast.py

def annotate_with_context(
        self, variant, transcript, germline_ctx, phase_resolver=None):
    """Use the established patient-baseline path for point edits.

    Structural haplotype composition remains experimental in ``transcript_model``.
    Do not send an SV's placeholder alleles to the point-edit builder.
    """
    if getattr(variant, "is_structural", False):
        return NotImplemented
    from ..germline import predict_germline_aware_effect
    return predict_germline_aware_effect(
        variant, transcript, germline_ctx, annotator=self,
        phase_resolver=phase_resolver)

varcode.ProteinDiffEffectAnnotator

Classify effects by diffing translated mutant protein against the reference protein.

Experimental alternative for point edits. The parity harness compares it with the default on their shared domain and records intentional divergences with issue links. Structural variants return NotImplemented. Shared splice/germline classification helpers do not depend on this annotator.

annotate_with_context(variant, transcript, germline_ctx, phase_resolver=None)

Share the default's patient-baseline path for point edits.

Source code in varcode/annotators/protein_diff.py

def annotate_with_context(
        self, variant, transcript, germline_ctx, phase_resolver=None):
    """Share the default's patient-baseline path for point edits."""
    return FastEffectAnnotator.annotate_with_context(
        self, variant, transcript, germline_ctx,
        phase_resolver=phase_resolver)

annotate_on_transcript(variant, transcript)

Classify the effect of variant on transcript.

Runs fast first to detect splice-adjacent variants (which stay fast-classified); for everything else, builds a :class:MutantTranscript and diffs the translated protein.

Source code in varcode/annotators/protein_diff.py

def annotate_on_transcript(self, variant, transcript):
    """Classify the effect of ``variant`` on ``transcript``.

    Runs fast first to detect splice-adjacent variants (which
    stay fast-classified); for everything else, builds a
    :class:`MutantTranscript` and diffs the translated protein.
    """
    from pyensembl import Transcript
    if not isinstance(transcript, Transcript):
        raise TypeError(
            "Expected %s : %s to have type Transcript" % (
                transcript, type(transcript)))

    if getattr(variant, "is_structural", False):
        return NotImplemented

    if not transcript.is_protein_coding:
        return NoncodingTranscript(variant, transcript)

    if not transcript.complete:
        return IncompleteTranscript(variant, transcript)

    # Run fast to get the splice classification.
    fast_effect = FastEffectAnnotator().annotate_on_transcript(
        variant, transcript)

    # Pure-intronic splice effects: fast only — no protein-
    # level diff to compute. SpliceDonor/SpliceAcceptor are
    # IntronicSpliceSite subclasses, so one check covers all.
    if isinstance(fast_effect, IntronicSpliceSite):
        return fast_effect

    # Non-splice location-based classes (UTR, deep intronic): fast's
    # position-based classification is authoritative. The protein may
    # be unchanged but "outside the CDS" carries location semantics a
    # whole-protein diff doesn't. Closes #318.
    if isinstance(fast_effect, (ThreePrimeUTR, FivePrimeUTR, Intronic)):
        return fast_effect

    # ExonicSpliceSite: dual-dispatch. Fast provides the
    # splice class; protein-diff provides the alternate_effect
    # via protein diff.
    if isinstance(fast_effect, ExonicSpliceSite):
        mt = apply_variant_to_transcript(variant, transcript)
        if mt is not None and mt.mutant_protein_sequence is not None:
            alt = classify_from_protein_diff(
                variant=variant,
                transcript=transcript,
                ref_protein=str(transcript.protein_sequence),
                mut_protein=mt.mutant_protein_sequence,
                length_delta=mt.total_length_delta,
                mutant_transcript=mt)
            return ExonicSpliceSite(
                variant=variant,
                transcript=transcript,
                exon=fast_effect.exon,
                alternate_effect=alt)
        return fast_effect

    # Non-splice: protein-diff slow path.
    mt = apply_variant_to_transcript(variant, transcript)
    if mt is None or mt.mutant_protein_sequence is None:
        # UTR, ref-mismatch, splice-junction-spanning, etc.
        return fast_effect

    ref_protein = str(transcript.protein_sequence)
    mut_protein = mt.mutant_protein_sequence

    # Alternate start codon rewrite: if the first codon changed to
    # another recognised start codon in the transcript's codon
    # table (e.g. ATG→CTG/GTG/TTG, or MT ATG→GTG under table 2),
    # the initiator tRNA still loads Met regardless of what the
    # codon would decode to internally. Rewrite the mutant
    # protein's first residue to 'M' so the shared diff classifier
    # sees the biologically correct protein. Closes #320.
    cds_start = min(transcript.start_codon_spliced_offsets)
    ref_first_codon = str(
        transcript.sequence)[cds_start:cds_start + 3]
    mutant_cds_start = _mutant_cds_start(transcript, mt.edits)
    mut_first_codon = mt.cdna_sequence[
        mutant_cds_start:mutant_cds_start + 3].upper()
    if (mut_first_codon != ref_first_codon
            and mut_protein
            and mut_protein[0] != "M"
            and ref_protein
            and ref_protein[0] == "M"):
        codon_table = codon_table_for_transcript(transcript)
        if mut_first_codon in codon_table.start_codons:
            mut_protein = "M" + mut_protein[1:]

    # Proteins match → Silent or AlternateStartCodon. Handle
    # both here because the shared classifier doesn't have
    # access to the cDNA edit offset for the correct aa_pos.
    if ref_protein == mut_protein:
        # An insertion immediately before the retained start can have a
        # CDS anchor on the minus strand, but only changes the 5' UTR.
        if mt.edits and all(e.cdna_end <= cds_start for e in mt.edits):
            return FivePrimeUTR(variant, transcript)
        if ref_first_codon != mut_first_codon:
            codon_table = codon_table_for_transcript(transcript)
            if mut_first_codon in codon_table.start_codons:
                return AlternateStartCodon(
                    variant=variant,
                    transcript=transcript,
                    ref_codon=ref_first_codon,
                    alt_codon=mut_first_codon)
        from ..effects.effect_classes import Silent
        edit = mt.edits[0] if mt.edits else None
        aa_pos = (edit.cdna_start - cds_start) // 3 if edit else 0
        aa_ref = (
            ref_protein[aa_pos]
            if 0 <= aa_pos < len(ref_protein)
            else "")
        return Silent(
            variant=variant,
            transcript=transcript,
            aa_pos=aa_pos,
            aa_ref=aa_ref)

    return classify_from_protein_diff(
        variant=variant,
        transcript=transcript,
        ref_protein=ref_protein,
        mut_protein=mut_protein,
        length_delta=mt.total_length_delta,
        mutant_transcript=mt)

Registry

varcode.register_annotator(annotator)

Add an annotator to the process-global registry, keyed by its .name. Re-registering under the same name overrides the previous entry — this is deliberate so callers can swap implementations in tests.

Source code in varcode/annotators/registry.py

def register_annotator(annotator):
    """Add an annotator to the process-global registry, keyed by its
    ``.name``. Re-registering under the same name overrides the
    previous entry — this is deliberate so callers can swap
    implementations in tests.
    """
    name = getattr(annotator, "name", None)
    if not name:
        raise ValueError(
            "Annotator %r has no .name attribute; cannot register." % annotator)
    _REGISTRY[name] = annotator
    return annotator

varcode.get_annotator(name)

Look up a registered annotator by name. Raises KeyError if no annotator is registered under that name.

Source code in varcode/annotators/registry.py

def get_annotator(name):
    """Look up a registered annotator by name. Raises ``KeyError``
    if no annotator is registered under that name.
    """
    return _REGISTRY[name]

varcode.get_default_annotator()

Return the annotator currently configured as the default.

Current default is "fast" (restored as the default in 7.0.0; see #397). "protein_diff" stays available as an opt-in.

Source code in varcode/annotators/registry.py

def get_default_annotator():
    """Return the annotator currently configured as the default.

    Current default is ``"fast"`` (restored as the default in 7.0.0;
    see #397). ``"protein_diff"`` stays available as an opt-in.
    """
    return _REGISTRY[_DEFAULT_NAME]

varcode.set_default_annotator(name)

Swap the process-wide default annotator. name must refer to a registered annotator.

Source code in varcode/annotators/registry.py

def set_default_annotator(name):
    """Swap the process-wide default annotator. ``name`` must refer
    to a registered annotator.
    """
    global _DEFAULT_NAME
    if name not in _REGISTRY:
        raise KeyError(
            "No annotator registered under %r — call register_annotator() "
            "first or pick from %r." % (name, sorted(_REGISTRY)))
    _DEFAULT_NAME = name

varcode.use_annotator(name_or_instance)

Context manager that temporarily swaps the default annotator.

Useful for A/B comparisons and scoped overrides without mutating global state across the codebase::

with varcode.use_annotator("protein_diff"):
    effects = variant_collection.effects()

Accepts the same argument shape as the annotator= kwarg: a registered-name string, or an annotator instance. Passing an instance registers it temporarily under its .name so that name-based lookups inside the block find it; on exit the previous default and any previously-registered annotator under that name are restored.

Source code in varcode/annotators/registry.py

@contextmanager
def use_annotator(name_or_instance):
    """Context manager that temporarily swaps the default annotator.

    Useful for A/B comparisons and scoped overrides without mutating
    global state across the codebase::

        with varcode.use_annotator("protein_diff"):
            effects = variant_collection.effects()

    Accepts the same argument shape as the ``annotator=`` kwarg:
    a registered-name string, or an annotator instance. Passing an
    instance registers it temporarily under its ``.name`` so that
    name-based lookups inside the block find it; on exit the
    previous default and any previously-registered annotator under
    that name are restored.
    """
    global _DEFAULT_NAME
    prior_default = _DEFAULT_NAME

    if isinstance(name_or_instance, str):
        if name_or_instance not in _REGISTRY:
            raise KeyError(
                "No annotator registered under %r — register one first or "
                "pass an instance." % name_or_instance)
        _DEFAULT_NAME = name_or_instance
        prior_registration = None
    else:
        name = getattr(name_or_instance, "name", None)
        if not name:
            raise ValueError(
                "Annotator instance has no .name attribute; cannot scope.")
        prior_registration = _REGISTRY.get(name)
        _REGISTRY[name] = name_or_instance
        _DEFAULT_NAME = name

    try:
        yield
    finally:
        _DEFAULT_NAME = prior_default
        if not isinstance(name_or_instance, str):
            if prior_registration is None:
                _REGISTRY.pop(name, None)
            else:
                _REGISTRY[name] = prior_registration

Experimental transcript model

This annotator is opt-in with annotator="transcript_model"; fast remains the default. The former realized name and public imports remain aliases. See supported inputs and results before using it.

varcode.TranscriptModelEffectAnnotator

Experimental, opt-in annotator backed by the transcript model.

annotate_with_context(variant, transcript, germline_ctx, phase_resolver=None)

Annotate through the same pipeline with patient germline edits.

Source code in varcode/transcript_model.py

def annotate_with_context(
        self, variant, transcript, germline_ctx, phase_resolver=None):
    """Annotate through the same pipeline with patient germline edits."""
    from .germline import Completeness, detect_loh

    if getattr(variant, "is_structural", False):
        start = variant.affected_start
        end = variant.affected_end
    else:
        start = variant.trimmed_base1_start
        end = variant.trimmed_base1_end
    start = max(transcript.start, start - 90)
    end = min(transcript.end, end + 90)
    germline = tuple(germline_ctx.variants_in_window(
        variant.contig, start, end))
    result = self._predict(
        variant, transcript,
        germline_variants=germline,
        phase_resolver=phase_resolver)
    if result is NotImplemented:
        return result
    if (not germline and germline_ctx.completeness in (
            Completeness.SPARSE, Completeness.HOTSPOTS_ONLY)):
        result.germline_unknown = True
    if detect_loh(variant, germline):
        result.is_loh = True
    return result

varcode.predict_transcript_model_effect(variants, transcript, germline_variants=(), phase_resolver=None, sequence_provider=None, max_hypotheses=64)

Predict one ordinary top effect with alternatives in .candidates.

Canonical and exon-skip paths resolve from transcript annotation alone. Intron retention and cryptic sites are also realized when genomic sequence is available, and remain explicit :class:Unresolved candidates at sequence-free tier 0. Rule order is never mislabeled as probability.

Source code in varcode/transcript_model.py

def predict_transcript_model_effect(
        variants, transcript, germline_variants=(), phase_resolver=None,
        sequence_provider=None, max_hypotheses=64):
    """Predict one ordinary top effect with alternatives in ``.candidates``.

    Canonical and exon-skip paths resolve from transcript annotation alone.
    Intron retention and cryptic sites are also realized when genomic
    sequence is available, and remain explicit :class:`Unresolved`
    candidates at sequence-free tier 0. Rule order is never mislabeled as
    probability.
    """
    variants = tuple(variants)
    if not variants:
        raise ValueError("predict_transcript_model_effect requires a somatic variant")
    germline_variants = tuple(germline_variants)
    primary = variants[0]
    if getattr(primary, "sv_type", None) == "BND":
        if len(variants) != 1 or germline_variants:
            return Unresolved(
                primary, transcript, mechanism="breakend_haplotype",
                reason=(
                    "BND composition with additional phased variants "
                    "requires an assembled allele"))
        # The established fusion builder already resolves partner transcript,
        # orientation, fused cDNA and translation from this same BND. Keep it
        # as the BND realization path while the layout engine handles local
        # span variants.
        from .effects.structural import predict_structural_variant_effect
        return predict_structural_variant_effect(primary, transcript)
    if not transcript.is_protein_coding:
        return NoncodingTranscript(primary, transcript)
    if not transcript.complete:
        return IncompleteTranscript(primary, transcript)
    if sequence_provider is None:
        sequence_provider = _provider_for_genome(primary.genome)

    phase_hypotheses = enumerate_phase_hypotheses(
        primary, germline_variants, phase_resolver=phase_resolver,
        max_hypotheses=max_hypotheses)
    outcomes = []
    enumeration_index = 0
    for phase_hypothesis in phase_hypotheses:
        reference = GenomicLayout.from_transcript(
            transcript, flank=50, sequence_provider=sequence_provider)
        baseline_layout = reference.apply_variants(phase_hypothesis.cis)
        mutant_layout = baseline_layout.apply_variants(variants)
        baseline_runs, baseline_statuses = _status_map(
            transcript, baseline_layout)
        mutant_runs, mutant_statuses = _status_map(transcript, mutant_layout)
        axes, shared_keys, baseline_fixed = _combined_mutant_axes(
            baseline_statuses, mutant_statuses, transcript,
            tuple(run.key for run in baseline_runs),
            tuple(run.key for run in mutant_runs))
        if axes:
            remaining = max_hypotheses - len(outcomes)
            if remaining < 1:
                raise ValueError(
                    "Combined phase/splice hypothesis count exceeds "
                    "max_hypotheses=%d" % max_hypotheses)
            try:
                plans = enumerate_splice_plans(
                    axes, tuple(run.key for run in mutant_runs),
                    max_plans=remaining)
            except ValueError as error:
                if "exceeds max_plans" not in str(error):
                    raise
                raise ValueError(
                    "Combined phase/splice hypothesis count exceeds "
                    "max_hypotheses=%d" % max_hypotheses) from error
        else:
            plans = (SplicePlan(
                choices=(),
                kept_runs=tuple(run.key for run in mutant_runs)),)

        for plan in plans:
            baseline_plan = _baseline_plan(
                transcript, baseline_runs, baseline_statuses, plan,
                shared_keys, baseline_fixed)
            baseline_product, baseline_unresolved = realize_splice_plan(
                transcript, baseline_layout, baseline_plan,
                tuple(baseline_statuses.values()))
            mutant_product, mutant_unresolved = realize_splice_plan(
                transcript, mutant_layout, plan,
                tuple(mutant_statuses.values()))
            hypothesis = EffectHypothesis(
                phase=_phase_items(phase_hypothesis),
                splice_plan=plan,
                phase_probability=_phase_probability(
                    phase_hypotheses, phase_hypothesis),
                evidence={
                    "phase_state": phase_hypothesis.phase_state,
                    "shared_splice_sites": tuple(sorted(shared_keys)),
                },
                enumeration_index=enumeration_index)
            unresolved = mutant_unresolved + baseline_unresolved
            if unresolved:
                mechanism = unresolved[0]
                result = Unresolved(
                    primary, transcript, mechanism=mechanism,
                    reason="genomic splice realization not available")
            else:
                result = classify_products(
                    primary, transcript, baseline_product, mutant_product)
            result.variants = variants
            result.splice_signal = tuple(sorted(mutant_statuses)) or None
            outcome = ClassifiedOutcome(
                hypothesis=hypothesis,
                effect=result,
                baseline=baseline_product,
                mutant=mutant_product,
                change_aa_offset=getattr(
                    result, "aa_mutation_start_offset", None),
                change_cdna_offset=_change_cdna_offset(primary, transcript))
            outcomes.append(outcome)
            enumeration_index += 1
            if len(outcomes) > max_hypotheses:
                raise ValueError(
                    "Combined phase/splice hypothesis count exceeds "
                    "max_hypotheses=%d" % max_hypotheses)

    candidates = merge_classified_outcomes(outcomes)
    ordered = order_realized_candidates(candidates, effect_priority)
    return _attach_candidate_set(ordered)