feat: Add CIViC to CategoricalFusion object translator (#236)

Author: jarbesfeld

pyproject.toml

@@ -30,7 +30,8 @@ dependencies = [
     "ga4gh.vrs ==2.*",
     "biocommons.seqrepo",
     "gene-normalizer ~=0.9.0",
-    "cool-seq-tool ~=0.13.0",
+    "civicpy ~=4.0",
+    "cool-seq-tool ~=0.13.0"
 ]
 dynamic=["version"]
 

src/fusor/fusion_caller_models.py

@@ -4,6 +4,7 @@
 from enum import Enum
 from typing import Literal
 
+from civicpy.civic import ExonCoordinate, MolecularProfile
 from pydantic import BaseModel, ConfigDict, Field
 
 
@@ -21,13 +22,26 @@ class Caller(str, Enum):
     GENIE = "GENIE"
 
 
+class KnowledgebaseList(str, Enum):
+    """Define supported knowledgebases"""
+
+    CIVIC = "CIVIC"
+
+
 class FusionCaller(ABC, BaseModel):
     """ABC for fusion callers"""
 
     type: Caller
     model_config = ConfigDict(extra="allow")
 
 
+class FusionKnowledgebase(ABC, BaseModel):
+    """ABC for Fusion Knowledgebases"""
+
+    type: KnowledgebaseList
+    model_config = ConfigDict(extra="allow")
+
+
 class JAFFA(FusionCaller):
     """Define parameters for JAFFA model"""
 
@@ -247,3 +261,22 @@ class Genie(FusionCaller):
     reading_frame: str = Field(
         ..., description="The reading frame status of the fusion"
     )
+
+
+class CIVIC(FusionKnowledgebase):
+    """Define parameters for CIVIC model"""
+
+    type: Literal[KnowledgebaseList.CIVIC] = KnowledgebaseList.CIVIC
+    model_config = ConfigDict(arbitrary_types_allowed=True)
+    vicc_compliant_name: str = Field(
+        ..., description="The VICC compliant name for the fusion"
+    )
+    five_prime_end_exon_coords: ExonCoordinate | None = Field(
+        ..., description="Data for the end exon of 5' fusion partner"
+    )
+    three_prime_start_exon_coords: ExonCoordinate | None = Field(
+        ..., description="Data for the start exon 3' fusion partner"
+    )
+    molecular_profiles: list[MolecularProfile] | None = Field(
+        ..., description="The molecular profiles associated with the fusion"
+    )

src/fusor/fusor.py

@@ -5,6 +5,7 @@
 from typing import Annotated
 
 from bioutils.accessions import coerce_namespace
+from civicpy.civic import MolecularProfile
 from cool_seq_tool.app import CoolSeqTool
 from cool_seq_tool.schemas import CoordinateType, Strand
 from ga4gh.core import ga4gh_identify
@@ -126,6 +127,7 @@ def fusion(self, fusion_type: FusionType | None = None, **kwargs) -> Fusion:
             categorical_attributes = any(
                 [
                     "critical_functional_domains" in kwargs,
+                    "civic_molecular_profiles" in kwargs,
                     self._contains_element_type(
                         kwargs, StructuralElementType.MULTIPLE_POSSIBLE_GENES_ELEMENT
                     ),
@@ -162,6 +164,7 @@ def categorical_fusion(
         regulatory_element: RegulatoryElement | None = None,
         critical_functional_domains: list[FunctionalDomain] | None = None,
         reading_frame_preserved: bool | None = None,
+        civic_molecular_profiles: list[MolecularProfile] | None = None,
     ) -> CategoricalFusion:
         """Construct a categorical fusion object
 
@@ -170,6 +173,7 @@ def categorical_fusion(
         :param critical_functional_domains: lost or preserved functional domains
         :param reading_frame_preserved: ``True`` if reading frame is preserved.
             ``False`` otherwise
+        :param civic_molecular_profiles: A list of MolecularProfile objects
         :return: CategoricalFusion if construction successful
         :raise: FUSORParametersException if given incorrect fusion properties
         """
@@ -179,6 +183,7 @@ def categorical_fusion(
                 criticalFunctionalDomains=critical_functional_domains,
                 readingFramePreserved=reading_frame_preserved,
                 regulatoryElement=regulatory_element,
+                civic_molecular_profiles=civic_molecular_profiles,
             )
         except ValidationError as e:
             error_message = get_error_message(e)

src/fusor/harvester.py

@@ -5,7 +5,11 @@
 from pathlib import Path
 from typing import ClassVar
 
+from civicpy.civic import FusionVariant
+from pydantic import BaseModel, ConfigDict
+
 from fusor.fusion_caller_models import (
+    CIVIC,
     JAFFA,
     Arriba,
     Cicero,
@@ -162,3 +166,26 @@ class GenieHarvester(FusionCallerHarvester):
     }
     delimeter = "\t"
     fusion_caller = Genie
+
+
+class CIVICHarvester(BaseModel):
+    """Class for harvesting CIViC Fusion objects"""
+
+    fusions_list: list[FusionVariant]
+    model_config = ConfigDict(arbitrary_types_allowed=True)
+
+    def load_records(self) -> list[CIVIC]:
+        """Extract data from CIVIC fusion objects
+
+        :return A list of CIVIC objects
+        """
+        processed_fusions = []
+        for fusion in self.fusions_list:
+            params = {
+                "vicc_compliant_name": fusion.vicc_compliant_name,
+                "five_prime_end_exon_coords": fusion.five_prime_end_exon_coordinates,
+                "three_prime_start_exon_coords": fusion.three_prime_start_exon_coordinates,
+                "molecular_profiles": fusion.molecular_profiles,
+            }
+            processed_fusions.append(CIVIC(**params))
+        return processed_fusions

src/fusor/models.py

@@ -4,6 +4,7 @@
 from enum import Enum
 from typing import Annotated, Any, Literal
 
+from civicpy.civic import MolecularProfile
 from cool_seq_tool.schemas import Strand
 from ga4gh.core.models import MappableConcept
 from ga4gh.vrs.models import (
@@ -833,8 +834,10 @@ class CategoricalFusion(AbstractFusion):
     type: Literal[FUSORTypes.CATEGORICAL_FUSION] = FUSORTypes.CATEGORICAL_FUSION
     criticalFunctionalDomains: list[FunctionalDomain] | None = None
     structure: list[CategoricalFusionElement]
+    civicMolecularProfiles: list[MolecularProfile] | None = None
 
     model_config = ConfigDict(
+        arbitrary_types_allowed=True,
         json_schema_extra={
             "example": {
                 "type": "CategoricalFusion",