Skip to content
Merged
101 changes: 67 additions & 34 deletions src/mavedb/lib/annotation/annotate.py
Original file line number Diff line number Diff line change
Expand Up @@ -8,14 +8,14 @@
See: https://va-spec.ga4gh.org/en/latest/va-standard-profiles/community-profiles/acmg-2015-profiles.html#variant-pathogenicity-statement-acmg-2015
"""

from typing import Optional, Union
from typing import Optional, Sequence, TypeVar, Union

from ga4gh.va_spec.acmg_2015 import VariantPathogenicityStatement
from ga4gh.va_spec.base.core import ExperimentalVariantFunctionalImpactStudyResult, Statement

from mavedb.lib.annotation.classification import functional_classification_of_variant
from mavedb.lib.annotation.exceptions import MappingDataDoesntExistException
from mavedb.lib.annotation.evidence_line import acmg_evidence_line, functional_evidence_line
from mavedb.lib.annotation.exceptions import MappingDataDoesntExistException
from mavedb.lib.annotation.proposition import (
mapped_variant_to_experimental_variant_clinical_impact_proposition,
mapped_variant_to_experimental_variant_functional_impact_proposition,
Expand All @@ -26,39 +26,62 @@
)
from mavedb.lib.annotation.study_result import mapped_variant_to_experimental_variant_impact_study_result
from mavedb.lib.annotation.util import (
calibration_scope_extension,
calibrations_available_for_annotation,
can_annotate_variant_for_functional_statement,
can_annotate_variant_for_pathogenicity_evidence,
score_calibration_may_be_used_for_annotation,
select_strongest_functional_calibration,
select_strongest_pathogenicity_calibration,
)
from mavedb.lib.permissions.principal import Principal
from mavedb.models.mapped_variant import MappedVariant
from mavedb.models.score_calibration import ScoreCalibration

Annotation = TypeVar(
"Annotation", ExperimentalVariantFunctionalImpactStudyResult, Statement, VariantPathogenicityStatement
)


def _disclosing_calibration_scope(annotation: Annotation, calibrations: Sequence[ScoreCalibration]) -> Annotation:
"""Record on the annotation which principal it was built for.

Applied at the top-level entry points only. Nested study results and statements built as components of
an evidence line inherit the scope of the object that contains them.
"""
# model_copy rather than assigning to `.extensions`: mypy resolves the field's element type to a
# `ga4gh.va_spec.base.core.Extension` that does not exist at runtime (the ga4gh namespace packages
# confuse its import resolution), so a direct assignment is a false positive.
return annotation.model_copy(
update={"extensions": [*(annotation.extensions or []), calibration_scope_extension(calibrations)]}
)


def variant_study_result(mapped_variant: MappedVariant) -> ExperimentalVariantFunctionalImpactStudyResult:
return mapped_variant_to_experimental_variant_impact_study_result(mapped_variant)
# A study result reports the measured score and carries no calibration-derived evidence, so its scope
# is public regardless of viewer. Disclosed anyway, so that a missing scope never has to be read as
# "public" or "generated before disclosure existed".
return _disclosing_calibration_scope(mapped_variant_to_experimental_variant_impact_study_result(mapped_variant), [])


def variant_functional_impact_statement(
mapped_variant: MappedVariant, allow_research_use_only_calibrations: bool = False
mapped_variant: MappedVariant,
allow_research_use_only_calibrations: bool = False,
principal: Optional[Principal] = None,
) -> Optional[Statement]:
if not can_annotate_variant_for_functional_statement(
mapped_variant, allow_research_use_only_calibrations=allow_research_use_only_calibrations
mapped_variant, allow_research_use_only_calibrations=allow_research_use_only_calibrations, principal=principal
):
return None

study_result = mapped_variant_to_experimental_variant_impact_study_result(mapped_variant)
functional_proposition = mapped_variant_to_experimental_variant_functional_impact_proposition(mapped_variant)

# Collect eligible calibrations
eligible_calibrations = []
for score_calibration in mapped_variant.variant.score_set.score_calibrations:
if score_calibration_may_be_used_for_annotation(
score_calibration,
annotation_type="functional",
allow_research_use_only_calibrations=allow_research_use_only_calibrations,
):
eligible_calibrations.append(score_calibration)
eligible_calibrations = calibrations_available_for_annotation(
mapped_variant,
"functional",
allow_research_use_only_calibrations=allow_research_use_only_calibrations,
principal=principal,
)

# Select the calibration with the strongest evidence
strongest_calibration, strongest_range = select_strongest_functional_calibration(
Expand All @@ -77,32 +100,34 @@ def variant_functional_impact_statement(
for score_calibration in eligible_calibrations:
functional_evidence.append(functional_evidence_line(mapped_variant, score_calibration, [study_result]))

return mapped_variant_to_functional_statement(
mapped_variant, functional_proposition, functional_evidence, strongest_calibration, classification
return _disclosing_calibration_scope(
mapped_variant_to_functional_statement(
mapped_variant, functional_proposition, functional_evidence, strongest_calibration, classification
),
eligible_calibrations,
)


def variant_pathogenicity_statement(
mapped_variant: MappedVariant, allow_research_use_only_calibrations: bool = False
mapped_variant: MappedVariant,
allow_research_use_only_calibrations: bool = False,
principal: Optional[Principal] = None,
) -> Optional[VariantPathogenicityStatement]:
if not can_annotate_variant_for_pathogenicity_evidence(
mapped_variant, allow_research_use_only_calibrations=allow_research_use_only_calibrations
mapped_variant, allow_research_use_only_calibrations=allow_research_use_only_calibrations, principal=principal
):
return None

study_result = mapped_variant_to_experimental_variant_impact_study_result(mapped_variant)
functional_proposition = mapped_variant_to_experimental_variant_functional_impact_proposition(mapped_variant)
clinical_proposition = mapped_variant_to_experimental_variant_clinical_impact_proposition(mapped_variant)

# Collect eligible calibrations
eligible_calibrations = []
for score_calibration in mapped_variant.variant.score_set.score_calibrations:
if score_calibration_may_be_used_for_annotation(
score_calibration,
annotation_type="pathogenicity",
allow_research_use_only_calibrations=allow_research_use_only_calibrations,
):
eligible_calibrations.append(score_calibration)
eligible_calibrations = calibrations_available_for_annotation(
mapped_variant,
"pathogenicity",
allow_research_use_only_calibrations=allow_research_use_only_calibrations,
principal=principal,
)

# Select the calibration with the strongest evidence
strongest_calibration, strongest_range = select_strongest_pathogenicity_calibration(
Expand Down Expand Up @@ -130,25 +155,33 @@ def variant_pathogenicity_statement(
acmg_evidence_line(mapped_variant, score_calibration, clinical_proposition, [functional_statement])
)

return mapped_variant_to_pathogenicity_statement(
mapped_variant, clinical_proposition, clinical_evidence, strongest_calibration, strongest_range
return _disclosing_calibration_scope(
mapped_variant_to_pathogenicity_statement(
mapped_variant, clinical_proposition, clinical_evidence, strongest_calibration, strongest_range
),
eligible_calibrations,
)


def variant_highest_level_annotation(
mapped_variant: MappedVariant,
principal: Optional[Principal] = None,
) -> Optional[Union[ExperimentalVariantFunctionalImpactStudyResult, Statement, VariantPathogenicityStatement]]:
"""
Build the single highest-materialized VA-Spec layer for a mapped variant.

Layer ladder (highest to lowest): pathogenicity statement -> functional impact statement -> study result.
Returns None when the variant has no post-mapped allele and therefore cannot be annotated.

The viewer decides which layer is reachable as well as what the layer contains: a variant whose only
calibration is invisible to this principal degrades to a study result rather than yielding a statement
with nothing in it.
"""
try:
if can_annotate_variant_for_pathogenicity_evidence(mapped_variant):
return variant_pathogenicity_statement(mapped_variant)
if can_annotate_variant_for_functional_statement(mapped_variant):
return variant_functional_impact_statement(mapped_variant)
if can_annotate_variant_for_pathogenicity_evidence(mapped_variant, principal=principal):
return variant_pathogenicity_statement(mapped_variant, principal=principal)
if can_annotate_variant_for_functional_statement(mapped_variant, principal=principal):
return variant_functional_impact_statement(mapped_variant, principal=principal)
return variant_study_result(mapped_variant)
except MappingDataDoesntExistException:
return None
123 changes: 95 additions & 28 deletions src/mavedb/lib/annotation/util.py
Original file line number Diff line number Diff line change
@@ -1,4 +1,4 @@
from typing import Literal, Optional
from typing import Iterable, Literal, Optional

from ga4gh.core.models import Extension
from ga4gh.va_spec.base.enums import StrengthOfEvidenceProvided as VaSpecStrengthOfEvidenceProvided
Expand All @@ -21,12 +21,17 @@
)
from mavedb.lib.annotation.exceptions import MappingDataDoesntExistException
from mavedb.lib.mapping import extract_ids_from_post_mapped_metadata
from mavedb.lib.permissions.principal import Principal
from mavedb.lib.permissions.score_calibration import ScoreCalibrationViewer
from mavedb.lib.types.annotation import SequenceFeature
from mavedb.lib.variants import target_for_variant
from mavedb.models.mapped_variant import MappedVariant
from mavedb.models.score_calibration import ScoreCalibration
from mavedb.models.score_calibration_functional_classification import ScoreCalibrationFunctionalClassification

CALIBRATION_SCOPE_EXTENSION_NAME = "mavedb_calibration_scope"
"""Extension naming the principal an annotation was built for. See ``calibration_scope_extension``."""


def allele_from_mapped_variant_dictionary_result(allelic_mapping_results: dict) -> Allele:
"""
Expand Down Expand Up @@ -230,40 +235,84 @@ def score_calibration_may_be_used_for_annotation(
return True


def _variant_score_calibrations_have_required_calibrations_and_ranges_for_annotation(
def calibrations_available_for_annotation(
mapped_variant: MappedVariant,
annotation_type: Literal["pathogenicity", "functional"],
allow_research_use_only_calibrations: bool = False,
) -> bool:
principal: Optional[Principal] = None,
) -> list[ScoreCalibration]:
"""
Check if a mapped variant's score set contains any of the required calibrations for annotation.
Select the calibrations on a mapped variant's score set that may build the requested annotation.

Two independent questions decide this, and they are asked by different collaborators.
- Eligibility: Does this calibration carry the classifications the annotation type needs, and is
its research-use-only standing permitted here. This is ``score_calibration_may_be_used_for_annotation``.
- Visibility: May this principal read it at all. This is the viewer's, because a calibration's READ rule
is stricter than its score set's.

Args:
mapped_variant (MappedVariant): The mapped variant object containing the variant with score set data.
annotation_type (Literal["pathogenicity", "functional"]): The type of annotation to check for.
Must be either "pathogenicity" or "functional".
allow_research_use_only_calibrations (bool, optional): Whether to consider calibrations marked as
research use only as valid for annotation. Defaults to False.
mapped_variant (MappedVariant): The mapped variant whose score set's calibrations are considered.
annotation_type (Literal["pathogenicity", "functional"]): The type of annotation to be built.
allow_research_use_only_calibrations (bool, optional): Whether calibrations marked research use
only are eligible. Defaults to False.
principal (Optional[Principal], optional): The caller being served. Defaults to None, an anonymous
caller, so a function that omits it gets public calibrations only.

Returns:
bool: True if the variant's score set contains at least one valid calibration with the required
classifications for the specified annotation type. False otherwise.
list[ScoreCalibration]: The eligible, visible calibrations, in score set order.
"""
if mapped_variant.variant.score_set.score_calibrations is None:
return False
viewer = (principal if principal is not None else Principal()).viewer_for(ScoreCalibrationViewer)

return any(
score_calibration_may_be_used_for_annotation(
return [
score_calibration
for score_calibration in viewer.visible(mapped_variant.variant.score_set.score_calibrations)
if score_calibration_may_be_used_for_annotation(
score_calibration,
annotation_type,
allow_research_use_only_calibrations=allow_research_use_only_calibrations,
)
for score_calibration in mapped_variant.variant.score_set.score_calibrations
]


def calibration_scope_extension(calibrations: Iterable[ScoreCalibration]) -> Extension:
"""
Describe the principal an annotation was built for, given the calibrations behind it.

VA-Spec statements carry no stable identifier, so two callers can receive materially different
statements from the same URL. Naming the scope on the object itself is what keeps that honest: a
consumer holding a record can tell whether it is the one anyone would get, or one widened by the
requester's own access.

Args:
calibrations (Iterable[ScoreCalibration]): The calibrations contributing evidence to the annotation.

Returns:
Extension: A ``mavedb_calibration_scope`` extension, ``restricted`` when any contributing
calibration is private and ``public`` otherwise.
"""
if any(calibration.private for calibration in calibrations):
return Extension(
name=CALIBRATION_SCOPE_EXTENSION_NAME,
value="restricted",
description=(
"Built from at least one private score calibration, visible to the requesting viewer. "
"Another viewer requesting this variant may receive fewer evidence lines, or none."
),
)

return Extension(
name=CALIBRATION_SCOPE_EXTENSION_NAME,
value="public",
description=(
"Built only from public score calibrations. Any viewer requesting this variant receives the same evidence."
),
)


def can_annotate_variant_for_pathogenicity_evidence(
mapped_variant: MappedVariant, allow_research_use_only_calibrations=False
mapped_variant: MappedVariant,
allow_research_use_only_calibrations=False,
principal: Optional[Principal] = None,
) -> bool:
"""
Determine if a mapped variant can be annotated for pathogenicity evidence.
Expand All @@ -275,6 +324,11 @@ def can_annotate_variant_for_pathogenicity_evidence(
Args:
mapped_variant (MappedVariant): The mapped variant object to evaluate
for pathogenicity evidence annotation eligibility.
allow_research_use_only_calibrations (bool, optional): Whether calibrations marked research use
only are eligible. Defaults to False.
principal (Optional[Principal], optional): The caller being served. Defaults to None, an anonymous
caller. Must match the principal the annotation itself will be built for, or this answers a
different question than the one the caller is about to act on.

Returns:
bool: True if the variant can be annotated for pathogenicity evidence,
Expand All @@ -290,16 +344,21 @@ def can_annotate_variant_for_pathogenicity_evidence(
"""
if not _can_annotate_variant_base_assumptions(mapped_variant):
return False
if not _variant_score_calibrations_have_required_calibrations_and_ranges_for_annotation(
mapped_variant, "pathogenicity", allow_research_use_only_calibrations=allow_research_use_only_calibrations
):
return False

return True
return bool(
calibrations_available_for_annotation(
mapped_variant,
"pathogenicity",
allow_research_use_only_calibrations=allow_research_use_only_calibrations,
principal=principal,
)
)


def can_annotate_variant_for_functional_statement(
mapped_variant: MappedVariant, allow_research_use_only_calibrations=False
mapped_variant: MappedVariant,
allow_research_use_only_calibrations=False,
principal: Optional[Principal] = None,
) -> bool:
"""
Determine if a mapped variant can be annotated for functional statements.
Expand All @@ -311,6 +370,11 @@ def can_annotate_variant_for_functional_statement(
Args:
mapped_variant (MappedVariant): The variant object to check for annotation
eligibility, containing mapping information and score data.
allow_research_use_only_calibrations (bool, optional): Whether calibrations marked research use
only are eligible. Defaults to False.
principal (Optional[Principal], optional): The caller being served. Defaults to None, an anonymous
caller. Must match the principal the annotation itself will be built for, or this answers a
different question than the one the caller is about to act on.

Returns:
bool: True if the variant can be annotated for functional statements,
Expand All @@ -323,12 +387,15 @@ def can_annotate_variant_for_functional_statement(
"""
if not _can_annotate_variant_base_assumptions(mapped_variant):
return False
if not _variant_score_calibrations_have_required_calibrations_and_ranges_for_annotation(
mapped_variant, "functional", allow_research_use_only_calibrations=allow_research_use_only_calibrations
):
return False

return True
return bool(
calibrations_available_for_annotation(
mapped_variant,
"functional",
allow_research_use_only_calibrations=allow_research_use_only_calibrations,
principal=principal,
)
)


def sequence_feature_for_mapped_variant(mapped_variant: MappedVariant) -> SequenceFeature:
Expand Down
Loading
Loading