feat: Variant Level CSV Download

src/mavedb/lib/acmg.py

@@ -8,6 +8,48 @@
 from mavedb.models.enums.strength_of_evidence import StrengthOfEvidenceProvided
 
 
+def acmg_evidence_outcome_code(criterion: str, evidence_strength: Optional[str]) -> str:
+    """Build the ACMG 2015 evidence outcome code for a criterion and the strength it was met at.
+
+    Three rules, which are the ACMG convention rather than anything MaveDB invented:
+
+    - no strength means the criterion was evaluated and *not* met, written ``"PS3_not_met"``
+    - STRONG is the criterion's baseline, so it is written bare: ``"PS3"``
+    - any other strength is suffixed: ``"PS3_moderate"``
+
+    Takes the criterion code and strength *name* as strings rather than enums so that the VA-Spec
+    annotation builders and the flat exports can share one implementation despite drawing their
+    enumerations from different places. That also means this survives any future decision about which
+    enumeration is canonical.
+
+    Parameters
+    ----------
+    criterion : str
+        The criterion code, e.g. ``"PS3"`` or ``"BS3"``.
+    evidence_strength : Optional[str]
+        The strength name, e.g. ``"MODERATE"``. None when the criterion was not met.
+
+    Returns
+    -------
+    str
+        The evidence outcome code.
+
+    Examples
+    --------
+    >>> acmg_evidence_outcome_code("PS3", "STRONG")
+    'PS3'
+    >>> acmg_evidence_outcome_code("PS3", "MODERATE")
+    'PS3_moderate'
+    >>> acmg_evidence_outcome_code("BS3", None)
+    'BS3_not_met'
+    """
+    if evidence_strength is None:
+        return f"{criterion}_not_met"
+    if evidence_strength.upper() == StrengthOfEvidenceProvided.STRONG.name:
+        return criterion
+    return f"{criterion}_{evidence_strength.lower()}"
+
+
 def points_evidence_strength_equivalent(
     points: int,
 ) -> tuple[Optional[ACMGCriterion], Optional[StrengthOfEvidenceProvided]]:

src/mavedb/lib/annotation/classification.py

@@ -21,13 +21,35 @@ class ExperimentalVariantFunctionalImpactClassification(StrEnum):
     INDETERMINATE = "indeterminate"
 
 
+def _classification_contains_variant(
+    functional_classification: ScoreCalibrationFunctionalClassification,
+    mapped_variant: MappedVariant,
+    containing_classification_ids: Optional[set[int]],
+) -> bool:
+    """Whether this classification's score range contains the variant.
+
+    Prefers a pre-resolved id set, which is an O(1) check. Falls back to the ORM relationship, which is
+    correct but loads every variant of the range.
+    """
+    if containing_classification_ids is not None:
+        return functional_classification.id in containing_classification_ids
+    return mapped_variant.variant in functional_classification.variants
+
+
 def functional_classification_of_variant(
-    mapped_variant: MappedVariant, score_calibration: ScoreCalibration
+    mapped_variant: MappedVariant,
+    score_calibration: ScoreCalibration,
+    containing_classification_ids: Optional[set[int]] = None,
 ) -> tuple[Optional[ScoreCalibrationFunctionalClassification], ExperimentalVariantFunctionalImpactClassification]:
     """Classify a variant's functional impact as normal, abnormal, or indeterminate.
 
     Uses the primary score calibration and its functional ranges.
     Raises ValueError if required calibration or score is missing.
+
+    *containing_classification_ids*, when given, is the set of functional-classification ids already known
+    to contain this variant. Pass it to avoid the ORM membership check below, which loads every variant of
+    every range. A caller classifying many variants should resolve membership once from the association
+    table; see ``mavedb.lib.csv.variant``.
     """
     if not mapped_variant.variant.score_set.score_calibrations:
         raise ValueError(
@@ -41,11 +63,8 @@ def functional_classification_of_variant(
             " Unable to classify functional impact."
         )
 
-    # TODO#XXX: Performance: avoid ORM relationship membership checks (`variant in functional_range.variants`) in this
-    #           DB-agnostic function. Resolve class-based matches in an upstream DB-aware layer using the association table,
-    #           pass matched functional classification IDs into this function, and use O(1) ID membership checks here.
     for functional_range in score_calibration.functional_classifications:
-        if mapped_variant.variant in functional_range.variants:
+        if _classification_contains_variant(functional_range, mapped_variant, containing_classification_ids):
             if functional_range.functional_classification is FunctionalClassificationOptions.normal:
                 return functional_range, ExperimentalVariantFunctionalImpactClassification.NORMAL
             elif functional_range.functional_classification is FunctionalClassificationOptions.abnormal:
@@ -58,6 +77,7 @@ def functional_classification_of_variant(
 def pathogenicity_classification_of_variant(
     mapped_variant: MappedVariant,
     score_calibration: ScoreCalibration,
+    containing_classification_ids: Optional[set[int]] = None,
 ) -> tuple[
     Optional[ScoreCalibrationFunctionalClassification],
     VariantPathogenicityEvidenceLine.Criterion,
@@ -87,11 +107,8 @@ def pathogenicity_classification_of_variant(
             " Unable to classify clinical impact."
         )
 
-    # TODO#XXX: Performance: avoid ORM relationship membership checks (`variant in pathogenicity_range.variants`) in this
-    #           DB-agnostic function. Resolve class-based matches in an upstream DB-aware layer using the association table,
-    #           pass matched functional classification IDs into this function, and use O(1) ID membership checks here.
     for pathogenicity_range in score_calibration.functional_classifications:
-        if mapped_variant.variant in pathogenicity_range.variants:
+        if _classification_contains_variant(pathogenicity_range, mapped_variant, containing_classification_ids):
             if pathogenicity_range.acmg_classification is None:
                 return (pathogenicity_range, VariantPathogenicityEvidenceLine.Criterion.PS3, None)
 

src/mavedb/lib/annotation/evidence_line.py

@@ -10,8 +10,8 @@
     StudyResult,
     VariantPathogenicityProposition,
 )
-from ga4gh.va_spec.base.enums import StrengthOfEvidenceProvided
 
+from mavedb.lib.acmg import acmg_evidence_outcome_code
 from mavedb.lib.annotation.classification import (
     functional_classification_of_variant,
     pathogenicity_classification_of_variant,
@@ -44,16 +44,14 @@ def acmg_evidence_line(
         mapped_variant, score_calibration
     )
 
+    evidence_outcome_code = acmg_evidence_outcome_code(
+        evidence_outcome.value, evidence_strength.name if evidence_strength else None
+    )
+
     if not evidence_strength:
-        evidence_outcome_code = f"{evidence_outcome.value}_not_met"
         strength_of_evidence = None
         direction_of_evidence = Direction.NEUTRAL
     else:
-        evidence_outcome_code = (
-            f"{evidence_outcome.value}_{evidence_strength.name.lower()}"
-            if evidence_strength != StrengthOfEvidenceProvided.STRONG
-            else evidence_outcome.value
-        )
         strength_of_evidence = MappableConcept(
             primaryCoding=Coding(
                 code=evidence_strength,

src/mavedb/lib/annotation/flatten.py

@@ -0,0 +1,140 @@
+"""Flatten a variant's VA-Spec clinical interpretation into scalar values.
+
+The rest of this package builds nested VA-Spec structures, which are faithful to the standard but not
+consumable by a spreadsheet. This projects the same classification onto flat scalars.
+"""
+
+from dataclasses import dataclass
+from typing import Optional
+
+from ga4gh.va_spec.acmg_2015 import AcmgClassification
+
+from mavedb.lib.acmg import acmg_evidence_outcome_code
+from mavedb.lib.annotation.classification import (
+    functional_classification_of_variant,
+    pathogenicity_classification_of_variant,
+)
+from mavedb.models.mapped_variant import MappedVariant
+from mavedb.models.score_calibration import ScoreCalibration
+
+
+@dataclass(frozen=True)
+class FlatAnnotation:
+    """A single variant's clinical interpretation under one calibration, flattened to scalars.
+
+    A field is ``None`` when the calibration cannot support it; exporters render that as NA.
+    """
+
+    functional_classification: Optional[str] = None
+    """"normal", "abnormal", or "indeterminate"."""
+
+    acmg_criterion: Optional[str] = None
+    """The ACMG 2015 criterion evaluated, e.g. "PS3" or "BS3"."""
+
+    acmg_evidence_strength: Optional[str] = None
+    """Strength of evidence, e.g. "MODERATE" or "MODERATE_PLUS"; None when the criterion was not met.
+
+    MaveDB's own enumeration, finer-grained than VA-Spec's: an M+ range reports MODERATE_PLUS here while
+    its VA-Spec annotation must report moderate.
+    """
+
+    acmg_evidence_outcome_code: Optional[str] = None
+    """ACMG evidence outcome code, e.g. "PS3_moderate", "PS3" (strong), or "BS3_not_met"."""
+
+    pathogenicity_classification: Optional[str] = None
+    """"PATHOGENIC", "BENIGN", or "UNCERTAIN_SIGNIFICANCE"."""
+
+    calibration_urn: Optional[str] = None
+    calibration_title: Optional[str] = None
+
+    research_use_only: Optional[bool] = None
+    """Whether the calibration is marked research use only.
+
+    Carried so a consumer holding only exported rows can tell that a criterion came from thresholds never
+    validated for clinical use.
+    """
+
+
+def flatten_annotation(
+    mapped_variant: MappedVariant,
+    score_calibration: Optional[ScoreCalibration],
+    containing_classification_ids: Optional[set[int]] = None,
+) -> FlatAnnotation:
+    """Flatten a variant's clinical interpretation under *score_calibration* into scalar values.
+
+    A calibration with ranges but no ACMG classifications yields a functional classification only, matching
+    the annotation layer. Evidence strength uses MaveDB's own enumeration, so MODERATE_PLUS is preserved.
+
+    Args:
+        containing_classification_ids: forwarded to the classifiers. Supply it when flattening many
+            variants; the fallback loads every variant of every score range.
+
+    Returns:
+        An all-``None`` annotation when *score_calibration* is None. A calibration that exists but defines
+        no ranges reports its identity and standing with no interpretation.
+    """
+    # No calibration means nothing to say under this namespace (e.g. it belongs to another score set).
+    if score_calibration is None:
+        return FlatAnnotation()
+
+    # Rangeless: reporting identity distinguishes "defines no ranges" from "no calibration applies here".
+    if not score_calibration.functional_classifications:
+        return FlatAnnotation(
+            calibration_urn=score_calibration.urn,
+            calibration_title=score_calibration.title,
+            research_use_only=bool(score_calibration.research_use_only),
+        )
+
+    _, functional_classification = functional_classification_of_variant(
+        mapped_variant, score_calibration, containing_classification_ids
+    )
+
+    functional_only = FlatAnnotation(
+        functional_classification=functional_classification.value,
+        calibration_urn=score_calibration.urn,
+        calibration_title=score_calibration.title,
+        research_use_only=bool(score_calibration.research_use_only),
+    )
+
+    # No ACMG classification on any range: stop at the functional classification rather than reporting a
+    # not-met PS3 the curator never asserted.
+    if all(fc.acmg_classification is None for fc in score_calibration.functional_classifications):
+        return functional_only
+
+    # VA-Spec strength deliberately unused: it has already collapsed MODERATE_PLUS to MODERATE.
+    containing_range, criterion, _va_spec_evidence_strength = pathogenicity_classification_of_variant(
+        mapped_variant, score_calibration, containing_classification_ids
+    )
+
+    # Read the strength off the containing range, which keeps MODERATE_PLUS. Reachable in practice:
+    # `points_evidence_strength_equivalent` assigns M+ to +/-3 point ranges (e.g. the Excalibr
+    # calibrations).
+    #
+    # TODO(#XXX): move the lossy MODERATE_PLUS -> MODERATE conversion to the VA-Spec boundary instead of
+    # `classification.py`, upstream of every consumer; this special case then disappears.
+    native_evidence_strength = (
+        containing_range.acmg_classification.evidence_strength
+        if containing_range is not None and containing_range.acmg_classification is not None
+        else None
+    )
+    evidence_strength_name = native_evidence_strength.name if native_evidence_strength is not None else None
+
+    # `pathogenicity_classification_of_variant` returns PS3 even for variants in no range, so the range,
+    # not the criterion, tells us whether evidence exists. No strength means evaluated and not met.
+    if containing_range is None or evidence_strength_name is None:
+        pathogenicity_classification = AcmgClassification.UNCERTAIN_SIGNIFICANCE
+    elif criterion.name.startswith("B"):
+        pathogenicity_classification = AcmgClassification.BENIGN
+    else:
+        pathogenicity_classification = AcmgClassification.PATHOGENIC
+
+    return FlatAnnotation(
+        functional_classification=functional_only.functional_classification,
+        acmg_criterion=criterion.value,
+        acmg_evidence_strength=evidence_strength_name,
+        acmg_evidence_outcome_code=acmg_evidence_outcome_code(criterion.value, evidence_strength_name),
+        pathogenicity_classification=pathogenicity_classification.name,
+        calibration_urn=functional_only.calibration_urn,
+        calibration_title=functional_only.calibration_title,
+        research_use_only=functional_only.research_use_only,
+    )