wip: VA-Spec implementation for mapped variants

Author: bencap

src/mavedb/lib/annotation/agent.py

@@ -0,0 +1,39 @@
+import logging
+
+from ga4gh.core.entity_models import Agent, Extension, AgentSubtype
+
+from mavedb import __version__
+
+logger = logging.getLogger(__name__)
+
+
+def mavedb_api_agent():
+    version_at_time_of_generation = Extension(
+        name="mavedb_api_version",
+        value=__version__,
+        # TODO: stitched from constant base URL?
+        description="The MaveDB API version used to generate this record. See: https://github.com/VariantEffect/mavedb-api/releases",
+    )
+
+    return Agent(
+        subtype=AgentSubtype.SOFTWARE,
+        label="mavedb api",
+        description=f"mavedb api agent, version {__version__}",
+        extensions=[version_at_time_of_generation],
+    )
+
+
+def mavedb_vrs_agent(version: str):
+    version_at_time_of_variant_generation = Extension(
+        name="mavedb_vrs_version",
+        value=version,
+        # TODO: stitched from constant base URL?
+        description="The VRS mapping version used to generate this record. See: https://github.com/VariantEffect/dcd_mapping2/releases",
+    )
+
+    return Agent(
+        subtype=AgentSubtype.SOFTWARE,
+        label="mavedb vrs mapper",
+        description=f"mavedb vrs mapping agent, version {version_at_time_of_variant_generation.value}",
+        extensions=[version_at_time_of_variant_generation],
+    )

src/mavedb/lib/annotation/classification.py

@@ -0,0 +1,31 @@
+import logging
+
+from fastapi.encoders import jsonable_encoder
+from ga4gh.va_spec.profiles.assay_var_effect import AveFunctionalClassification
+
+from mavedb.models.mapped_variant import MappedVariant
+from mavedb.lib.validation.utilities import inf_or_float
+from mavedb.view_models.score_set import ScoreRanges
+
+logger = logging.getLogger(__name__)
+
+
+def functional_classification_of_variant(mapped_variant: MappedVariant) -> AveFunctionalClassification:
+    if mapped_variant.variant.score_set.score_ranges is None:
+        return AveFunctionalClassification.INDETERMINATE
+
+    # This view model object is much simpler to work with.
+    score_ranges = ScoreRanges(**jsonable_encoder(mapped_variant.variant.score_set.score_ranges))
+
+    # This property of this column is guaranteed to be defined.
+    functional_score: float = mapped_variant.variant.data["score"]  # type: ignore
+    for range in score_ranges.ranges:
+        lower_bound, upper_bound = inf_or_float(range.range[0], lower=True), inf_or_float(range.range[1], lower=False)
+        if functional_score > lower_bound and functional_score <= upper_bound:
+            return (
+                AveFunctionalClassification.NORMAL
+                if range.classification == "normal"
+                else AveFunctionalClassification.ABNORMAL
+            )
+
+    return AveFunctionalClassification.INDETERMINATE

src/mavedb/lib/annotation/contribution.py

@@ -0,0 +1,23 @@
+import logging
+from datetime import datetime
+
+from ga4gh.core.entity_models import Contribution
+
+from mavedb.models.mapped_variant import MappedVariant
+from mavedb.lib.annotation.agent import mavedb_api_agent, mavedb_vrs_agent
+from mavedb.lib.annotation.method import mavedb_api_as_method, mavedb_vrs_as_method
+
+
+logger = logging.getLogger(__name__)
+
+
+def mavedb_api_contribution():
+    return Contribution(contributor=[mavedb_api_agent()], date=datetime.today(), specifiedBy=[mavedb_api_as_method()])
+
+
+def mavedb_vrs_contribution(mapped_variant: MappedVariant):
+    return Contribution(
+        contributor=[mavedb_vrs_agent(mapped_variant.mapping_api_version)],
+        date=mapped_variant.mapped_date,
+        specifiedBy=[mavedb_vrs_as_method()],
+    )

src/mavedb/lib/annotation/dataset.py

@@ -0,0 +1,24 @@
+import logging
+
+from ga4gh.core.entity_models import DataSet
+
+from mavedb.models.score_set import ScoreSet
+
+from mavedb.lib.annotation.document import score_set_to_document
+
+logger = logging.getLogger(__name__)
+
+
+# TODO: How deep should these objects get? See: https://github.com/ga4gh/va-spec/blob/1.x/examples/mavedb/pten-variant-example.json
+def score_set_to_data_set(score_set: ScoreSet) -> DataSet:
+    """
+    Transforms a score set object to a GA4GH DataSet object.
+    """
+    return DataSet(
+        id=score_set.urn,
+        subtype="variant effect data set",
+        license=score_set.license.short_name,
+        # TODO: Stitch together constants to create this string.
+        # TODO: Better to use this IRI reference, or the actual `Document` GA4GH object?
+        reportedIn=score_set_to_document(score_set),
+    )

src/mavedb/lib/annotation/document.py

@@ -0,0 +1,46 @@
+import logging
+import urllib.parse
+
+from ga4gh.core.entity_models import Document
+
+from mavedb.models.experiment import Experiment
+from mavedb.models.score_set import ScoreSet
+from mavedb.models.variant import Variant
+
+logger = logging.getLogger(__name__)
+
+# TODO: move centrally
+BASE_URL = "https://mavedb.org"
+
+
+def experiment_to_document(experiment: Experiment) -> Document:
+    return Document(
+        id=experiment.urn,
+        # TODO: necessary? better as label?
+        subtype="MaveDB experiment",
+        title=experiment.title,
+        urls=f"{BASE_URL}/experiments/{experiment.urn}",
+    )
+
+
+def score_set_to_document(score_set: ScoreSet) -> Document:
+    return Document(
+        id=score_set.urn,
+        # TODO: necessary? better as label?
+        subtype="MaveDB score set",
+        title=score_set.title,
+        urls=f"{BASE_URL}/score-sets/{score_set.urn}",
+    )
+
+
+def variant_to_document(variant: Variant) -> Document:
+    return Document(
+        id=variant.urn,
+        # TODO: necessary? better as label?
+        subtype="MaveDB variant",
+        # TODO: Maybe the variant specific page? Maybe nothing, if we don't guarantee a
+        #       variant specific page exists?
+        # TODO: This should be stitched from constants
+        # TODO(#372)
+        urls=f"https://mavedb.org/score-sets/{variant.score_set.urn}?variant={urllib.parse.quote_plus(variant.urn)}",  # type: ignore
+    )

src/mavedb/lib/annotation/method.py

@@ -0,0 +1,58 @@
+import logging
+from typing import Optional, Sequence, Union
+
+from ga4gh.core.entity_models import Method
+
+from mavedb.models.publication_identifier import PublicationIdentifier
+from mavedb.models.score_set_publication_identifier import ScoreSetPublicationIdentifierAssociation
+from mavedb.models.experiment_publication_identifier import ExperimentPublicationIdentifierAssociation
+
+logger = logging.getLogger(__name__)
+
+
+# TODO: How deep should these objects get?
+def publication_identifier_to_method(publication: PublicationIdentifier) -> Method:
+    """
+    Generate a GA4GH method definition based on the provided publication
+    """
+    return Method(
+        # TODO: Is this necessarily true?
+        subtype="experimental protocol",
+        reportedIn=publication.url,
+    )
+
+
+PublicationIdentifierAssociations = Union[
+    Sequence[ScoreSetPublicationIdentifierAssociation],
+    Sequence[ExperimentPublicationIdentifierAssociation],
+]
+
+
+def publication_identifiers_to_method(publications: PublicationIdentifierAssociations) -> Optional[Method]:
+    """
+    Generate a GA4GH method definition based on the list of provided publications
+    """
+    primary_publication_identifier = next(
+        (publication.publication for publication in publications if publication.primary), None
+    )
+    return (
+        Method(
+            # TODO: Is this necessarily true?
+            subtype="experimental protocol",
+            reportedIn=primary_publication_identifier.url,
+        )
+        if primary_publication_identifier
+        else None
+    )
+
+
+def mavedb_api_as_method():
+    return Method(subtype="software version", reportedIn="https://github.com/VariantEffect/mavedb-api/releases")
+
+
+def mavedb_vrs_as_method():
+    return Method(subtype="software version", reportedIn="https://github.com/VariantEffect/dcd_mapping2/releases")
+
+
+def variant_interpretation_guideline_method():
+    return Method(subtype="Variant Interpretation Guideline", reportedIn="https://pubmed.ncbi.nlm.nih.gov/29785012/")

src/mavedb/lib/annotation/py.typed

@@ -0,0 +1,87 @@
+import logging
+
+from ga4gh.va_spec.profiles import (
+    AssayVariantEffectMeasurementStudyResult,
+    AssayVariantEffectFunctionalClassificationStatement,
+    AssayVariantEffectClinicalClassificationStatement,
+)
+
+
+from mavedb.models.mapped_variant import MappedVariant
+from mavedb.lib.annotation.classification import functional_classification_of_variant
+from mavedb.lib.annotation.contribution import mavedb_api_contribution, mavedb_vrs_contribution
+from mavedb.lib.annotation.dataset import score_set_to_data_set
+from mavedb.lib.annotation.method import (
+    publication_identifiers_to_method,
+    variant_interpretation_functional_guideline_method,
+    variant_interpretation_clinical_guideline_method,
+)
+from mavedb.lib.annotation.document import variant_to_document, score_set_to_document
+
+
+logger = logging.getLogger(__name__)
+
+
+def mapped_variant_to_variant_effect_measurement_study_result(
+    mapped_variant: MappedVariant,
+) -> AssayVariantEffectMeasurementStudyResult:
+    return AssayVariantEffectMeasurementStudyResult(
+        id=mapped_variant.variant.urn,
+        sourceDataSet=score_set_to_data_set(mapped_variant.variant.score_set),
+        focusVariant=mapped_variant.post_mapped,
+        # This property of this column is guaranteed to be defined.
+        score=mapped_variant.variant.data["score"],  # type: ignore
+        specifiedBy=publication_identifiers_to_method(
+            mapped_variant.variant.score_set.publication_identifier_associations
+        ),
+        reportedIn=variant_to_document(mapped_variant.variant),
+        contributions=[mavedb_api_contribution(), mavedb_vrs_contribution(mapped_variant)],
+        # TODO: Is the score set more representative of the content, or the mapping?
+        dateAuthored=mapped_variant.mapped_date,
+    )
+
+
+def mapped_variant_to_functional_classification_statement(
+    mapped_variant: MappedVariant,
+) -> AssayVariantEffectFunctionalClassificationStatement:
+    return AssayVariantEffectFunctionalClassificationStatement(
+        id=mapped_variant.variant.urn,
+        subjectVariant=mapped_variant.post_mapped,
+        # TODO: What is going here? Example points to a MAVE minimum information document.
+        #       Maybe something from the keywords?
+        objectAssay="",
+        classification=functional_classification_of_variant(mapped_variant),
+        specifiedBy=variant_interpretation_functional_guideline_method(),
+        # TODO: Creation or modification?
+        # TODO: Is the score set more representative of the content, or the mapping?
+        dateAuthored=mapped_variant.variant.score_set.modification_date,
+        reportedIn=[
+            score_set_to_document(mapped_variant.variant.score_set),
+            variant_to_document(mapped_variant.variant),
+        ],
+        # TODO: Do we want user contributions?
+        # TODO: Is this misleading if we don't track all user contributions?
+        contributions=[mavedb_vrs_contribution(mapped_variant)],
+    )
+
+
+def mapped_variant_to_clinical_classification_statement(mapped_variant: MappedVariant):
+    return AssayVariantEffectClinicalClassificationStatement(
+        id=mapped_variant.variant.urn,
+        subjectVariant=mapped_variant.post_mapped,
+        # TODO: What is going here? Example points to a MAVE minimum information document.
+        #       Maybe something from the keywords?
+        objectAssay="",
+        classification=None,
+        specifiedBy=variant_interpretation_clinical_guideline_method(),
+        # TODO: Creation or modification?
+        # TODO: Is the score set more representative of the content, or the mapping?
+        dateAuthored=mapped_variant.variant.score_set.modification_date,
+        reportedIn=[
+            score_set_to_document(mapped_variant.variant.score_set),
+            variant_to_document(mapped_variant.variant),
+        ],
+        # TODO: Do we want user contributions?
+        # TODO: Is this misleading if we don't track all user contributions?
+        contributions=[mavedb_vrs_contribution(mapped_variant)],
+    )

src/mavedb/lib/annotation/variant.py

@@ -0,0 +1,19 @@
+from typing import TypedDict, Optional
+
+from ga4gh.va_spec.profiles import (
+    AssayVariantEffectMeasurementStudyResult,
+    AssayVariantEffectFunctionalClassificationStatement,
+    AssayVariantEffectClinicalClassificationStatement,
+)
+
+from mavedb.view_models.base.base import BaseModel
+
+
+class VariantAnnotationSpecification(TypedDict):
+    AssayVariantEffectMeasurementStudyResult: AssayVariantEffectMeasurementStudyResult
+    AssayVariantEffectFunctionalClassificationStatement: Optional[AssayVariantEffectFunctionalClassificationStatement]
+    AssayVariantEffectClinicalClassificationStatement: Optional[AssayVariantEffectClinicalClassificationStatement]
+
+
+class AnnotatedVariant(BaseModel):
+    __root__: dict[str, VariantAnnotationSpecification]