diff --git a/pori_python/ipr/constants.py b/pori_python/ipr/constants.py index 35c2a54..b694196 100644 --- a/pori_python/ipr/constants.py +++ b/pori_python/ipr/constants.py @@ -38,3 +38,4 @@ 'variantTypeName': 'moderate signature', }, } +HRD_SIGNATURE_OVER_CUTOFF = HRD_MAPPING['homologous recombination deficiency strong signature'] diff --git a/pori_python/ipr/content.spec.json b/pori_python/ipr/content.spec.json index a979012..5a1793a 100644 --- a/pori_python/ipr/content.spec.json +++ b/pori_python/ipr/content.spec.json @@ -548,12 +548,37 @@ }, "score": { "type": "number" + }, + "cutoff": { + "type": "number" } }, + "type": "object", "required": [ "score" ], - "type": "object" + "oneOf": [ + { + "required": [ + "kbCategory" + ], + "not": { + "required": [ + "cutoff" + ] + } + }, + { + "required": [ + "cutoff" + ], + "not": { + "required": [ + "kbCategory" + ] + } + } + ] }, "images": { "items": { diff --git a/pori_python/ipr/inputs.py b/pori_python/ipr/inputs.py index 0197660..f14fc69 100644 --- a/pori_python/ipr/inputs.py +++ b/pori_python/ipr/inputs.py @@ -27,6 +27,7 @@ HLA_SIGNATURE_VARIANT_TYPE, MSI_MAPPING, HRD_MAPPING, + HRD_SIGNATURE_OVER_CUTOFF, TMB_SIGNATURE, TMB_SIGNATURE_VARIANT_TYPE, ) @@ -558,13 +559,42 @@ def preprocess_hrd(hrd: Any) -> Iterable[Dict]: """ Process hrd input into preformatted signature input. HRD gets mapped to corresponding GraphKB Signature CategoryVariants. + + Either a cutoff or a kbcategory is expected. + If a cutoff is provided, the score is compared to the cutoff + to determine whether to create the signature variant. + If a kbCategory is provided, the signature variant is created based on the category. + If neither are provided, a warning is logged and no signature variant is created. """ if hrd: + hrd_cutoff = hrd.get('cutoff', None) hrd_cat = hrd.get('kbCategory', '') + hrd_score = hrd.get('score', None) - hrd_variant = HRD_MAPPING.get(hrd_cat, None) + if hrd_cutoff and hrd_cat: + raise ValueError( + 'In the HRD section, only one of cutoff and kbcategory should be provided.' + ) - # Signature CategoryVariant created either for msi or mss + if not (hrd_cutoff or hrd_cat): + logger.warning( + 'No hrd category or cutoff provided; score will be loaded with no variant matching.' + ) + + if hrd_cutoff: + if not hrd_score: + raise ValueError( + 'In the HRD section, if cutoff is provided a score must also be provided.' + ) + + if hrd_score >= hrd_cutoff: + hrd_variant = HRD_SIGNATURE_OVER_CUTOFF + else: + return [] + elif hrd_cat: + hrd_variant = HRD_MAPPING.get(hrd_cat, None) + + # Signature CategoryVariant created for hrd if hrd_variant: return [hrd_variant] diff --git a/pori_python/ipr/ipr.py b/pori_python/ipr/ipr.py index 06d9efb..f5cd687 100644 --- a/pori_python/ipr/ipr.py +++ b/pori_python/ipr/ipr.py @@ -668,7 +668,6 @@ def get_kb_disease_matches( verbose: bool = True, useSubgraphsRoute: bool = True, ) -> list[Dict]: - disease_matches = [] if not kb_disease_match: diff --git a/tests/test_ipr/test_inputs.py b/tests/test_ipr/test_inputs.py index 56e0c4d..519359c 100644 --- a/tests/test_ipr/test_inputs.py +++ b/tests/test_ipr/test_inputs.py @@ -292,6 +292,83 @@ def test_preprocess_hrd(self) -> None: signatureNames = {r.get('signatureName', '') for r in self.hrd} assert len(EXPECTED_HRD.symmetric_difference(signatureNames)) == 0 + def test_preprocess_hrd_cutoff_above(self) -> None: + """Test HRD with cutoff where score >= cutoff returns strong signature.""" + hrd = preprocess_hrd( + { + 'score': 75, + 'cutoff': 50, + } + ) + assert len(hrd) == 1 + assert hrd[0]['signatureName'] == 'homologous recombination deficiency' + assert hrd[0]['variantTypeName'] == 'strong signature' + + def test_preprocess_hrd_cutoff_below(self) -> None: + """Test HRD with cutoff where score < cutoff returns moderate signature.""" + hrd = preprocess_hrd( + { + 'score': 25, + 'cutoff': 50, + } + ) + assert len(hrd) == 0 + + def test_preprocess_hrd_cutoff_equal(self) -> None: + """Test HRD with cutoff where score == cutoff returns strong signature.""" + hrd = preprocess_hrd( + { + 'score': 50, + 'cutoff': 50, + } + ) + assert len(hrd) == 1 + assert hrd[0]['signatureName'] == 'homologous recombination deficiency' + assert hrd[0]['variantTypeName'] == 'strong signature' + + def test_preprocess_hrd_cutoff_missing_score(self) -> None: + """Test HRD with cutoff but missing score raises ValueError.""" + with pytest.raises(ValueError, match='if cutoff is provided a score must also be provided'): + preprocess_hrd( + { + 'cutoff': 50, + } + ) + + def test_preprocess_hrd_cutoff_and_kbcategory(self) -> None: + """Test HRD with both cutoff and kbCategory raises ValueError.""" + with pytest.raises( + ValueError, match='only one of cutoff and kbcategory should be provided' + ): + preprocess_hrd( + { + 'score': 75, + 'cutoff': 50, + 'kbCategory': 'homologous recombination deficiency strong signature', + } + ) + + def test_preprocess_hrd_kbcategory_moderate(self) -> None: + """Test HRD with kbCategory moderate signature.""" + hrd = preprocess_hrd( + { + 'kbCategory': 'homologous recombination deficiency moderate signature', + } + ) + assert len(hrd) == 1 + assert hrd[0]['signatureName'] == 'homologous recombination deficiency' + assert hrd[0]['variantTypeName'] == 'moderate signature' + + def test_preprocess_hrd_empty(self) -> None: + """Test HRD with empty input returns empty list.""" + hrd = preprocess_hrd({}) + assert hrd == [] + + def test_preprocess_hrd_none(self) -> None: + """Test HRD with None input returns empty list.""" + hrd = preprocess_hrd(None) + assert hrd == [] + def test_preprocess_signature_variants(self) -> None: records = preprocess_signature_variants( [ diff --git a/tests/test_ipr/test_upload.py b/tests/test_ipr/test_upload.py index 79568f7..2c6fb73 100644 --- a/tests/test_ipr/test_upload.py +++ b/tests/test_ipr/test_upload.py @@ -74,7 +74,7 @@ def loaded_reports(tmp_path_factory) -> Generator: ], 'hrd': { 'score': 9999.0, - 'kbCategory': 'homologous recombination deficiency strong signature', + 'cutoff': 5, }, 'expressionVariants': json.loads( pd.read_csv(get_test_file('expression.short.tab'), sep='\t').to_json(orient='records') @@ -106,6 +106,7 @@ def loaded_reports(tmp_path_factory) -> Generator: 'caption': 'Test adding a caption to an image', } ], + 'config': 'test config', } json_file.write_text( @@ -254,15 +255,30 @@ def test_copy_variants_loaded(self, loaded_reports) -> None: async_equals_sync = stringify_sorted(section) == stringify_sorted(async_section) assert async_equals_sync - # # Uncomment when signatureVariants are supported in pori_ipr_api - # def test_signature_variants_loaded(self, loaded_reports) -> None: - # section = get_section(loaded_reports["sync"], "signature-variants") - # kbmatched = [item for item in section if item["kbMatches"]] - # assert ("SBS2", "high signature") in [ - # (item["signatureName"], item["variantTypeName"]) for item in kbmatched - # ] - # async_section = get_section(loaded_reports["async"], "signature-variants") - # assert compare_sections(section, async_section) + def test_signature_variants_loaded(self, loaded_reports) -> None: + section = get_section(loaded_reports['sync'], 'signature-variants') + kbmatched = [item for item in section if item['kbMatches']] + # Check for COSMIC signatures + assert ('SBS2', 'high signature') in [ + (item['signatureName'], item['variantTypeName']) for item in kbmatched + ] + # Check for HRD signature (score 9999 > cutoff 5, so strong signature) + assert ('homologous recombination deficiency', 'strong signature') in [ + (item['signatureName'], item['variantTypeName']) for item in kbmatched + ] + # Check for MSI signature + assert ('microsatellite instability', 'high signature') in [ + (item['signatureName'], item['variantTypeName']) for item in kbmatched + ] + async_section = get_section(loaded_reports['async'], 'signature-variants') + async_equals_sync = stringify_sorted(section) == stringify_sorted(async_section) + assert async_equals_sync + + def test_hrd_score_in_report(self, loaded_reports) -> None: + """Test that HRD score is present in the loaded report.""" + report = loaded_reports['sync'][1]['reports'][0] + assert 'hrdScore' in report + assert report['hrdScore'] == 9999.0 def test_kb_matches_loaded(self, loaded_reports) -> None: section = get_section(loaded_reports['sync'], 'kb-matches')