lint

Author: elewis2

pori_python/ipr/annotate.py

@@ -47,9 +47,7 @@ def get_second_pass_variants(
     }
 
     for reference1, variant_type in inferred_variants:
-        variants = gkb_match.match_category_variant(
-            graphkb_conn, reference1, variant_type
-        )
+        variants = gkb_match.match_category_variant(graphkb_conn, reference1, variant_type)
 
         for variant in variants:
             all_inferred_matches[variant["@rid"]] = variant
@@ -83,8 +81,7 @@ def get_ipr_statements_from_variants(
     inferred_statements = [
         s
         for s in get_statements_from_variants(graphkb_conn, inferred_matches)
-        if s["@rid"]
-        not in existing_statements  # do not duplicate if non-inferred match
+        if s["@rid"] not in existing_statements  # do not duplicate if non-inferred match
     ]
 
     for ipr_row in convert_statements_to_alterations(
@@ -129,9 +126,7 @@ def annotate_expression_variants(
             continue
         try:
             matches = gkb_match.match_expression_variant(graphkb_conn, gene, variant)
-            for ipr_row in get_ipr_statements_from_variants(
-                graphkb_conn, matches, disease_name
-            ):
+            for ipr_row in get_ipr_statements_from_variants(graphkb_conn, matches, disease_name):
                 ipr_row["variant"] = row["key"]
                 ipr_row["variantType"] = row.get("variantType", "exp")
                 alterations.append(ipr_row)
@@ -180,15 +175,11 @@ def annotate_copy_variants(
         if variant not in REPORTED_COPY_VARIANTS:
             # https://www.bcgsc.ca/jira/browse/GERO-77
             skipped += 1
-            logger.debug(
-                f"Dropping {gene} copy change '{variant}' - not in REPORTED_COPY_VARIANTS"
-            )
+            logger.debug(f"Dropping {gene} copy change '{variant}' - not in REPORTED_COPY_VARIANTS")
             continue
         try:
             matches = gkb_match.match_copy_variant(graphkb_conn, gene, variant)
-            for ipr_row in get_ipr_statements_from_variants(
-                graphkb_conn, matches, disease_name
-            ):
+            for ipr_row in get_ipr_statements_from_variants(graphkb_conn, matches, disease_name):
                 ipr_row["variant"] = row["key"]
                 ipr_row["variantType"] = row.get("variantType", "cnv")
                 alterations.append(ipr_row)
@@ -204,9 +195,7 @@ def annotate_copy_variants(
         )
     if problem_genes:
         logger.error(f"gene finding failures for copy variants {sorted(problem_genes)}")
-        logger.error(
-            f"gene finding failure for {len(problem_genes)} copy variant genes"
-        )
+        logger.error(f"gene finding failure for {len(problem_genes)} copy variant genes")
     logger.info(
         f"matched {len(variants)} copy category variants to {len(alterations)} graphkb annotations"
     )
@@ -264,9 +253,7 @@ def annotate_positional_variants(
                             f"Assuming malformed deletion variant {variant} is {variant[:-2] + 'del'}"
                         )
                         variant = variant[:-2] + "del"
-                        matches = gkb_match.match_positional_variant(
-                            graphkb_conn, variant
-                        )
+                        matches = gkb_match.match_positional_variant(graphkb_conn, variant)
                     else:
                         raise parse_err
 
@@ -299,9 +286,7 @@ def annotate_positional_variants(
 
     if problem_genes:
         logger.error(f"gene finding failures for {sorted(problem_genes)}")
-        logger.error(
-            f"{len(problem_genes)} gene finding failures for positional variants"
-        )
+        logger.error(f"{len(problem_genes)} gene finding failures for positional variants")
     if errors:
         logger.error(f"skipped {errors} positional variants due to errors")
 
@@ -349,9 +334,7 @@ def annotate_msi(
     )
     if msi_categories:
         msi_variants = [cast(Variant, var) for var in msi_categories]
-        for ipr_row in get_ipr_statements_from_variants(
-            graphkb_conn, msi_variants, disease_name
-        ):
+        for ipr_row in get_ipr_statements_from_variants(graphkb_conn, msi_variants, disease_name):
             ipr_row["variant"] = msi_category
             ipr_row["variantType"] = "msi"
             gkb_matches.append(ipr_row)
@@ -382,9 +365,7 @@ def annotate_tmb(
                 "filters": {
                     "reference1": {
                         "target": "Signature",
-                        "filters": {
-                            "OR": [{"name": category}, {"displayName": category}]
-                        },
+                        "filters": {"OR": [{"name": category}, {"displayName": category}]},
                     }
                 },
             },
@@ -394,9 +375,7 @@ def annotate_tmb(
     )
     if categories:
         cat_variants = [cast(Variant, var) for var in categories]
-        for ipr_row in get_ipr_statements_from_variants(
-            graphkb_conn, cat_variants, disease_name
-        ):
+        for ipr_row in get_ipr_statements_from_variants(graphkb_conn, cat_variants, disease_name):
             ipr_row["variant"] = category
             ipr_row["variantType"] = "tmb"
             gkb_matches.append(ipr_row)

pori_python/ipr/ipr.py

@@ -49,13 +49,9 @@ 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 = []
@@ -93,9 +89,7 @@ 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:
@@ -144,9 +138,7 @@ 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:
@@ -159,9 +151,7 @@ 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:
@@ -178,9 +168,7 @@ 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
@@ -201,12 +189,8 @@ 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:
@@ -216,16 +200,10 @@ def convert_statements_to_alterations(
                     "approvedTherapy": approved_therapy or False,
                     "category": ipr_section or "unknown",
                     "context": (
-                        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)
+                        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)),
                     "evidenceLevel": evidence_level_str or "",
                     "iprEvidenceLevel": ipr_evidence_levels_str or "",
                     "kbStatementId": statement["@rid"],
@@ -349,9 +327,7 @@ 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]
 
 
@@ -394,9 +370,7 @@ 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
@@ -473,19 +447,15 @@ 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
 
@@ -560,9 +530,7 @@ 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
     ]
 
 
@@ -608,9 +576,7 @@ def get_kb_matched_statements(
     for item in gkb_matches:
         stmt = copy(item)
         stmt["requiredKbMatches"].sort()
-        kbs = KbMatchedStatement(
-            {key: val for (key, val) in stmt.items() if key in kbs_keys}
-        )
+        kbs = KbMatchedStatement({key: val for (key, val) in stmt.items() if key in kbs_keys})
         dict_key = str(kbs)
         kbMatchedStatements[dict_key] = kbs
     return [*kbMatchedStatements.values()]
@@ -656,11 +622,7 @@ 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):
@@ -694,9 +656,7 @@ 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(
                 {