Refactor Score Ranges to Score Calibrations and Strengthen Their Data Model

alembic/manual_migrations/migrate_score_ranges_to_calibrations.py

@@ -0,0 +1,155 @@
+
+from typing import Union
+
+import sqlalchemy as sa
+from sqlalchemy.orm import Session
+
+# SQLAlchemy needs access to all models to properly map relationships.
+from mavedb.models import *
+
+from mavedb.db.session import SessionLocal
+from mavedb.models.score_set import ScoreSet
+from mavedb.models.score_calibration import ScoreCalibration as ScoreCalibrationDBModel
+from mavedb.models.publication_identifier import PublicationIdentifier
+from mavedb.view_models.score_range import (
+    ScoreRangeCreate,
+    ScoreSetRangesAdminCreate,
+    ZeibergCalibrationScoreRangesAdminCreate,
+    ScottScoreRangesAdminCreate,
+    InvestigatorScoreRangesAdminCreate,
+    IGVFCodingVariantFocusGroupControlScoreRangesAdminCreate,
+    IGVFCodingVariantFocusGroupMissenseScoreRangesAdminCreate,
+)
+
+score_range_kinds: dict[
+    str,
+    Union[
+        type[ZeibergCalibrationScoreRangesAdminCreate],
+        type[ScottScoreRangesAdminCreate],
+        type[InvestigatorScoreRangesAdminCreate],
+        type[IGVFCodingVariantFocusGroupControlScoreRangesAdminCreate],
+        type[IGVFCodingVariantFocusGroupMissenseScoreRangesAdminCreate],
+    ],
+] = {
+    "zeiberg_calibration": ZeibergCalibrationScoreRangesAdminCreate,
+    "scott_calibration": ScottScoreRangesAdminCreate,
+    "investigator_provided": InvestigatorScoreRangesAdminCreate,
+    "cvfg_all_variants": IGVFCodingVariantFocusGroupControlScoreRangesAdminCreate,
+    "cvfg_missense_variants": IGVFCodingVariantFocusGroupMissenseScoreRangesAdminCreate,
+}
+
+EVIDENCE_STRENGTH_FROM_POINTS = {
+    8: "Very Strong",
+    4: "Strong",
+    3: "Moderate+",
+    2: "Moderate",
+    1: "Supporting",
+}
+
+
+def do_migration(session: Session) -> None:
+    score_sets_with_ranges = (
+        session.execute(sa.select(ScoreSet).where(ScoreSet.score_ranges.isnot(None))).scalars().all()
+    )
+
+    for score_set in score_sets_with_ranges:
+        if not score_set.score_ranges:
+            continue    
+
+        score_set_ranges = ScoreSetRangesAdminCreate.model_validate(score_set.score_ranges)
+
+        for field in score_set_ranges.model_fields_set:
+            if field == "record_type":
+                continue
+
+            ranges = getattr(score_set_ranges, field)
+            if not ranges:
+                continue
+
+            range_model = score_range_kinds.get(field)
+            inferred_ranges = range_model.model_validate(ranges)
+
+            model_thresholds = []
+            for range in inferred_ranges.ranges:
+                model_thresholds.append(ScoreRangeCreate.model_validate(range.__dict__).model_dump())
+
+            # We should migrate the zeiberg evidence classifications to be explicitly part of the calibration ranges.
+            if field == "zeiberg_calibration":
+                for inferred_range, model_range in zip(
+                    inferred_ranges.ranges,
+                    model_thresholds,
+                ):
+                    model_range["label"] = f"PS3 {EVIDENCE_STRENGTH_FROM_POINTS.get(inferred_range.evidence_strength, 'Unknown')}" if inferred_range.evidence_strength > 0 else f"BS3 {EVIDENCE_STRENGTH_FROM_POINTS.get(abs(inferred_range.evidence_strength), 'Unknown')}"
+                    model_range["acmg_classification"] = {"points": inferred_range.evidence_strength}
+
+            # Reliant on existing behavior that these sources have been created already.
+            # If not present, no sources will be associated.
+            if "odds_path_source" in inferred_ranges.model_fields_set and inferred_ranges.odds_path_source:
+                oddspaths_sources = (
+                    session.execute(
+                        sa.select(PublicationIdentifier).where(
+                            PublicationIdentifier.identifier.in_(
+                                [src.identifier for src in (inferred_ranges.odds_path_source or [])]
+                            )
+                        )
+                    )
+                    .scalars()
+                    .all()
+                )
+            else:
+                oddspaths_sources = []
+
+            if "source" in inferred_ranges.model_fields_set and inferred_ranges.source:
+                range_sources = (
+                    session.execute(
+                        sa.select(PublicationIdentifier).where(
+                            PublicationIdentifier.identifier.in_(
+                                [src.identifier for src in (inferred_ranges.source or [])]
+                            )
+                        )
+                    )
+                    .scalars()
+                    .all()
+                )
+            else:
+                range_sources = []
+
+            sources = set()
+            for publication in oddspaths_sources:
+                setattr(publication, "relation", "method")
+                sources.add(publication)
+            for publication in range_sources:
+                setattr(publication, "relation", "threshold")
+                sources.add(publication)
+
+            score_calibration = ScoreCalibrationDBModel(
+                score_set_id=score_set.id,
+                title=inferred_ranges.title,
+                research_use_only=inferred_ranges.research_use_only,
+                primary=inferred_ranges.primary,
+                private=False, # All migrated calibrations are public.
+                investigator_provided=True if field == "investigator_provided" else False,
+                baseline_score=inferred_ranges.baseline_score
+                if "baseline_score" in inferred_ranges.model_fields_set
+                else None,
+                baseline_score_description=inferred_ranges.baseline_score_description
+                if "baseline_score_description" in inferred_ranges.model_fields_set
+                else None,
+                functional_ranges=None if not model_thresholds else model_thresholds,
+                calibration_metadata=None,
+                publication_identifiers=sources,
+                # If investigator_provided, set to creator of score set, else set to default system user (1).
+                created_by_id=score_set.created_by_id if field == "investigator_provided" else 1,
+                modified_by_id=score_set.created_by_id if field == "investigator_provided" else 1,
+            )
+            session.add(score_calibration)
+
+
+if __name__ == "__main__":
+    db = SessionLocal()
+    db.current_user = None  # type: ignore
+
+    do_migration(db)
+
+    db.commit()
+    db.close()

alembic/versions/002f6f9ec7ac_add_score_calibration_table.py

@@ -0,0 +1,92 @@
+"""add score calibration table
+
+Revision ID: 002f6f9ec7ac
+Revises: 019eb75ad9ae
+Create Date: 2025-10-08 08:59:10.563528
+
+"""
+
+from alembic import op
+import sqlalchemy as sa
+from sqlalchemy.dialects import postgresql
+
+
+# revision identifiers, used by Alembic.
+revision = "002f6f9ec7ac"
+down_revision = "019eb75ad9ae"
+branch_labels = None
+depends_on = None
+
+
+def upgrade():
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.create_table(
+        "score_calibrations",
+        sa.Column("id", sa.Integer(), nullable=False),
+        sa.Column("urn", sa.String(length=64), nullable=True),
+        sa.Column("score_set_id", sa.Integer(), nullable=False),
+        sa.Column("title", sa.String(), nullable=False),
+        sa.Column("research_use_only", sa.Boolean(), nullable=False),
+        sa.Column("primary", sa.Boolean(), nullable=False),
+        sa.Column("investigator_provided", sa.Boolean(), nullable=False),
+        sa.Column("private", sa.Boolean(), nullable=False),
+        sa.Column("notes", sa.String(), nullable=True),
+        sa.Column("baseline_score", sa.Float(), nullable=True),
+        sa.Column("baseline_score_description", sa.String(), nullable=True),
+        sa.Column("functional_ranges", postgresql.JSONB(astext_type=sa.Text(), none_as_null=True), nullable=True),
+        sa.Column("calibration_metadata", postgresql.JSONB(astext_type=sa.Text(), none_as_null=True), nullable=True),
+        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(
+            ["score_set_id"],
+            ["scoresets.id"],
+        ),
+        sa.ForeignKeyConstraint(
+            ["created_by_id"],
+            ["users.id"],
+        ),
+        sa.ForeignKeyConstraint(
+            ["modified_by_id"],
+            ["users.id"],
+        ),
+        sa.PrimaryKeyConstraint("id"),
+    )
+    op.create_table(
+        "score_calibration_publication_identifiers",
+        sa.Column("score_calibration_id", sa.Integer(), nullable=False),
+        sa.Column("publication_identifier_id", sa.Integer(), nullable=False),
+        sa.Column(
+            "relation",
+            sa.Enum(
+                "thresholds",
+                "classifications",
+                "methods",
+                name="scorecalibrationrelation",
+                native_enum=False,
+                length=32,
+            ),
+            nullable=False,
+        ),
+        sa.ForeignKeyConstraint(
+            ["publication_identifier_id"],
+            ["publication_identifiers.id"],
+        ),
+        sa.ForeignKeyConstraint(
+            ["score_calibration_id"],
+            ["score_calibrations.id"],
+        ),
+        sa.PrimaryKeyConstraint("score_calibration_id", "publication_identifier_id", "relation"),
+    )
+    op.create_index(op.f("ix_score_calibrations_urn"), "score_calibrations", ["urn"], unique=True)
+
+    # ### end Alembic commands ###
+
+
+def downgrade():
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.drop_index(op.f("ix_score_calibrations_urn"), table_name="score_calibrations")
+    op.drop_table("score_calibration_publication_identifiers")
+    op.drop_table("score_calibrations")
+    # ### end Alembic commands ###

alembic/versions/f5a72192fafd_remove_score_range_property_from_score_.py

@@ -0,0 +1,32 @@
+"""remove score range property from score sets
+
+Revision ID: f5a72192fafd
+Revises: 002f6f9ec7ac
+Create Date: 2025-10-08 15:35:49.275162
+
+"""
+
+from alembic import op
+import sqlalchemy as sa
+from sqlalchemy.dialects import postgresql
+
+# revision identifiers, used by Alembic.
+revision = "f5a72192fafd"
+down_revision = "002f6f9ec7ac"
+branch_labels = None
+depends_on = None
+
+
+def upgrade():
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.drop_column("scoresets", "score_ranges")
+    # ### end Alembic commands ###
+
+
+def downgrade():
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.add_column(
+        "scoresets",
+        sa.Column("score_ranges", postgresql.JSONB(astext_type=sa.Text()), autoincrement=False, nullable=True),
+    )
+    # ### end Alembic commands ###