Merge pull request #435 from VariantEffect/jstone-dev/clingen-linkage

bencap · web-flow · commit 0febe7874d0e · 2025-05-23T11:09:18.000-07:00
ClinGen &amp; ClinVar script enhancements
diff --git a/src/mavedb/lib/clingen/content_constructors.py b/src/mavedb/lib/clingen/content_constructors.py
@@ -1,4 +1,5 @@
 from datetime import datetime
+from typing import Optional
 from uuid import uuid4
 from urllib.parse import quote_plus
 
@@ -15,7 +16,7 @@ def construct_ldh_submission_event(sbj: LdhContentSubject) -> LdhEvent:
         "type": LDH_SUBMISSION_TYPE,
         "name": LDH_ENTITY_NAME,
         "uuid": str(uuid4()),
-        "sbj": {"id": sbj["Variant"]["hgvs"], "type": "Variant", "format": "hgvs", "add": True},
+        "sbj": {"id": sbj["Variant"]["hgvs"], "type": "Variant", "format": "hgvs", "add": True, "iri": None},
         "triggered": {
             "by": {
                 "host": MAVEDB_BASE_GIT,
@@ -31,26 +32,29 @@ def construct_ldh_submission_subject(hgvs: str) -> LdhContentSubject:
     return {"Variant": {"hgvs": hgvs}}
 
 
-def construct_ldh_submission_entity(variant: Variant, mapped_variant: MappedVariant) -> LdhContentLinkedData:
-    return {
+def construct_ldh_submission_entity(variant: Variant, mapped_variant: Optional[MappedVariant]) -> LdhContentLinkedData:
+    entity: LdhContentLinkedData = {
         # TODO#372: We try to make all possible fields that are non-nullable represented that way.
         "MaveDBMapping": [
             {
                 "entContent": {
                     "mavedb_id": variant.urn,  # type: ignore
-                    "pre_mapped": mapped_variant.pre_mapped,  # type: ignore
-                    "post_mapped": mapped_variant.post_mapped,  # type: ignore
-                    "mapping_api_version": mapped_variant.mapping_api_version,  # type: ignore
                     "score": variant.data["score_data"]["score"],  # type: ignore
+                    "score_set_description": variant.score_set.short_description,  # type: ignore
                 },
                 "entId": variant.urn,  # type: ignore
                 "entIri": f"{MAVEDB_FRONTEND_URL}/score-sets/{quote_plus(variant.score_set.urn)}?variant={quote_plus(variant.urn)}",  # type: ignore
             }
         ]
     }
+    if mapped_variant is not None:
+        entity["MaveDBMapping"][0]["entContent"]["pre_mapped"] = mapped_variant.pre_mapped
+        entity["MaveDBMapping"][0]["entContent"]["post_mapped"] = mapped_variant.post_mapped
+        entity["MaveDBMapping"][0]["entContent"]["mapping_api_version"] = mapped_variant.mapping_api_version
+    return entity
 
 
-def construct_ldh_submission(variant_content: list[tuple[str, Variant, MappedVariant]]) -> list[LdhSubmission]:
+def construct_ldh_submission(variant_content: list[tuple[str, Variant, Optional[MappedVariant]]]) -> list[LdhSubmission]:
     content_submission: list[LdhSubmission] = []
     for hgvs, variant, mapped_variant in variant_content:
         subject = construct_ldh_submission_subject(hgvs)
diff --git a/src/mavedb/lib/types/clingen.py b/src/mavedb/lib/types/clingen.py
@@ -1,4 +1,4 @@
-from typing import TypedDict, Literal
+from typing import Optional, TypedDict, Literal
 from typing_extensions import NotRequired
 
 
@@ -14,6 +14,7 @@ class EventSbj(TypedDict):
     type: str
     format: Literal["hgvs", "alleleRegistryID", "clinvarID", "geneSymbol"]
     add: bool
+    iri: Optional[str]
 
 
 # Who/what triggered the event
@@ -52,9 +53,9 @@ class LdhContentSubject(TypedDict):
 # The entities we are submitting
 class LdhMapping(TypedDict):
     mavedb_id: str
-    pre_mapped: str
-    post_mapped: str
-    mapping_api_version: str
+    pre_mapped: Optional[str]
+    post_mapped: Optional[str]
+    mapping_api_version: Optional[str]
     score: float
 
 
diff --git a/src/mavedb/scripts/clingen_ldh_submission.py b/src/mavedb/scripts/clingen_ldh_submission.py
@@ -1,8 +1,9 @@
 import click
 import logging
-from typing import Sequence
+import re
+from typing import Optional, Sequence
 
-from sqlalchemy import select
+from sqlalchemy import and_, select
 from sqlalchemy.orm import Session
 
 from mavedb.models.score_set import ScoreSet
@@ -16,8 +17,10 @@
 
 logger = logging.getLogger(__name__)
 
+intronic_variant_with_reference_regex = re.compile(r":c\..*[+-]")
+variant_with_reference_regex = re.compile(r":")
 
-def submit_urns_to_clingen(db: Session, urns: Sequence[str], debug: bool) -> list[str]:
+def submit_urns_to_clingen(db: Session, urns: Sequence[str], unlinked_only: bool, prefer_unmapped_hgvs: bool, debug: bool) -> list[str]:
     ldh_service = ClinGenLdhService(url=LDH_SUBMISSION_ENDPOINT)
     ldh_service.authenticate()
 
@@ -37,13 +40,12 @@ def submit_urns_to_clingen(db: Session, urns: Sequence[str], debug: bool) -> lis
                 continue
 
             logger.info(f"Submitting mapped variants to LDH service for score set with URN: {urn}")
+            mapped_variant_join_clause = and_(MappedVariant.variant_id == Variant.id, MappedVariant.post_mapped.is_not(None), MappedVariant.current.is_(True))
             variant_objects = db.execute(
                 select(Variant, MappedVariant)
-                .join(MappedVariant)
+                .join(MappedVariant, mapped_variant_join_clause, isouter=True)
                 .join(ScoreSet)
                 .where(ScoreSet.urn == urn)
-                .where(MappedVariant.post_mapped.is_not(None))
-                .where(MappedVariant.current.is_(True))
             ).all()
 
             if not variant_objects:
@@ -52,12 +54,48 @@ def submit_urns_to_clingen(db: Session, urns: Sequence[str], debug: bool) -> lis
 
             logger.debug(f"Preparing {len(variant_objects)} mapped variants for submission")
 
-            variant_content: list[tuple[str, Variant, MappedVariant]] = []
+            variant_content: list[tuple[str, Variant, Optional[MappedVariant]]] = []
             for variant, mapped_variant in variant_objects:
-                variation = hgvs_from_mapped_variant(mapped_variant)
+                if mapped_variant is None:
+                    if variant.hgvs_nt is not None and intronic_variant_with_reference_regex.search(variant.hgvs_nt):
+                        # Use the hgvs_nt string for unmapped intronic variants. This is because our mapper does not yet
+                        # support mapping intronic variants.
+                        variation = [variant.hgvs_nt]
+                        if variation:
+                            logger.info(f"Using hgvs_nt for unmapped intronic variant {variant.urn}: {variation}")
+                    elif variant.hgvs_nt is not None and variant_with_reference_regex.search(variant.hgvs_nt):
+                        # Use the hgvs_nt string for other unmapped NT variants in accession-based score sets.
+                        variation = [variant.hgvs_nt]
+                        if variation:
+                            logger.info(f"Using hgvs_nt for unmapped non-intronic variant {variant.urn}: {variation}")
+                    elif variant.hgvs_pro is not None and variant_with_reference_regex.search(variant.hgvs_pro):
+                        # Use the hgvs_pro string for unmapped PRO variants in accession-based score sets.
+                        variation = [variant.hgvs_pro]
+                        if variation:
+                            logger.info(f"Using hgvs_pro for unmapped non-intronic variant {variant.urn}: {variation}")
+                    else:
+                        logger.warning(f"No variation found for unmapped variant {variant.urn} (nt: {variant.hgvs_nt}, aa: {variant.hgvs_pro}, splice: {variant.hgvs_splice}).")
+                        continue
+                else:
+                    if unlinked_only and mapped_variant.clingen_allele_id:
+                        continue
+                    # If the script was run with the --prefer-unmapped-hgvs flag, use the hgvs_nt string rather than the
+                    # mapped variant, as long as the variant is accession-based.
+                    if prefer_unmapped_hgvs and variant.hgvs_nt is not None and variant_with_reference_regex.search(variant.hgvs_nt):
+                        variation = [variant.hgvs_nt]
+                        if variation:
+                            logger.info(f"Using hgvs_nt for mapped variant {variant.urn}: {variation}")
+                    elif prefer_unmapped_hgvs and variant.hgvs_pro is not None and variant_with_reference_regex.search(variant.hgvs_pro):
+                        variation = [variant.hgvs_pro]
+                        if variation:
+                            logger.info(f"Using hgvs_pro for mapped variant {variant.urn}: {variation}")                # continue  # TEMPORARY. Only submit unmapped variants.
+                    else:
+                        variation = hgvs_from_mapped_variant(mapped_variant)
+                        if variation:
+                            logger.info(f"Using mapped variant for {variant.urn}: {variation}")
 
                 if not variation:
-                    logger.warning(f"No variation found for variant {variant.urn}.")
+                    logger.warning(f"No variation found for mapped variant {variant.urn} (nt: {variant.hgvs_nt}, aa: {variant.hgvs_pro}, splice: {variant.hgvs_splice}).")
                     continue
 
                 for allele in variation:
@@ -90,11 +128,13 @@ def submit_urns_to_clingen(db: Session, urns: Sequence[str], debug: bool) -> lis
 @click.command()
 @with_database_session
 @click.argument("urns", nargs=-1)
-@click.option("--all", help="Submit mapped variants for every score set in MaveDB.", is_flag=True)
+@click.option("--all", help="Submit variants for every score set in MaveDB.", is_flag=True)
+@click.option("--unlinked", default=False, help="Only submit variants that have not already been linked to ClinGen alleles.", is_flag=True)
+@click.option("--prefer-unmapped-hgvs", default=False, help="If the unmapped HGVS string is accession-based, use it in the submission instead of the mapped variant.", is_flag=True)
 @click.option("--suppress-output", help="Suppress final print output to the console.", is_flag=True)
 @click.option("--debug", help="Enable debug mode. This will send only one request at most to ClinGen", is_flag=True)
 def submit_clingen_urns_command(
-    db: Session, urns: Sequence[str], all: bool, suppress_output: bool, debug: bool
+    db: Session, urns: Sequence[str], all: bool, unlinked: bool, prefer_unmapped_hgvs: bool, suppress_output: bool, debug: bool
 ) -> None:
     """
     Submit data to ClinGen for mapped variant allele ID generation for the given URNs.
@@ -111,7 +151,7 @@ def submit_clingen_urns_command(
         logger.error("No URNs provided. Please provide at least one URN.")
         return
 
-    submitted_variant_urns = submit_urns_to_clingen(db, urns, debug)
+    submitted_variant_urns = submit_urns_to_clingen(db, urns, unlinked, prefer_unmapped_hgvs, debug)
 
     if not suppress_output:
         print(", ".join(submitted_variant_urns))
diff --git a/src/mavedb/scripts/link_clingen_variants.py b/src/mavedb/scripts/link_clingen_variants.py
@@ -2,7 +2,7 @@
 import logging
 from typing import Sequence
 
-from sqlalchemy import select
+from sqlalchemy import and_, select
 from sqlalchemy.orm import Session
 
 from mavedb.lib.clingen.linked_data_hub import get_clingen_variation, clingen_allele_id_from_ldh_variation
@@ -18,7 +18,7 @@
 @with_database_session
 @click.argument("urns", nargs=-1)
 @click.option("--score-sets/--variants", default=False)
-@click.option("--unlinked", default=False)
+@click.option("--unlinked", default=False, is_flag=True)
 def link_clingen_variants(db: Session, urns: Sequence[str], score_sets: bool, unlinked: bool) -> None:
     """
     Submit data to ClinGen for mapped variant allele ID generation for the given URNs.
@@ -51,7 +51,7 @@ def link_clingen_variants(db: Session, urns: Sequence[str], score_sets: bool, un
             failed_urns.append(urn)
             continue
 
-        mapped_variant = db.scalar(select(MappedVariant).join(Variant).where(Variant.urn == urn))
+        mapped_variant = db.scalar(select(MappedVariant).join(Variant).where(and_(Variant.urn == urn, MappedVariant.current.is_(True))))
 
         if not mapped_variant:
             logger.warning(f"No mapped variant found for URN {urn}.")
diff --git a/src/mavedb/scripts/refresh_clinvar_variant_data.py b/src/mavedb/scripts/refresh_clinvar_variant_data.py
@@ -1,4 +1,6 @@
 import click
+from mavedb.models.score_set import ScoreSet
+from mavedb.models.variant import Variant
 import requests
 import csv
 import time
@@ -7,10 +9,10 @@
 import random
 import io
 
-from typing import Dict, Any, Optional
+from typing import Dict, Any, Optional, Sequence
 from datetime import date
 
-from sqlalchemy import select, distinct, func
+from sqlalchemy import and_, select, distinct
 from sqlalchemy.orm import Session
 
 from mavedb.models.mapped_variant import MappedVariant
@@ -62,20 +64,33 @@ def query_clingen_allele_api(allele_id: str) -> Dict[str, Any]:
     return response.json()
 
 
-def refresh_clinvar_variants(db: Session, month: Optional[str], year: Optional[str]) -> None:
+def refresh_clinvar_variants(db: Session, month: Optional[str], year: Optional[str], urns: Sequence[str]) -> None:
     tsv_content = fetch_clinvar_variant_summary_tsv(month, year)
     tsv_data = parse_tsv(tsv_content)
     version = f"{month}_{year}" if month and year else f"{date.today().month}_{date.today().year}"
     logger.info(f"Fetched TSV variant data for ClinVar for {version}.")
 
-    total_variants_with_clingen_ids = db.scalar(func.count(distinct(MappedVariant.clingen_allele_id)))
-    clingen_ids = db.scalars(
-        select(distinct(MappedVariant.clingen_allele_id)).where(MappedVariant.clingen_allele_id.is_not(None))
-    ).all()
+    if urns:
+        clingen_ids = db.scalars(
+            select(distinct(MappedVariant.clingen_allele_id))
+                .join(Variant)
+                .join(ScoreSet)
+                .where(MappedVariant.current.is_(True), MappedVariant.post_mapped.is_not(None))
+                .where(and_(
+                    MappedVariant.clingen_allele_id.is_not(None),
+                    MappedVariant.current.is_(True),
+                    ScoreSet.urn.in_(urns)
+                ))
+        ).all()
+    else:
+        clingen_ids = db.scalars(
+            select(distinct(MappedVariant.clingen_allele_id)).where(MappedVariant.clingen_allele_id.is_not(None))
+        ).all()
+    total_variants_with_clingen_ids = len(clingen_ids)
 
     logger.info(f"Fetching ClinGen data for {total_variants_with_clingen_ids} variants.")
     for index, clingen_id in enumerate(clingen_ids):
-        if total_variants_with_clingen_ids > 0 and index % (total_variants_with_clingen_ids // 100) == 0:
+        if total_variants_with_clingen_ids > 0 and index % (max(total_variants_with_clingen_ids // 100, 1)) == 0:
             logger.info(f"Progress: {index / total_variants_with_clingen_ids:.0%}")
 
         # Guaranteed based on our query filters.
@@ -116,6 +131,8 @@ def refresh_clinvar_variants(db: Session, month: Optional[str], year: Optional[s
             select(MappedVariant).where(MappedVariant.clingen_allele_id == clingen_id)
         ).all()
         for mapped_variant in variants_with_clingen_allele_id:
+            if clinvar_variant.id in [c.id for c in mapped_variant.clinical_controls]:
+                continue
             mapped_variant.clinical_controls.append(clinvar_variant)
             db.add(mapped_variant)
 
@@ -127,10 +144,11 @@ def refresh_clinvar_variants(db: Session, month: Optional[str], year: Optional[s
 
 @click.command()
 @with_database_session
+@click.argument("urns", nargs=-1)
 @click.option("--month", default=None, help="Populate mapped variants for every score set in MaveDB.")
 @click.option("--year", default=None, help="Populate mapped variants for every score set in MaveDB.")
-def refresh_clinvar_variants_command(db: Session, month: Optional[str], year: Optional[str]) -> None:
-    refresh_clinvar_variants(db, month, year)
+def refresh_clinvar_variants_command(db: Session, month: Optional[str], year: Optional[str], urns: Sequence[str]) -> None:
+    refresh_clinvar_variants(db, month, year, urns)
 
 
 if __name__ == "__main__":
diff --git a/tests/lib/clingen/test_content_constructors.py b/tests/lib/clingen/test_content_constructors.py
@@ -11,6 +11,7 @@
 )
 from mavedb.lib.clingen.constants import LDH_ENTITY_NAME, LDH_SUBMISSION_TYPE
 from mavedb import __version__
+import pytest
 
 from tests.helpers.constants import (
     TEST_HGVS_IDENTIFIER,
@@ -44,6 +45,7 @@ def test_construct_ldh_submission_event():
             "type": "Variant",
             "format": "hgvs",
             "add": True,
+            "iri": None
         }
         assert result["triggered"]["by"] == {
             "host": MAVEDB_BASE_GIT,
@@ -52,30 +54,40 @@ def test_construct_ldh_submission_event():
         }
 
 
-def test_construct_ldh_submission_entity(mock_variant, mock_mapped_variant):
-    result = construct_ldh_submission_entity(mock_variant, mock_mapped_variant)
+@pytest.mark.parametrize("has_mapped_variant", [(True), (False)])
+def test_construct_ldh_submission_entity(mock_variant, mock_mapped_variant, has_mapped_variant: bool):
+    mapped_variant = mock_mapped_variant if has_mapped_variant else None
+    result = construct_ldh_submission_entity(mock_variant, mapped_variant)
 
     assert "MaveDBMapping" in result
     assert len(result["MaveDBMapping"]) == 1
     mapping = result["MaveDBMapping"][0]
 
     assert mapping["entContent"]["mavedb_id"] == VALID_VARIANT_URN
-    assert mapping["entContent"]["pre_mapped"] == TEST_VALID_PRE_MAPPED_VRS_ALLELE_VRS2_X
-    assert mapping["entContent"]["post_mapped"] == TEST_VALID_POST_MAPPED_VRS_ALLELE_VRS2_X
-    assert mapping["entContent"]["mapping_api_version"] == "pytest.mapping.1.0"
     assert mapping["entContent"]["score"] == 1.0
 
+    if has_mapped_variant:
+        assert mapping["entContent"]["pre_mapped"] == TEST_VALID_PRE_MAPPED_VRS_ALLELE_VRS2_X
+        assert mapping["entContent"]["post_mapped"] == TEST_VALID_POST_MAPPED_VRS_ALLELE_VRS2_X
+        assert mapping["entContent"]["mapping_api_version"] == "pytest.mapping.1.0"
+    else:
+        assert "pre_mapped" not in mapping["entContent"]
+        assert "post_mapped" not in mapping["entContent"]
+        assert "mapping_api_version" not in mapping["entContent"]
+
     assert mapping["entId"] == VALID_VARIANT_URN
     assert (
         mapping["entIri"]
         == f"{MAVEDB_FRONTEND_URL}/score-sets/{quote_plus(VALID_SCORE_SET_URN)}?variant={quote_plus(VALID_VARIANT_URN)}"
     )
 
 
-def test_construct_ldh_submission(mock_variant, mock_mapped_variant):
+@pytest.mark.parametrize("has_mapped_variant", [(True), (False)])
+def test_construct_ldh_submission(mock_variant, mock_mapped_variant, has_mapped_variant: bool):
+    mapped_variant = mock_mapped_variant if has_mapped_variant else None
     variant_content = [
-        (TEST_HGVS_IDENTIFIER, mock_variant, mock_mapped_variant),
-        (TEST_HGVS_IDENTIFIER, mock_variant, mock_mapped_variant),
+        (TEST_HGVS_IDENTIFIER, mock_variant, mapped_variant),
+        (TEST_HGVS_IDENTIFIER, mock_variant, mapped_variant),
     ]
 
     uuid_1 = UUID("12345678-1234-5678-1234-567812345678")
