lint

Author: elewis2

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"],
@@ -288,9 +266,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]
 
 
@@ -333,9 +309,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
@@ -412,19 +386,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
 
@@ -499,9 +469,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
     ]
 
 
@@ -547,9 +515,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()]
@@ -591,11 +557,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"]:
@@ -616,9 +578,7 @@ def get_kb_statement_matched_conditions(
 
         # remove empty sets from requirements if allowing partial matches
         if allow_partial_matches:
-            requirements = {
-                key: val for (key, val) in requirements.items() if len(val) > 0
-            }
+            requirements = {key: val for (key, val) in requirements.items() if len(val) > 0}
 
         variantConditionSets = list(product(*requirements.values()))
         conditionSets = [

pori_python/ipr/main.py

@@ -53,9 +53,7 @@
 
 def file_path(path: str) -> str:
     if not os.path.exists(path):
-        raise argparse.ArgumentTypeError(
-            f"{repr(path)} is not a valid filename. does not exist"
-        )
+        raise argparse.ArgumentTypeError(f"{repr(path)} is not a valid filename. does not exist")
     return path
 
 
@@ -72,9 +70,7 @@ def command_interface() -> None:
         default=os.environ.get("USER"),
         help="username to use connecting to graphkb/ipr",
     )
-    req.add_argument(
-        "--password", required=True, help="password to use connecting to graphkb/ipr"
-    )
+    req.add_argument("--password", required=True, help="password to use connecting to graphkb/ipr")
     req.add_argument(
         "-c", "--content", required=True, type=file_path, help="Report Content as JSON"
     )
@@ -168,9 +164,7 @@ def clean_unsupported_content(upload_content: Dict, ipr_spec: Dict = {}) -> Dict
         and "genesCreate" in ipr_spec["components"]["schemas"].keys()
         and "properties" in ipr_spec["components"]["schemas"]["genesCreate"].keys()
     ):
-        genes_spec = ipr_spec["components"]["schemas"]["genesCreate"][
-            "properties"
-        ].keys()
+        genes_spec = ipr_spec["components"]["schemas"]["genesCreate"]["properties"].keys()
 
         # check what ipr report upload expects and adjust contents to match
         for old_name, new_name in RENAMED_GENE_PROPERTIES.items():
@@ -205,9 +199,7 @@ def clean_unsupported_content(upload_content: Dict, ipr_spec: Dict = {}) -> Dict
                     removed_keys[key] = 1
                 gene.pop(key)
         for key, count in removed_keys.items():
-            logger.warning(
-                f"IPR unsupported property '{key}' removed from {count} genes."
-            )
+            logger.warning(f"IPR unsupported property '{key}' removed from {count} genes.")
 
     drop_columns = ["variant", "variantType", "histogramImage"]
     # DEVSU-2034 - use a 'displayName'
@@ -224,9 +216,7 @@ def clean_unsupported_content(upload_content: Dict, ipr_spec: Dict = {}) -> Dict
         for variant in upload_content.get(variant_list_section, []):
             if not variant.get("displayName"):
                 variant["displayName"] = (
-                    variant.get("variant")
-                    or variant.get("kbCategory")
-                    or variant.get("key", "")
+                    variant.get("variant") or variant.get("kbCategory") or variant.get("key", "")
                 )
             if variant_list_section == "probeResults":
                 # currently probeResults will error if they do NOT have a 'variant' column.
@@ -255,9 +245,7 @@ def clean_unsupported_content(upload_content: Dict, ipr_spec: Dict = {}) -> Dict
 
 
 def create_report(**kwargs) -> Dict:
-    logger.warning(
-        "Deprecated function 'create_report' called - use ipr_report instead"
-    )
+    logger.warning("Deprecated function 'create_report' called - use ipr_report instead")
     return ipr_report(**kwargs)
 
 
@@ -329,22 +317,16 @@ def ipr_report(
     try:
         validate_report_content(content)
     except jsonschema.exceptions.ValidationError as err:
-        logger.error(
-            "Failed schema check - report variants may be corrupted or unmatched."
-        )
+        logger.error("Failed schema check - report variants may be corrupted or unmatched.")
         logger.error(f"Failed schema check: {err}")
 
     kb_disease_match = content["kbDiseaseMatch"]
 
     # validate the input variants
     small_mutations = preprocess_small_mutations(content.get("smallMutations", []))
-    structural_variants = preprocess_structural_variants(
-        content.get("structuralVariants", [])
-    )
+    structural_variants = preprocess_structural_variants(content.get("structuralVariants", []))
     copy_variants = preprocess_copy_variants(content.get("copyVariants", []))
-    expression_variants = preprocess_expression_variants(
-        content.get("expressionVariants", [])
-    )
+    expression_variants = preprocess_expression_variants(content.get("expressionVariants", []))
     if expression_variants:
         check_comparators(content, expression_variants)
 
@@ -393,9 +375,7 @@ def ipr_report(
                 tmb["kbCategory"] = TMB_HIGH_CATEGORY
 
             # GERO-296 - try matching to graphkb
-            tmb_matches = annotate_tmb(
-                graphkb_conn, kb_disease_match, TMB_HIGH_CATEGORY
-            )
+            tmb_matches = annotate_tmb(graphkb_conn, kb_disease_match, TMB_HIGH_CATEGORY)
             if tmb_matches:
                 tmb_variant["kbCategory"] = TMB_HIGH_CATEGORY  # type: ignore
                 tmb_variant["variant"] = TMB_HIGH_CATEGORY
@@ -404,9 +384,7 @@ def ipr_report(
                 logger.info(
                     f"GERO-296 '{TMB_HIGH_CATEGORY}' matches {len(tmb_matches)} statements."
                 )
-                gkb_matches.extend(
-                    [Hashabledict(tmb_statement) for tmb_statement in tmb_matches]
-                )
+                gkb_matches.extend([Hashabledict(tmb_statement) for tmb_statement in tmb_matches])
                 logger.debug(f"\tgkb_matches: {len(gkb_matches)}")
 
     # MATCHING MSI
@@ -429,9 +407,7 @@ def ipr_report(
             msi_variant["variant"] = msi_cat
             msi_variant["key"] = msi_cat
             msi_variant["variantType"] = "msi"
-            logger.info(
-                f"GERO-295 '{msi_cat}' matches {len(msi_matches)} msi statements."
-            )
+            logger.info(f"GERO-295 '{msi_cat}' matches {len(msi_matches)} msi statements.")
             gkb_matches.extend([Hashabledict(msi) for msi in msi_matches])
             logger.debug(f"\tgkb_matches: {len(gkb_matches)}")
 
@@ -496,20 +472,15 @@ def ipr_report(
         # verify germline kb statements matched germline observed variants, not somatic variants
         org_len = len(gkb_matches)
         gkb_matches = [
-            Hashabledict(match)
-            for match in germline_kb_matches(gkb_matches, all_variants)
+            Hashabledict(match) for match in germline_kb_matches(gkb_matches, all_variants)
         ]
         num_removed = org_len - len(gkb_matches)
         if num_removed:
-            logger.info(
-                f"Removing {num_removed} germline events without medical matches."
-            )
+            logger.info(f"Removing {num_removed} germline events without medical matches.")
 
     if custom_kb_match_filter:
         logger.info(f"custom_kb_match_filter on {len(gkb_matches)} variants")
-        gkb_matches = [
-            Hashabledict(match) for match in custom_kb_match_filter(gkb_matches)
-        ]
+        gkb_matches = [Hashabledict(match) for match in custom_kb_match_filter(gkb_matches)]
         logger.info(f"\t custom_kb_match_filter left {len(gkb_matches)} variants")
 
     # TODO: can probably be removed with change to kbmatch processing, but double check
@@ -576,9 +547,7 @@ def ipr_report(
             # remove after testing
             # "kbMatches": [trim_empty_values(a) for a in gkb_matches],  # type: ignore
             "copyVariants": [
-                trim_empty_values(c)
-                for c in copy_variants
-                if c["gene"] in genes_with_variants
+                trim_empty_values(c) for c in copy_variants if c["gene"] in genes_with_variants
             ],
             "smallMutations": [trim_empty_values(s) for s in small_mutations],
             "expressionVariants": [
@@ -602,9 +571,7 @@ def ipr_report(
             "therapeuticTarget": targets,
         }
     )
-    output.setdefault("images", []).extend(
-        select_expression_plots(gkb_matches, all_variants)
-    )
+    output.setdefault("images", []).extend(select_expression_plots(gkb_matches, all_variants))
 
     output = clean_unsupported_content(output, ipr_spec)
     ipr_result = None