minor cleanup

Author: elewis2

pori_python/ipr/ipr.py

@@ -49,9 +49,13 @@ def filter_structural_variants(
     Filter structural variants to remove non-high quality events unless they are matched/annotated or
     they involve a gene that is a known fusion partner
     """
-    matched_svs = {match["variant"] for match in kb_matches if match["variantType"] == "sv"}
+    matched_svs = {
+        match["variant"] for match in kb_matches if match["variantType"] == "sv"
+    }
     fusion_genes = {
-        gene["name"] for gene in gene_annotations if gene.get("knownFusionPartner", False)
+        gene["name"]
+        for gene in gene_annotations
+        if gene.get("knownFusionPartner", False)
     }
 
     result = []
@@ -89,7 +93,9 @@ def get_evidencelevel_mapping(graphkb_conn: GraphKBConnection) -> Dict[str, str]
 
     # Filter IPR EvidenceLevel and map each outgoing CrossReferenceOf to displayName
     ipr_source_rid = graphkb_conn.get_source("ipr")["@rid"]
-    ipr_evidence_levels = filter(lambda d: d.get("source") == ipr_source_rid, evidence_levels)
+    ipr_evidence_levels = filter(
+        lambda d: d.get("source") == ipr_source_rid, evidence_levels
+    )
     cross_references_mapping: Dict[str, str] = dict()
     ipr_rids_to_displayname: Dict[str, str] = dict()
     for level in ipr_evidence_levels:
@@ -138,7 +144,9 @@ def convert_statements_to_alterations(
     """
     disease_matches = {
         r["@rid"]
-        for r in gkb_vocab.get_term_tree(graphkb_conn, disease_name, ontology_class="Disease")
+        for r in gkb_vocab.get_term_tree(
+            graphkb_conn, disease_name, ontology_class="Disease"
+        )
     }
 
     if not disease_matches:
@@ -151,7 +159,9 @@ def convert_statements_to_alterations(
 
     # get the recruitment status for any trial associated with a statement
     clinical_trials = [
-        s["subject"]["@rid"] for s in statements if s["subject"]["@class"] == "ClinicalTrial"
+        s["subject"]["@rid"]
+        for s in statements
+        if s["subject"]["@class"] == "ClinicalTrial"
     ]
     recruitment_statuses = {}
     if clinical_trials:
@@ -168,7 +178,9 @@ def convert_statements_to_alterations(
 
     for statement in statements:
         variants = [
-            cast(Variant, c) for c in statement["conditions"] if c["@class"] in VARIANT_CLASSES
+            cast(Variant, c)
+            for c in statement["conditions"]
+            if c["@class"] in VARIANT_CLASSES
         ]
         diseases = [c for c in statement["conditions"] if c["@class"] == "Disease"]
         disease_match = len(diseases) == 1 and diseases[0]["@rid"] in disease_matches
@@ -189,8 +201,12 @@ def convert_statements_to_alterations(
 
         evidence_level_str = display_evidence_levels(statement)
         evidence_levels = statement.get("evidenceLevel") or []
-        ipr_evidence_levels = [ev_map[el.get("@rid", "")] for el in evidence_levels if el]
-        ipr_evidence_levels_str = ";".join(sorted(set([el for el in ipr_evidence_levels])))
+        ipr_evidence_levels = [
+            ev_map[el.get("@rid", "")] for el in evidence_levels if el
+        ]
+        ipr_evidence_levels_str = ";".join(
+            sorted(set([el for el in ipr_evidence_levels]))
+        )
 
         for variant in variants:
             if variant["@rid"] not in variant_matches:
@@ -200,10 +216,16 @@ def convert_statements_to_alterations(
                     "approvedTherapy": approved_therapy or False,
                     "category": ipr_section or "unknown",
                     "context": (
-                        statement["subject"]["displayName"] if statement["subject"] else ""
+                        statement["subject"]["displayName"]
+                        if statement["subject"]
+                        else ""
+                    ),
+                    "kbContextId": (
+                        statement["subject"]["@rid"] if statement["subject"] else ""
+                    ),
+                    "disease": ";".join(
+                        sorted(d.get("displayName", "") for d in diseases)
                     ),
-                    "kbContextId": (statement["subject"]["@rid"] if statement["subject"] else ""),
-                    "disease": ";".join(sorted(d.get("displayName", "") for d in diseases)),
                     "evidenceLevel": evidence_level_str or "",
                     "iprEvidenceLevel": ipr_evidence_levels_str or "",
                     "kbStatementId": statement["@rid"],
@@ -327,7 +349,9 @@ def select_expression_plots(
         gene = str(variant.get("gene", ""))
         hist = str(variant.get("histogramImage", ""))
         if hist:
-            images_by_gene[gene] = ImageDefinition({"key": f"expDensity.{gene}", "path": hist})
+            images_by_gene[gene] = ImageDefinition(
+                {"key": f"expDensity.{gene}", "path": hist}
+            )
     return [images_by_gene[gene] for gene in selected_genes if gene in images_by_gene]
 
 
@@ -370,7 +394,9 @@ def create_key_alterations(
         counts[type_mapping[variant_type]].add(variant_key)
 
         if variant_type == "exp":
-            alterations.append(f'{variant.get("gene","")} ({variant.get("expressionState")})')
+            alterations.append(
+                f'{variant.get("gene","")} ({variant.get("expressionState")})'
+            )
         elif variant_type == "cnv":
             alterations.append(f'{variant.get("gene","")} ({variant.get("cnvState")})')
         # only show germline if relevant
@@ -447,15 +473,19 @@ def germline_kb_matches(
         # Remove any matches to germline events
         for alt in somatic_alts:
             var_list = [v for v in all_variants if v["key"] == alt["variant"]]
-            somatic_var_list = [v for v in var_list if not v.get("germline", not assume_somatic)]
+            somatic_var_list = [
+                v for v in var_list if not v.get("germline", not assume_somatic)
+            ]
             if var_list and not somatic_var_list:
                 logger.debug(
                     f"Dropping germline match to somatic statement kbStatementId:{alt['kbStatementId']}: {alt['kbVariant']} {alt['category']}"
                 )
             elif somatic_var_list:
                 ret_list.append(alt)  # match to somatic variant
             else:
-                ret_list.append(alt)  # alteration not in any specific keys matches to check.
+                ret_list.append(
+                    alt
+                )  # alteration not in any specific keys matches to check.
 
     return ret_list
 
@@ -530,11 +560,15 @@ def multi_variant_filtering(
 
     # Filtering out incompleted matches of gkb_matches
     return [
-        match for match in gkb_matches if match["kbStatementId"] in complete_matching_statements
+        match
+        for match in gkb_matches
+        if match["kbStatementId"] in complete_matching_statements
     ]
 
 
-def get_kb_variants(gkb_matches: List[KbMatch] | List[Hashabledict]) -> List[KbVariantMatch]:
+def get_kb_variants(
+    gkb_matches: List[KbMatch] | List[Hashabledict],
+) -> List[KbVariantMatch]:
     """Extracts the set of distinct kb variant records from the input
     list of gkb_matches records, which combine statement and variant matches.
 
@@ -572,17 +606,12 @@ def get_kb_matched_statements(
     kbMatchedStatements = {}
     kbs_keys = KbMatchedStatement.__annotations__.keys()
     for item in gkb_matches:
-        x = copy(item)
-        x['requiredKbMatches'].sort()
-        kbs = KbMatchedStatement({key: val for (key, val) in x.items() if key in kbs_keys})
-        # import pdb; pdb.set_trace()
-        # kbs = {key: val for (key, val) in kbs.items() if key in kbs_keys}
-        # import pdb; pdb.set_trace()
-        # sort list to ensure there are no extra string representations of kbs
-        # TODO add test for this
-        # kbs['requiredKbMatches'].sort()
+        stmt = copy(item)
+        stmt["requiredKbMatches"].sort()
+        kbs = KbMatchedStatement(
+            {key: val for (key, val) in stmt.items() if key in kbs_keys}
+        )
         dict_key = str(kbs)
-        # kbs = cast(KbMatchedStatement, kbs)
         kbMatchedStatements[dict_key] = kbs
     return [*kbMatchedStatements.values()]
 
@@ -627,7 +656,11 @@ def get_kb_statement_matched_conditions(
     kbMatchedStatementConditions = {}
 
     for kbStmt in kbMatchedStatements:
-        stmts = [item for item in gkb_matches if item["kbStatementId"] == kbStmt["kbStatementId"]]
+        stmts = [
+            item
+            for item in gkb_matches
+            if item["kbStatementId"] == kbStmt["kbStatementId"]
+        ]
         requirements = {}
         for requirement in stmts[0]["requiredKbMatches"]:
             if not requirements.get(requirement, False):
@@ -645,7 +678,7 @@ def get_kb_statement_matched_conditions(
                         for item in gkb_matches
                         if (
                             item["kbVariantId"] == requirement
-                            and item['kbStatementId'] == kbStmt['kbStatementId']
+                            and item["kbStatementId"] == kbStmt["kbStatementId"]
                         )
                     ]
                 requirements[requirement] = reqlist
@@ -661,7 +694,9 @@ def get_kb_statement_matched_conditions(
         ]
         for conditionSet in conditionSets:
             # remove Nones
-            observedVariantKeys = [item for item in conditionSet["observedVariantKeys"] if item]
+            observedVariantKeys = [
+                item for item in conditionSet["observedVariantKeys"] if item
+            ]
             observedVariantKeys.sort()
             kbmc = KbMatchedStatementConditionSet(
                 {