Calibration Control Support

alembic/versions/024370c4bca7_add_calibration_controls_table_and_phi_.py

@@ -0,0 +1,113 @@
+"""add calibration controls, mondo terms, and phi/disease columns
+
+Revision ID: 024370c4bca7
+Revises: a7f3c2e9b104
+Create Date: 2026-09-15 16:01:06.013964
+
+"""
+
+import sqlalchemy as sa
+
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision = "024370c4bca7"
+down_revision = "a7f3c2e9b104"
+branch_labels = None
+depends_on = None
+
+# The generic "disease or disorder" MONDO term seeded so the non-nullable calibration FK always resolves.
+MONDO_SYSTEM = "https://purl.obolibrary.org/obo/mondo.owl"
+MONDO_GENERIC_CODE = "MONDO:0000001"
+MONDO_GENERIC_LABEL = "disease or disorder"
+
+
+def upgrade():
+    op.create_table(
+        "calibration_controls",
+        sa.Column("id", sa.Integer(), nullable=False),
+        sa.Column("calibration_id", sa.Integer(), nullable=False),
+        sa.Column("variant_id", sa.Integer(), nullable=False),
+        sa.Column(
+            "clinical_status",
+            sa.Enum("pathogenic", "benign", name="calibrationcontrolstatus", native_enum=False, length=32),
+            nullable=False,
+        ),
+        sa.Column("created_by_id", sa.Integer(), nullable=False),
+        sa.Column("modified_by_id", sa.Integer(), nullable=False),
+        sa.Column("creation_date", sa.Date(), nullable=False),
+        sa.Column("modification_date", sa.Date(), nullable=False),
+        sa.ForeignKeyConstraint(["calibration_id"], ["score_calibrations.id"], ondelete="CASCADE"),
+        sa.ForeignKeyConstraint(["variant_id"], ["variants.id"]),
+        sa.ForeignKeyConstraint(["created_by_id"], ["users.id"]),
+        sa.ForeignKeyConstraint(["modified_by_id"], ["users.id"]),
+        sa.PrimaryKeyConstraint("id"),
+        sa.UniqueConstraint("calibration_id", "variant_id", name="uq_calibration_controls_calibration_id_variant_id"),
+    )
+    op.create_index(
+        op.f("ix_calibration_controls_calibration_id"), "calibration_controls", ["calibration_id"], unique=False
+    )
+    op.create_index(op.f("ix_calibration_controls_variant_id"), "calibration_controls", ["variant_id"], unique=False)
+    op.create_index(
+        op.f("ix_calibration_controls_created_by_id"), "calibration_controls", ["created_by_id"], unique=False
+    )
+    op.create_index(
+        op.f("ix_calibration_controls_modified_by_id"), "calibration_controls", ["modified_by_id"], unique=False
+    )
+
+    # MONDO disease terms, with the generic root seeded as the default for calibrations naming no disease.
+    op.create_table(
+        "mondo_terms",
+        sa.Column("id", sa.Integer(), nullable=False),
+        sa.Column("code", sa.String(), nullable=False),
+        sa.Column("system", sa.String(), nullable=False),
+        sa.Column("system_version", sa.String(), nullable=True),
+        sa.Column("label", sa.String(), nullable=False),
+        sa.Column("creation_date", sa.Date(), nullable=False),
+        sa.Column("modification_date", sa.Date(), nullable=False),
+        sa.PrimaryKeyConstraint("id"),
+        sa.UniqueConstraint("system", "code", name="uq_mondo_terms_system_code"),
+    )
+    op.execute(
+        sa.text(
+            "INSERT INTO mondo_terms (code, system, label, creation_date, modification_date) "
+            "VALUES (:code, :system, :label, CURRENT_DATE, CURRENT_DATE)"
+        ).bindparams(code=MONDO_GENERIC_CODE, system=MONDO_SYSTEM, label=MONDO_GENERIC_LABEL)
+    )
+
+    # Disease FK: add nullable, backfill existing calibrations to the generic term, then enforce NOT NULL.
+    op.add_column("score_calibrations", sa.Column("disease_term_id", sa.Integer(), nullable=True))
+    op.execute(
+        sa.text(
+            "UPDATE score_calibrations SET disease_term_id = "
+            "(SELECT id FROM mondo_terms WHERE system = :system AND code = :code)"
+        ).bindparams(system=MONDO_SYSTEM, code=MONDO_GENERIC_CODE)
+    )
+    op.alter_column("score_calibrations", "disease_term_id", nullable=False)
+    op.create_index(
+        op.f("ix_score_calibrations_disease_term_id"), "score_calibrations", ["disease_term_id"], unique=False
+    )
+    op.create_foreign_key(
+        "fk_score_calibrations_disease_term_id_mondo_terms",
+        "score_calibrations",
+        "mondo_terms",
+        ["disease_term_id"],
+        ["id"],
+    )
+
+    op.add_column("score_calibrations", sa.Column("controls_not_phi", sa.Boolean(), nullable=True))
+
+
+def downgrade():
+    op.drop_column("score_calibrations", "controls_not_phi")
+
+    op.drop_constraint("fk_score_calibrations_disease_term_id_mondo_terms", "score_calibrations", type_="foreignkey")
+    op.drop_index(op.f("ix_score_calibrations_disease_term_id"), table_name="score_calibrations")
+    op.drop_column("score_calibrations", "disease_term_id")
+    op.drop_table("mondo_terms")
+
+    op.drop_index(op.f("ix_calibration_controls_modified_by_id"), table_name="calibration_controls")
+    op.drop_index(op.f("ix_calibration_controls_created_by_id"), table_name="calibration_controls")
+    op.drop_index(op.f("ix_calibration_controls_variant_id"), table_name="calibration_controls")
+    op.drop_index(op.f("ix_calibration_controls_calibration_id"), table_name="calibration_controls")
+    op.drop_table("calibration_controls")

src/mavedb/lib/annotation/annotate.py

@@ -120,7 +120,6 @@ def variant_pathogenicity_statement(
 
     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)
 
     eligible_calibrations = calibrations_available_for_annotation(
         mapped_variant,
@@ -137,6 +136,17 @@ def variant_pathogenicity_statement(
     if not strongest_calibration:
         return None
 
+    # The statement carries one proposition (targeted by every evidence line), so its disease context
+    # is the strongest calibration's — the same calibration that anchors the ACMG classification.
+    #
+    # TODO#XXX - Pooling calibrations with different disease contexts is not supported by the current VA-Spec model. If a
+    # variant has multiple calibrations with different disease contexts, the strongest calibration is used for the
+    # statement's disease context and we should consider whether to filter out calibrations with different disease contexts
+    # from the evidence lines.
+    clinical_proposition = mapped_variant_to_experimental_variant_clinical_impact_proposition(
+        mapped_variant, strongest_calibration
+    )
+
     # Get the classification from the strongest range (used for the functional statement within clinical evidence)
     # If strongest_range is None, the variant is not in any range, so classification will be INDETERMINATE
     _, classification = functional_classification_of_variant(mapped_variant, strongest_calibration)

src/mavedb/lib/annotation/condition.py

@@ -1,21 +1,13 @@
-from ga4gh.core.models import Coding, iriReference as IRI, MappableConcept
 from ga4gh.va_spec.base.domain_entities import Condition
 
-from mavedb.lib.annotation.constants import GENERIC_DISEASE_MEDGEN_CODE, MEDGEN_SYSTEM
+from mavedb.lib.mondo import mondo_term_to_mappable_concept
+from mavedb.models.score_calibration import ScoreCalibration
 
 
-def generic_disease_condition_iri() -> IRI:
-    return IRI(root=f"http://identifiers.org/medgen/{GENERIC_DISEASE_MEDGEN_CODE}")
+def calibration_disease_condition(score_calibration: ScoreCalibration) -> Condition:
+    """The disease/disorder a calibration applies to, as a VA-Spec ``Condition``.
 
-
-def generic_disease_condition() -> Condition:
-    return Condition(
-        root=MappableConcept(
-            conceptType="Disease",
-            primaryCoding=Coding(
-                code=GENERIC_DISEASE_MEDGEN_CODE,
-                system=MEDGEN_SYSTEM,
-                iris=[generic_disease_condition_iri()],
-            ),
-        )
-    )
+    Every calibration carries a non-null MONDO disease term (often the generic "disease or
+    disorder" (``MONDO:0000001``)) which is serialized directly to a condition.
+    """
+    return Condition(root=mondo_term_to_mappable_concept(score_calibration.disease_term))

src/mavedb/lib/annotation/proposition.py

@@ -1,14 +1,16 @@
 from ga4gh.core.models import Coding, MappableConcept
 from ga4gh.va_spec.base.core import ExperimentalVariantFunctionalImpactProposition, VariantPathogenicityProposition
 
-from mavedb.lib.annotation.condition import generic_disease_condition
+from mavedb.lib.annotation.condition import calibration_disease_condition
 from mavedb.lib.annotation.document import experiment_to_document
 from mavedb.lib.annotation.util import sequence_feature_for_mapped_variant, variation_from_mapped_variant
 from mavedb.models.mapped_variant import MappedVariant
+from mavedb.models.score_calibration import ScoreCalibration
 
 
 def mapped_variant_to_experimental_variant_clinical_impact_proposition(
     mapped_variant: MappedVariant,
+    score_calibration: ScoreCalibration,
 ) -> VariantPathogenicityProposition:
     coding, system = sequence_feature_for_mapped_variant(mapped_variant)
     sequence_feature = MappableConcept(
@@ -19,7 +21,7 @@ def mapped_variant_to_experimental_variant_clinical_impact_proposition(
         description=f"Variant pathogenicity proposition for {mapped_variant.variant.urn}.",
         subjectVariant=variation_from_mapped_variant(mapped_variant),
         predicate="isCausalFor",
-        objectCondition=generic_disease_condition(),
+        objectCondition=calibration_disease_condition(score_calibration),
         geneContextQualifier=sequence_feature
         if system == "https://www.genenames.org/"
         else None,  # only include gene context if we have a gene identifier

src/mavedb/lib/exceptions.py

@@ -222,6 +222,12 @@ class HGNCServiceError(Exception):
     pass
 
 
+class MondoServiceError(Exception):
+    """Raised when the MONDO/OLS disease ontology service cannot provide search results."""
+
+    pass
+
+
 class LDHSubmissionFailureError(Exception):
     """Raised when submission to ClinGen Linked Data Hub (LDH) fails for all submissions."""
 

src/mavedb/lib/mondo.py

@@ -0,0 +1,99 @@
+"""MONDO disease-term vocabulary: constants, MappableConcept serialization, and the generic term.
+
+Calibrations carry a disease/disorder context drawn from the Monarch Disease Ontology (MONDO). This
+module owns the vocabulary's static surface: the generic "disease or disorder" root, conversion of a
+stored :class:`MondoTerm` (or an OLS suggestion) to the GA4GH ``MappableConcept`` served on the wire, and
+get-or-create of the generic term (mirroring :func:`mavedb.lib.taxonomies.find_or_create_taxonomy`).
+
+The generic term (``MONDO:0000001``) is the write-time default for a calibration with no specific
+disease, so ``disease_term_id`` is never null. The annotation layer serializes whatever term a
+calibration carries — generic or specific — as its VA-Spec disease condition (see
+:func:`mavedb.lib.annotation.condition.calibration_disease_condition`).
+
+Typeahead search and OLS-backed validation of a submitted code live in :mod:`mavedb.lib.mondo_ols`,
+kept separate because they need :mod:`mavedb.lib.logging.context` (an optional "server" dependency,
+tracked for a core-compatible rework in
+`#459 <https://github.com/VariantEffect/mavedb-api/issues/459>`_); this module has no such dependency,
+so callers that only need the constants or serialization (e.g. the score calibration view model) stay
+importable under core dependencies.
+"""
+
+from typing import TypedDict
+
+from ga4gh.core.models import Coding, MappableConcept, iriReference
+from sqlalchemy.orm import Session
+
+from mavedb.models.mondo_term import MondoTerm
+
+# The Monarch Disease Ontology (MONDO), the controlled vocabulary for disease terms used across MaveDB.
+MONDO_SYSTEM = "https://purl.obolibrary.org/obo/mondo.owl"
+
+# The generic "disease or disorder" root, used when a calibration names no specific disease.
+MONDO_GENERIC_CODE = "MONDO:0000001"
+MONDO_GENERIC_LABEL = "disease or disorder"
+
+
+class MondoSuggestion(TypedDict):
+    """A single MONDO term resolved from OLS: the CURIE code, its label, and its resolvable IRI."""
+
+    code: str
+    label: str
+    iri: str
+
+
+def mondo_iri(code: str) -> str:
+    """The resolvable OBO IRI for a MONDO CURIE (e.g. ``MONDO:0015263`` → ``.../obo/MONDO_0015263``)."""
+    return f"https://purl.obolibrary.org/obo/{code.replace(':', '_')}"
+
+
+def mondo_term_to_mappable_concept(term: MondoTerm) -> MappableConcept:
+    """Serialize a stored :class:`MondoTerm` as a GA4GH disease ``MappableConcept``."""
+    return MappableConcept(
+        conceptType="Disease",
+        name=term.label,
+        primaryCoding=Coding(
+            code=term.code,
+            system=term.system,
+            systemVersion=term.system_version,
+            iris=[iriReference(root=mondo_iri(term.code))],
+        ),
+    )
+
+
+def mondo_suggestion_to_mappable_concept(suggestion: MondoSuggestion) -> MappableConcept:
+    """Build a disease ``MappableConcept`` from an OLS search suggestion (for typeahead results)."""
+    return MappableConcept(
+        conceptType="Disease",
+        name=suggestion["label"],
+        primaryCoding=Coding(
+            code=suggestion["code"],
+            system=MONDO_SYSTEM,
+            iris=[iriReference(root=suggestion["iri"])],
+        ),
+    )
+
+
+def generic_disease_mappable_concept() -> MappableConcept:
+    """The generic "disease or disorder" concept, MaveDB's single unspecified-disease sentinel."""
+    return MappableConcept(
+        conceptType="Disease",
+        name=MONDO_GENERIC_LABEL,
+        primaryCoding=Coding(
+            code=MONDO_GENERIC_CODE,
+            system=MONDO_SYSTEM,
+            iris=[iriReference(root=mondo_iri(MONDO_GENERIC_CODE))],
+        ),
+    )
+
+
+def get_generic_disease_term(db: Session) -> MondoTerm:
+    """Get-or-create the generic disease term. Known-canonical, so it needs no OLS round trip."""
+    term = (
+        db.query(MondoTerm).filter(MondoTerm.system == MONDO_SYSTEM, MondoTerm.code == MONDO_GENERIC_CODE).one_or_none()
+    )
+    if term is None:
+        term = MondoTerm(code=MONDO_GENERIC_CODE, system=MONDO_SYSTEM, label=MONDO_GENERIC_LABEL)  # type: ignore[call-arg]
+        db.add(term)
+        db.flush()
+
+    return term