Refactor analysistype 3 (#3982)

### Changed

- Removed AnalysisType from constants.py
- Moved AnalysisType to tb.py
- Renamed AnalysisTypes to AnalysisType

cg/apps/demultiplex/demultiplex_api.py

@@ -13,7 +13,7 @@
 from cg.constants.constants import FileFormat, Workflow
 from cg.constants.demultiplexing import DemultiplexingDirsAndFiles
 from cg.constants.priority import SlurmQos
-from cg.constants.tb import AnalysisTypes
+from cg.constants.tb import AnalysisType
 from cg.exc import HousekeeperFileMissingError
 from cg.io.controller import WriteFile
 from cg.models.demultiplex.sbatch import SbatchCommand, SbatchError
@@ -210,7 +210,7 @@ def add_to_trailblazer(
         )
         tb_api.add_pending_analysis(
             case_id=sequencing_run.id,
-            analysis_type=AnalysisTypes.OTHER,
+            analysis_type=AnalysisType.OTHER,
             config_path=sequencing_run.trailblazer_config_path.as_posix(),
             out_dir=sequencing_run.trailblazer_config_path.parent.as_posix(),
             slurm_quality_of_service=self.slurm_quality_of_service,

cg/cli/workflow/raw_data/raw_data_service.py

@@ -1,8 +1,8 @@
 import datetime as dt
 
 from cg.apps.tb.api import TrailblazerAPI
-from cg.constants.constants import AnalysisType, Workflow
-from cg.constants.tb import AnalysisStatus
+from cg.constants.constants import Workflow
+from cg.constants.tb import AnalysisStatus, AnalysisType
 from cg.exc import CaseNotFoundError
 from cg.store.models import Analysis, Case
 from cg.store.store import Store

cg/constants/constants.py

@@ -86,14 +86,6 @@ def statuses(cls) -> list[str]:
         return list(map(lambda status: status.value, cls))
 
 
-class AnalysisType(StrEnum):
-    TARGETED_GENOME_SEQUENCING: str = "tgs"
-    WHOLE_EXOME_SEQUENCING: str = "wes"
-    WHOLE_GENOME_SEQUENCING: str = "wgs"
-    WHOLE_TRANSCRIPTOME_SEQUENCING: str = "wts"
-    OTHER: str = "other"
-
-
 class CancerAnalysisType(StrEnum):
     TUMOR_NORMAL = auto()
     TUMOR_NORMAL_PANEL = auto()

cg/constants/tb.py

@@ -14,7 +14,7 @@ class AnalysisStatus:
     QC: str = "qc"
 
 
-class AnalysisTypes(StrEnum):
+class AnalysisType(StrEnum):
     OTHER: str = "other"
     TGS: str = SeqLibraryPrepCategory.TARGETED_GENOME_SEQUENCING
     WES: str = SeqLibraryPrepCategory.WHOLE_EXOME_SEQUENCING

cg/meta/delivery_report/balsamic.py

@@ -17,8 +17,8 @@
     REQUIRED_SAMPLE_TIMESTAMP_FIELDS,
     Workflow,
 )
-from cg.constants.constants import AnalysisType
 from cg.constants.scout import ScoutUploadKey
+from cg.constants.tb import AnalysisType
 from cg.meta.delivery_report.data_validators import get_million_read_pairs
 from cg.meta.delivery_report.delivery_report_api import DeliveryReportAPI
 from cg.meta.workflow.balsamic import BalsamicAnalysisAPI
@@ -52,9 +52,7 @@ def get_sample_metadata(
         passed_initial_qc: bool | None = self.lims_api.has_sample_passed_initial_qc(
             sample.internal_id
         )
-        if AnalysisType.WHOLE_GENOME_SEQUENCING in self.analysis_api.get_data_analysis_type(
-            case.internal_id
-        ):
+        if AnalysisType.WGS in self.analysis_api.get_data_analysis_type(case.internal_id):
             return self.get_wgs_metadata(
                 million_read_pairs=million_read_pairs,
                 passed_initial_qc=passed_initial_qc,

cg/meta/upload/scout/balsamic_config_builder.py

@@ -5,12 +5,7 @@
 from cg.apps.lims import LimsAPI
 from cg.constants.constants import SampleType
 from cg.constants.housekeeper_tags import HK_DELIVERY_REPORT_TAG
-from cg.constants.scout import (
-    BALSAMIC_CASE_TAGS,
-    BALSAMIC_SAMPLE_TAGS,
-    GenomeBuild,
-    UploadTrack,
-)
+from cg.constants.scout import BALSAMIC_CASE_TAGS, BALSAMIC_SAMPLE_TAGS, GenomeBuild, UploadTrack
 from cg.constants.subject import PhenotypeStatus
 from cg.meta.upload.scout.hk_tags import CaseTags, SampleTags
 from cg.meta.upload.scout.scout_config_builder import ScoutConfigBuilder
@@ -71,7 +66,7 @@ def build_config_sample(self, case_sample: CaseSample) -> ScoutCancerIndividual:
     def get_balsamic_analysis_type(self, sample: Sample) -> str:
         """Returns a formatted balsamic analysis type"""
 
-        analysis_type: str = BalsamicAnalysisAPI.get_application_type(sample_obj=sample)
+        analysis_type: str = BalsamicAnalysisAPI.get_analysis_type(sample=sample)
         if analysis_type == "tgs":
             analysis_type = "panel"
         if analysis_type == "wgs":

cg/meta/workflow/analysis.py

@@ -13,7 +13,6 @@
 from cg.clients.chanjo2.models import CoverageMetrics
 from cg.constants import EXIT_FAIL, EXIT_SUCCESS, Priority, SequencingFileTag, Workflow
 from cg.constants.constants import (
-    AnalysisType,
     CaseActions,
     CustomerId,
     FileFormat,
@@ -23,7 +22,8 @@
 from cg.constants.gene_panel import GenePanelCombo, GenePanelMasterList
 from cg.constants.priority import SlurmQos
 from cg.constants.scout import HGNC_ID, ScoutExportFileName
-from cg.constants.tb import AnalysisStatus
+from cg.constants.sequencing import SeqLibraryPrepCategory
+from cg.constants.tb import AnalysisStatus, AnalysisType
 from cg.exc import AnalysisNotReadyError, BundleAlreadyAddedError, CgDataError, CgError
 from cg.io.controller import WriteFile
 from cg.meta.archive.archive import SpringArchiveAPI
@@ -168,25 +168,26 @@ def get_bundle_deliverables_type(self, case_id: str) -> str | None:
         return None
 
     @staticmethod
-    def get_application_type(sample_obj: Sample) -> str:
+    def get_analysis_type(sample: Sample) -> str:
         """
-        Gets application type for sample. Only application types supported by trailblazer (or other)
+        Return the analysis type for sample.
+        Only analysis types supported by Trailblazer
         are valid outputs.
         """
-        prep_category: str = sample_obj.prep_category
+        prep_category: str = sample.prep_category
         if prep_category and prep_category.lower() in {
-            AnalysisType.TARGETED_GENOME_SEQUENCING,
-            AnalysisType.WHOLE_EXOME_SEQUENCING,
-            AnalysisType.WHOLE_GENOME_SEQUENCING,
-            AnalysisType.WHOLE_TRANSCRIPTOME_SEQUENCING,
+            SeqLibraryPrepCategory.TARGETED_GENOME_SEQUENCING,
+            SeqLibraryPrepCategory.WHOLE_EXOME_SEQUENCING,
+            SeqLibraryPrepCategory.WHOLE_GENOME_SEQUENCING,
+            SeqLibraryPrepCategory.WHOLE_TRANSCRIPTOME_SEQUENCING,
         }:
             return prep_category.lower()
         return AnalysisType.OTHER
 
     def get_case_application_type(self, case_id: str) -> str:
         """Returns the application type for samples in a case."""
         samples: list[Sample] = self.status_db.get_samples_by_case_id(case_id)
-        application_types: set[str] = {self.get_application_type(sample) for sample in samples}
+        application_types: set[str] = {self.get_analysis_type(sample) for sample in samples}
 
         if len(application_types) > 1:
             raise CgError(
@@ -197,7 +198,7 @@ def get_case_application_type(self, case_id: str) -> str:
 
     def are_case_samples_rna(self, case_id: str) -> bool:
         analysis_type: str = self.get_case_application_type(case_id)
-        return analysis_type == AnalysisType.WHOLE_TRANSCRIPTOME_SEQUENCING
+        return analysis_type == AnalysisType.WTS
 
     def get_case_source_type(self, case_id: str) -> str | None:
         """
@@ -217,7 +218,7 @@ def get_case_source_type(self, case_id: str) -> str | None:
     def has_case_only_exome_samples(self, case_id: str) -> bool:
         """Returns True if the application type for all samples in a case is WHOLE_EXOME_SEQUENCING."""
         application_type: str = self.get_case_application_type(case_id)
-        return application_type == AnalysisType.WHOLE_EXOME_SEQUENCING
+        return application_type == AnalysisType.WES
 
     def upload_bundle_housekeeper(
         self, case_id: str, dry_run: bool = False, force: bool = False
@@ -282,7 +283,7 @@ def add_pending_trailblazer_analysis(
         tower_workflow_id: str | None = None,
     ) -> None:
         self.check_analysis_ongoing(case_id)
-        application_type: str = self.get_application_type(
+        analysis_type: str = self.get_analysis_type(
             self.status_db.get_case_by_internal_id(case_id).links[0].sample
         )
         config_path: str = self.get_job_ids_path(case_id).as_posix()
@@ -295,7 +296,7 @@ def add_pending_trailblazer_analysis(
         workflow_manager: str = self.get_workflow_manager()
         is_case_for_development: bool = self._is_case_for_development(case_id)
         self.trailblazer_api.add_pending_analysis(
-            analysis_type=application_type,
+            analysis_type=analysis_type,
             case_id=case_id,
             config_path=config_path,
             email=email,

cg/meta/workflow/balsamic.py

@@ -134,7 +134,7 @@ def get_bundle_deliverables_type(self, case_id: str) -> str:
             self.status_db.get_case_by_internal_id(internal_id=case_id).links
         )
 
-        application_type: str = self.get_application_type(
+        application_type: str = self.get_analysis_type(
             self.status_db.get_case_by_internal_id(internal_id=case_id).links[0].sample
         )
         sample_type = "tumor"
@@ -500,7 +500,7 @@ def get_sample_params(self, case_id: str, panel_bed: str | None) -> dict:
             link_object.sample.internal_id: {
                 "sex": link_object.sample.sex,
                 "tissue_type": self.get_sample_type(link_object.sample).value,
-                "application_type": self.get_application_type(link_object.sample),
+                "application_type": self.get_analysis_type(link_object.sample),
                 "target_bed": self.resolve_target_bed(panel_bed=panel_bed, link_object=link_object),
             }
             for link_object in self.status_db.get_case_by_internal_id(internal_id=case_id).links
@@ -512,7 +512,7 @@ def get_sample_params(self, case_id: str, panel_bed: str | None) -> dict:
     def resolve_target_bed(self, panel_bed: str | None, link_object: CaseSample) -> str | None:
         if panel_bed:
             return panel_bed
-        if self.get_application_type(link_object.sample) not in self.__BALSAMIC_BED_APPLICATIONS:
+        if self.get_analysis_type(link_object.sample) not in self.__BALSAMIC_BED_APPLICATIONS:
             return None
         return self.get_target_bed_from_lims(link_object.case.internal_id)
 

cg/meta/workflow/mip_dna.py

@@ -1,10 +1,10 @@
 import logging
 
 from cg.constants import DEFAULT_CAPTURE_KIT, Workflow
-from cg.constants.constants import AnalysisType
 from cg.constants.gene_panel import GENOME_BUILD_37
 from cg.constants.pedigree import Pedigree
 from cg.constants.scout import MIP_CASE_TAGS
+from cg.constants.tb import AnalysisType
 from cg.meta.workflow.mip import MipAnalysisAPI
 from cg.models.cg_config import CGConfig
 from cg.models.mip.mip_analysis import MipAnalysis
@@ -54,7 +54,7 @@ def config_sample(
     ) -> dict[str, str | int | None]:
         """Return config sample data."""
         sample_data: dict[str, str | int] = self.get_sample_data(link_obj=link_obj)
-        if sample_data["analysis_type"] == AnalysisType.WHOLE_GENOME_SEQUENCING:
+        if sample_data["analysis_type"] == AnalysisType.WGS:
             sample_data["capture_kit"]: str = panel_bed or DEFAULT_CAPTURE_KIT
         else:
             sample_data["capture_kit"]: str | None = panel_bed or self.get_target_bed_from_lims(
@@ -90,10 +90,8 @@ def get_data_analysis_type(self, case_id: str) -> str:
             link.sample.application_version.application.analysis_type for link in case.links
         }
         if len(analysis_types) > 1:
-            LOG.warning(
-                f"Multiple analysis types found. Defaulting to {AnalysisType.WHOLE_GENOME_SEQUENCING}."
-            )
-            return AnalysisType.WHOLE_GENOME_SEQUENCING
+            LOG.warning(f"Multiple analysis types found. Defaulting to {AnalysisType.WGS}.")
+            return AnalysisType.WGS
         return analysis_types.pop() if analysis_types else None
 
     def get_scout_upload_case_tags(self) -> dict:

cg/meta/workflow/raredisease.py

@@ -14,7 +14,7 @@
     CoverageSample,
 )
 from cg.constants import DEFAULT_CAPTURE_KIT, Workflow
-from cg.constants.constants import AnalysisType, GenomeVersion
+from cg.constants.constants import GenomeVersion
 from cg.constants.nf_analysis import (
     RAREDISEASE_COVERAGE_FILE_TAGS,
     RAREDISEASE_COVERAGE_INTERVAL_TYPE,
@@ -24,6 +24,7 @@
 )
 from cg.constants.scout import RAREDISEASE_CASE_TAGS, ScoutExportFileName
 from cg.constants.subject import PlinkPhenotypeStatus, PlinkSex
+from cg.constants.tb import AnalysisType
 from cg.meta.workflow.nf_analysis import NfAnalysisAPI
 from cg.models.cg_config import CGConfig
 from cg.models.deliverables.metric_deliverables import MetricsBase, MultiqcDataJson
@@ -98,13 +99,13 @@ def get_target_bed(self, case_id: str, analysis_type: str) -> str:
         """
         target_bed_file: str = self.get_target_bed_from_lims(case_id=case_id)
         if not target_bed_file:
-            if analysis_type == AnalysisType.WHOLE_GENOME_SEQUENCING:
+            if analysis_type == AnalysisType.WGS:
                 return DEFAULT_CAPTURE_KIT
             raise ValueError("No capture kit was found in LIMS")
         return target_bed_file
 
     def get_germlinecnvcaller_flag(self, analysis_type: str) -> bool:
-        if analysis_type == AnalysisType.WHOLE_GENOME_SEQUENCING:
+        if analysis_type == AnalysisType.WGS:
             return True
         return False
 

cg/services/deliver_files/deliver_files_service/deliver_files_service.py

@@ -4,7 +4,7 @@
 from cg.apps.tb import TrailblazerAPI
 from cg.apps.tb.models import TrailblazerAnalysis
 from cg.constants import Priority, Workflow
-from cg.constants.tb import AnalysisTypes
+from cg.constants.tb import AnalysisType
 from cg.services.analysis_service.analysis_service import AnalysisService
 from cg.services.deliver_files.deliver_files_service.error_handling import (
     handle_no_delivery_files_error,
@@ -128,7 +128,7 @@ def _add_trailblazer_tracking(self, case: Case, job_id: int, dry_run: bool) -> N
             LOG.debug(f"[TB SERVICE] Adding analysis for case {case.internal_id} to Trailblazer")
             analysis: TrailblazerAnalysis = self.tb_service.add_pending_analysis(
                 case_id=f"{case.internal_id}_rsync",
-                analysis_type=AnalysisTypes.OTHER,
+                analysis_type=AnalysisType.OTHER,
                 config_path=self.rsync_service.trailblazer_config_path.as_posix(),
                 order_id=case.latest_order.id,
                 out_dir=self.rsync_service.log_dir.as_posix(),

cg/services/deliver_files/rsync/service.py

@@ -12,18 +12,17 @@
 from cg.constants.constants import FileFormat
 from cg.constants.delivery import INBOX_NAME
 from cg.constants.priority import SlurmAccount, SlurmQos
-from cg.constants.tb import AnalysisTypes
+from cg.constants.tb import AnalysisType
 from cg.exc import CgError
 from cg.io.controller import WriteFile
+from cg.models.slurm.sbatch import Sbatch
 from cg.services.deliver_files.rsync.models import RsyncDeliveryConfig
 from cg.services.deliver_files.rsync.sbatch import (
+    COVID_REPORT_RSYNC,
     COVID_RSYNC,
     ERROR_RSYNC_FUNCTION,
     RSYNC_COMMAND,
-    COVID_REPORT_RSYNC,
 )
-
-from cg.models.slurm.sbatch import Sbatch
 from cg.store.models import Case
 from cg.store.store import Store
 
@@ -159,7 +158,7 @@ def add_to_trailblazer_api(
         )
         tb_api.add_pending_analysis(
             case_id=ticket,
-            analysis_type=AnalysisTypes.OTHER,
+            analysis_type=AnalysisType.OTHER,
             config_path=self.trailblazer_config_path.as_posix(),
             out_dir=self.log_dir.as_posix(),
             slurm_quality_of_service=self.slurm_quality_of_service,

tests/conftest.py

@@ -37,7 +37,7 @@
 from cg.constants.scout import ScoutExportFileName
 from cg.constants.sequencing import SequencingPlatform
 from cg.constants.subject import Sex
-from cg.constants.tb import AnalysisTypes
+from cg.constants.tb import AnalysisType
 from cg.io.controller import ReadFile, WriteFile
 from cg.io.json import read_json, write_json
 from cg.io.yaml import read_yaml, write_yaml
@@ -2581,7 +2581,7 @@ def raredisease_parameters_default(
         outdir=Path(raredisease_dir, raredisease_case_id),
         target_bed_file=bed_version_file_name,
         skip_germlinecnvcaller=False,
-        analysis_type=AnalysisTypes.WES,
+        analysis_type=AnalysisType.WES,
         save_mapped_as_cram=True,
         vcfanno_extra_resources=str(
             Path(raredisease_dir, raredisease_case_id + ScoutExportFileName.MANAGED_VARIANTS)

tests/models/delivery_report/test_models_validators.py

@@ -19,7 +19,8 @@
     YES_FIELD,
     Sex,
 )
-from cg.constants.constants import AnalysisType, Workflow
+from cg.constants.constants import Workflow
+from cg.constants.tb import AnalysisType
 from cg.meta.delivery_report.delivery_report_api import DeliveryReportAPI
 from cg.meta.delivery_report.rnafusion import RnafusionDeliveryReportAPI
 from cg.models.analysis import NextflowAnalysis
@@ -94,7 +95,7 @@ def test_get_number_as_string(input_value: Any, expected_output: str, caplog: Lo
 
     # GIVEN a list of number inputs and their expected values
 
-    if expected_output == ValueError:
+    if expected_output is ValueError:
         # WHEN getting a string representation of a number
         with pytest.raises(ValueError):
             get_number_as_string(input_value)
@@ -221,7 +222,7 @@ def test_get_analysis_type_as_string():
     """Test analysis type formatting for the delivery report generation."""
 
     # GIVEN a WHOLE_GENOME_SEQUENCING analysis type and a model info dictionary
-    analysis_type: str = AnalysisType.WHOLE_GENOME_SEQUENCING
+    analysis_type: str = AnalysisType.WGS
     model_info = ValidationInfo
     model_info.data = {"workflow": Workflow.MIP_DNA.value}