Feat/devsu 2885 accept hrd cutoff

pori_python/ipr/constants.py

@@ -38,3 +38,4 @@
         'variantTypeName': 'moderate signature',
     },
 }
+HRD_SIGNATURE_OVER_CUTOFF = HRD_MAPPING['homologous recombination deficiency strong signature']

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": {

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]
 

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:

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(
             [

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')