Feature/kbdev 1227 signature support

pori_python/graphkb/match.py

@@ -129,50 +129,62 @@ def cache_missing_features(conn: GraphKBConnection) -> None:
 
 def match_category_variant(
     conn: GraphKBConnection,
-    gene_name: str,
+    reference_name: str,
     category: str,
     root_exclude_term: str = "",
     gene_source: str = "",
     gene_is_source_id: bool = False,
     ignore_cache: bool = False,
+    reference_class: str = 'Feature',
 ) -> List[Variant]:
     """
     Returns a list of variants matching the input variant
 
     Args:
         conn (GraphKBConnection): the graphkb connection object
-        gene_name (str): the name of the gene the variant is in reference to
+        reference_name (str): the name of the Feature(gene)/Signature the variant is in reference to
         category (str): the variant category (ex. copy loss)
         gene_source: The source database the gene is defined by (ex. ensembl)
         gene_is_source_id: Indicates the gene name(s) input should be treated as sourceIds not names
+        reference_class (str): Class name of the variant reference. Default to 'Feature'
     Raises:
         FeatureNotFoundError: The gene could not be found in GraphKB
 
     Returns:
         Array.<dict>: List of variant records from GraphKB which match the input
     """
-    # disambiguate the gene to find all equivalent representations
-    features = convert_to_rid_list(
-        get_equivalent_features(
-            conn,
-            gene_name,
-            source=gene_source,
-            is_source_id=gene_is_source_id,
-            ignore_cache=ignore_cache,
+    # disambiguate the reference to find all equivalent representations
+    references: List[str] = []
+    if reference_class == 'Feature':
+        references = convert_to_rid_list(
+            get_equivalent_features(
+                conn,
+                reference_name,
+                source=gene_source,
+                is_source_id=gene_is_source_id,
+                ignore_cache=ignore_cache,
+            )
         )
-    )
-
-    if not features:
-        raise FeatureNotFoundError(
-            f"unable to find the gene ({gene_name}) or any equivalent representations"
+        if not references:
+            raise FeatureNotFoundError(
+                f"unable to find the gene ({reference_name}) or any equivalent representations"
+            )
+    if reference_class == 'Signature':
+        references = convert_to_rid_list(
+            get_equivalent_terms(
+                conn,
+                reference_name.lower(),
+                ontology_class='Signature',
+                ignore_cache=ignore_cache,
+            )
         )
 
     # get the list of terms that we should match
-    terms = convert_to_rid_list(
+    types = convert_to_rid_list(
         get_term_tree(conn, category, root_exclude_term, ignore_cache=ignore_cache)
     )
 
-    if not terms:
+    if not types:
         raise ValueError(f"unable to find the term/category ({category}) or any equivalent")
 
     # find the variant list
@@ -183,8 +195,8 @@ def match_category_variant(
                 "target": {
                     "target": "CategoryVariant",
                     "filters": [
-                        {"reference1": features, "operator": "IN"},
-                        {"type": terms, "operator": "IN"},
+                        {"reference1": references, "operator": "IN"},
+                        {"type": types, "operator": "IN"},
                     ],
                 },
                 "queryType": "similarTo",

pori_python/ipr/annotate.py

@@ -19,6 +19,7 @@
     Hashabledict,
     IprCopyVariant,
     IprExprVariant,
+    IprSignatureVariant,
     IprStructuralVariant,
     KbMatch,
     Statement,
@@ -380,3 +381,55 @@ def annotate_tmb(
             ipr_row["variantType"] = "tmb"
             gkb_matches.append(ipr_row)
     return gkb_matches
+
+
+def annotate_signature_variants(
+    graphkb_conn: GraphKBConnection,
+    variants: List[IprSignatureVariant] = [],
+    disease_name: str = "cancer",
+    show_progress: bool = False,
+) -> List[KbMatch]:
+    """Annotate Signature variants with GraphKB in the IPR alterations format.
+
+    Match to corresponding GraphKB Variants, then to linked GraphKB Statements
+
+    Args:
+        graphkb_conn: the graphkb api connection object
+        variants: list of signature variants
+        disease_name: oncotree disease name for graphkb matching
+        show_progress: progressbar displayed for long runs; default to False
+
+    Returns:
+        list of kbMatches records for IPR
+    """
+    alterations: List[Hashabledict] = []
+
+    iterfunc = tqdm if show_progress else iter
+    for variant in iterfunc(variants):
+        try:
+            # Matching signature variant to GKB Variants
+            matched_variants: List[Variant] = gkb_match.match_category_variant(
+                graphkb_conn,
+                variant["signatureName"],
+                variant["variantTypeName"],
+                reference_class="Signature",
+            )
+            # Matching GKB Variants to GKB Statements
+            for ipr_row in get_ipr_statements_from_variants(
+                graphkb_conn, matched_variants, disease_name
+            ):
+                ipr_row["variant"] = variant["key"]
+                ipr_row["variantType"] = "sigv"
+                alterations.append(Hashabledict(ipr_row))
+
+        except ValueError as err:
+            logger.error(f"failed to match signature category variant '{variant}': {err}")
+
+    # drop duplicates
+    alterations = list(set(alterations))
+
+    logger.info(
+        f"matched {len(variants)} signature category variants to {len(alterations)} graphkb annotations"
+    )
+
+    return alterations

pori_python/ipr/constants.py

@@ -5,5 +5,8 @@
 # all possible values for review status are: ['pending', 'not required', 'passed', 'failed', 'initial']
 FAILED_REVIEW_STATUS = "failed"
 
+# Signatures
+COSMIC_SIGNATURE_VARIANT_TYPE = "high signature"
+HLA_SIGNATURE_VARIANT_TYPE = "signature present"
 TMB_HIGH = 10.0  # genomic mutations per mb - https://www.bcgsc.ca/jira/browse/GERO-296
 TMB_HIGH_CATEGORY = "high mutation burden"

pori_python/ipr/content.spec.json

@@ -212,6 +212,16 @@
             },
             "type": "array"
         },
+        "cosmicSignatures": {
+            "default": [],
+            "description": "List of observed (above threshold) COSMIC signatures (DBS, SBS and ID) & dMMR",
+            "items": {
+                "description": "Signature name",
+                "example": "DBS1",
+                "type": "string"
+            },
+            "type": "array"
+        },
         "expressionVariants": {
             "default": [],
             "items": {

pori_python/ipr/inputs.py

@@ -16,11 +16,12 @@
     IprCopyVariant,
     IprExprVariant,
     IprFusionVariant,
+    IprSignatureVariant,
     IprSmallMutationVariant,
     IprVariant,
 )
 
-from .constants import DEFAULT_URL
+from .constants import COSMIC_SIGNATURE_VARIANT_TYPE, DEFAULT_URL, HLA_SIGNATURE_VARIANT_TYPE
 from .util import hash_key, logger, pandas_falsy
 
 protein_letters_3to1.setdefault("Ter", "*")
@@ -152,6 +153,12 @@
     "mavis_product_id",
 ]
 
+SIGV_REQ = ["signatureName", "variantTypeName"]
+SIGV_COSMIC = ["signature"]  # 1st element used as signatureName key
+SIGV_HLA = ["a1", "a2", "b1", "b2", "c1", "c2"]
+SIGV_OPTIONAL = ["displayName"]
+SIGV_KEY = SIGV_REQ[:]
+
 
 def validate_variant_rows(
     rows: Iterable[Dict], required: List[str], optional: List[str], row_to_key: Callable
@@ -388,6 +395,69 @@ def row_key(row: Dict) -> Tuple[str, ...]:
     return result
 
 
+def preprocess_signature_variants(rows: Iterable[Dict]) -> List[IprSignatureVariant]:
+    """
+    Validate the input rows contain the minimum required fields and
+    generate any default values where possible
+    """
+
+    def row_key(row: Dict) -> Tuple[str, ...]:
+        return tuple(["sigv"] + [row[key] for key in SIGV_KEY])
+
+    variants = validate_variant_rows(rows, SIGV_REQ, SIGV_OPTIONAL, row_key)
+    result = [cast(IprSignatureVariant, var) for var in variants]
+
+    # Adding additional required properties
+    for row in result:
+        row["variant"] = row["displayName"]
+        row["variantType"] = "sigv"
+
+    return result
+
+
+def preprocess_cosmic(rows: Iterable[Dict]) -> Iterable[Dict]:
+    """
+    Process cosmic inputs into preformatted signature inputs
+    Note: Cosmic and dMMR already evaluated against thresholds in gsc_report
+    """
+    cosmic = set()
+    for row in rows:
+        if not set(SIGV_COSMIC).issubset(row.keys()):
+            continue
+        cosmic.add(row[SIGV_COSMIC[0]])
+
+    return [
+        {
+            "displayName": f"{signature} {COSMIC_SIGNATURE_VARIANT_TYPE}",
+            "signatureName": signature,
+            "variantTypeName": COSMIC_SIGNATURE_VARIANT_TYPE,
+        }
+        for signature in cosmic
+    ]
+
+
+def preprocess_hla(rows: Iterable[Dict]) -> Iterable[Dict]:
+    """
+    Process hla inputs into preformatted signature inputs
+    """
+    hla: Set[str] = set()
+    for row in rows:  # 1 row per sample; should be 3
+        for k, v in row.items():
+            if k not in SIGV_HLA:
+                continue
+            hla.add(f"HLA-{v}")  # 2nd level, e.g. 'HLA-A*02:01'
+            hla.add(f"HLA-{v.split(':')[0]}")  # 1st level, e.g. 'HLA-A*02'
+
+    return [
+        {
+            "displayName": f"{signature} {HLA_SIGNATURE_VARIANT_TYPE}",
+            "signatureName": signature,
+            "variantTypeName": HLA_SIGNATURE_VARIANT_TYPE,
+        }
+        for signature in hla
+    ]
+
+
 def check_variant_links(
     small_mutations: List[IprSmallMutationVariant],
     expression_variants: List[IprExprVariant],