Refactor to pep0663 (#2712)

### Changed

- Do not use `.value` with StrEnum and IntEnums
- Use classmethods for mixed type constants, instead of detached constants
- Better type safety with single type in Enums

cg/apps/demultiplex/sample_sheet/models.py

@@ -26,43 +26,39 @@ class FlowCellSample(BaseModel):
 class FlowCellSampleBcl2Fastq(FlowCellSample):
     """Base class for NovaSeq6000 flow cell samples."""
 
-    flowcell_id: str = Field("", alias=SampleSheetBcl2FastqSections.Data.FLOW_CELL_ID.value)
-    lane: int = Field(..., alias=SampleSheetBcl2FastqSections.Data.LANE.value)
+    flowcell_id: str = Field("", alias=SampleSheetBcl2FastqSections.Data.FLOW_CELL_ID)
+    lane: int = Field(..., alias=SampleSheetBcl2FastqSections.Data.LANE)
     sample_ref: str = Field(
-        GenomeVersion.hg19.value, alias=SampleSheetBcl2FastqSections.Data.SAMPLE_REFERENCE.value
+        GenomeVersion.hg19, alias=SampleSheetBcl2FastqSections.Data.SAMPLE_REFERENCE
     )
-    index: str = Field(..., alias=SampleSheetBcl2FastqSections.Data.INDEX_1.value)
-    index2: str = Field("", alias=SampleSheetBcl2FastqSections.Data.INDEX_2.value)
-    sample_name: str = Field(..., alias=SampleSheetBcl2FastqSections.Data.SAMPLE_NAME.value)
-    control: str = Field("N", alias=SampleSheetBcl2FastqSections.Data.CONTROL.value)
-    recipe: str = Field("R1", alias=SampleSheetBcl2FastqSections.Data.RECIPE.value)
-    operator: str = Field("script", alias=SampleSheetBcl2FastqSections.Data.OPERATOR.value)
+    index: str = Field(..., alias=SampleSheetBcl2FastqSections.Data.INDEX_1)
+    index2: str = Field("", alias=SampleSheetBcl2FastqSections.Data.INDEX_2)
+    sample_name: str = Field(..., alias=SampleSheetBcl2FastqSections.Data.SAMPLE_NAME)
+    control: str = Field("N", alias=SampleSheetBcl2FastqSections.Data.CONTROL)
+    recipe: str = Field("R1", alias=SampleSheetBcl2FastqSections.Data.RECIPE)
+    operator: str = Field("script", alias=SampleSheetBcl2FastqSections.Data.OPERATOR)
 
     sample_id: SampleId = Field(
-        ..., alias=SampleSheetBcl2FastqSections.Data.SAMPLE_INTERNAL_ID_BCL2FASTQ.value
-    )
-    project: str = Field(
-        ..., alias=SampleSheetBcl2FastqSections.Data.SAMPLE_PROJECT_BCL2FASTQ.value
+        ..., alias=SampleSheetBcl2FastqSections.Data.SAMPLE_INTERNAL_ID_BCL2FASTQ
     )
+    project: str = Field(..., alias=SampleSheetBcl2FastqSections.Data.SAMPLE_PROJECT_BCL2FASTQ)
 
 
 class FlowCellSampleBCLConvert(FlowCellSample):
     """Class that represents a NovaSeqX flow cell sample."""
 
-    lane: int = Field(..., alias=SampleSheetBCLConvertSections.Data.LANE.value)
-    sample_id: SampleId = Field(
-        ..., alias=SampleSheetBCLConvertSections.Data.SAMPLE_INTERNAL_ID.value
-    )
-    index: str = Field(..., alias=SampleSheetBCLConvertSections.Data.INDEX_1.value)
-    index2: str = Field("", alias=SampleSheetBCLConvertSections.Data.INDEX_2.value)
-    override_cycles: str = Field("", alias=SampleSheetBCLConvertSections.Data.OVERRIDE_CYCLES.value)
-    adapter_read_1: str = Field("", alias=SampleSheetBCLConvertSections.Data.ADAPTER_READ_1.value)
-    adapter_read_2: str = Field("", alias=SampleSheetBCLConvertSections.Data.ADAPTER_READ_2.value)
+    lane: int = Field(..., alias=SampleSheetBCLConvertSections.Data.LANE)
+    sample_id: SampleId = Field(..., alias=SampleSheetBCLConvertSections.Data.SAMPLE_INTERNAL_ID)
+    index: str = Field(..., alias=SampleSheetBCLConvertSections.Data.INDEX_1)
+    index2: str = Field("", alias=SampleSheetBCLConvertSections.Data.INDEX_2)
+    override_cycles: str = Field("", alias=SampleSheetBCLConvertSections.Data.OVERRIDE_CYCLES)
+    adapter_read_1: str = Field("", alias=SampleSheetBCLConvertSections.Data.ADAPTER_READ_1)
+    adapter_read_2: str = Field("", alias=SampleSheetBCLConvertSections.Data.ADAPTER_READ_2)
     barcode_mismatches_1: int = Field(
-        1, alias=SampleSheetBCLConvertSections.Data.BARCODE_MISMATCHES_1.value
+        1, alias=SampleSheetBCLConvertSections.Data.BARCODE_MISMATCHES_1
     )
     barcode_mismatches_2: int = Field(
-        1, alias=SampleSheetBCLConvertSections.Data.BARCODE_MISMATCHES_2.value
+        1, alias=SampleSheetBCLConvertSections.Data.BARCODE_MISMATCHES_2
     )
 
 

cg/apps/demultiplex/sample_sheet/read_sample_sheet.py

@@ -62,10 +62,10 @@ def get_sample_type(sample_sheet_path: Path) -> Type[FlowCellSample]:
     for row in sample_sheet_content:
         if not row:
             continue
-        if SampleSheetBCLConvertSections.Data.HEADER.value in row[0]:
+        if SampleSheetBCLConvertSections.Data.HEADER in row[0]:
             LOG.info("Sample sheet was generated for BCL Convert")
             return FlowCellSampleBCLConvert
-        if SampleSheetBcl2FastqSections.Data.HEADER.value in row[0]:
+        if SampleSheetBcl2FastqSections.Data.HEADER in row[0]:
             LOG.info("Sample sheet was generated for BCL2FASTQ")
             return FlowCellSampleBcl2Fastq
     raise SampleSheetError("Could not determine sample sheet type")

cg/apps/demultiplex/sample_sheet/sample_sheet_creator.py

@@ -164,16 +164,16 @@ def add_override_cycles_to_samples(self) -> None:
     def get_additional_sections_sample_sheet(self) -> list[list[str]]:
         """Return all sections of the sample sheet that are not the data section."""
         return [
-            [SampleSheetBcl2FastqSections.Settings.HEADER.value],
-            SampleSheetBcl2FastqSections.Settings.BARCODE_MISMATCH_INDEX1.value,
-            SampleSheetBcl2FastqSections.Settings.BARCODE_MISMATCH_INDEX2.value,
+            [SampleSheetBcl2FastqSections.Settings.HEADER],
+            SampleSheetBcl2FastqSections.Settings.barcode_mismatch_index_1(),
+            SampleSheetBcl2FastqSections.Settings.barcode_mismatch_index_2(),
         ]
 
     def get_data_section_header_and_columns(self) -> list[list[str]]:
         """Return the header and column names of the data section of the sample sheet."""
         return [
             [SampleSheetBcl2FastqSections.Data.HEADER.value],
-            SampleSheetBcl2FastqSections.Data.COLUMN_NAMES.value,
+            SampleSheetBcl2FastqSections.Data.column_names(),
         ]
 
 
@@ -226,15 +226,15 @@ def get_additional_sections_sample_sheet(self) -> list[list[str]]:
         """Return all sections of the sample sheet that are not the data section."""
         header_section: list[list[str]] = [
             [SampleSheetBCLConvertSections.Header.HEADER.value],
-            SampleSheetBCLConvertSections.Header.FILE_FORMAT.value,
+            SampleSheetBCLConvertSections.Header.file_format(),
             [SampleSheetBCLConvertSections.Header.RUN_NAME.value, self.flow_cell_id],
             [
                 SampleSheetBCLConvertSections.Header.INSTRUMENT_PLATFORM_TITLE.value,
-                SampleSheetBCLConvertSections.Header.INSTRUMENT_PLATFORM_VALUE.value[
+                SampleSheetBCLConvertSections.Header.instrument_platform_sequencer().get(
                     self.flow_cell.sequencer_type
-                ],
+                ),
             ],
-            SampleSheetBCLConvertSections.Header.INDEX_ORIENTATION_FORWARD.value,
+            SampleSheetBCLConvertSections.Header.index_orientation_forward(),
         ]
         reads_section: list[list[str]] = [
             [SampleSheetBCLConvertSections.Reads.HEADER],
@@ -256,15 +256,15 @@ def get_additional_sections_sample_sheet(self) -> list[list[str]]:
             ],
         ]
         settings_section: list[list[str]] = [
-            [SampleSheetBCLConvertSections.Settings.HEADER.value],
-            SampleSheetBCLConvertSections.Settings.SOFTWARE_VERSION.value,
-            SampleSheetBCLConvertSections.Settings.FASTQ_COMPRESSION_FORMAT.value,
+            [SampleSheetBCLConvertSections.Settings.HEADER],
+            SampleSheetBCLConvertSections.Settings.software_version(),
+            SampleSheetBCLConvertSections.Settings.fastq_compression_format(),
         ]
         return header_section + reads_section + settings_section
 
     def get_data_section_header_and_columns(self) -> list[list[str]]:
         """Return the header and column names of the data section of the sample sheet."""
         return [
             [SampleSheetBCLConvertSections.Data.HEADER.value],
-            SampleSheetBCLConvertSections.Data.COLUMN_NAMES.value,
+            SampleSheetBCLConvertSections.Data.column_names(),
         ]

cg/apps/mip/confighandler.py

@@ -24,8 +24,8 @@ class SampleSchema(Schema):
             ]
         ),
     )
-    father = fields.Str(dump_default=RelationshipStatus.HAS_NO_PARENT.value)
-    mother = fields.Str(dump_default=RelationshipStatus.HAS_NO_PARENT.value)
+    father = fields.Str(dump_default=RelationshipStatus.HAS_NO_PARENT)
+    mother = fields.Str(dump_default=RelationshipStatus.HAS_NO_PARENT)
     phenotype = fields.Str(
         required=True,
         validate=validate.OneOf(choices=["affected", "unaffected", "unknown"]),
@@ -80,10 +80,10 @@ def parse_pedigree_config(data: dict) -> dict:
             LOG.info("setting 'unknown' phenotype to 'unaffected'")
             data_copy["samples"][0]["phenotype"] = "unaffected"
         for sample_data in data_copy["samples"]:
-            sample_data["mother"] = (
+            sample_data["mother"]: str = (
                 sample_data.get("mother") or RelationshipStatus.HAS_NO_PARENT.value
             )
-            sample_data["father"] = (
+            sample_data["father"]: str = (
                 sample_data.get("father") or RelationshipStatus.HAS_NO_PARENT.value
             )
             if sample_data["analysis_type"] == "wgs" and sample_data.get("capture_kit") is None:

cg/apps/sequencing_metrics_parser/models/bcl_convert.py

@@ -9,23 +9,21 @@
 class BclConvertQualityMetrics(BaseModel):
     """Model for the BCL Convert quality metrics."""
 
-    lane: int = Field(..., alias=BclConvertQualityMetricsColumnNames.LANE.value)
+    lane: int = Field(..., alias=BclConvertQualityMetricsColumnNames.LANE)
     sample_internal_id: str = Field(
-        ..., alias=BclConvertQualityMetricsColumnNames.SAMPLE_INTERNAL_ID.value
+        ..., alias=BclConvertQualityMetricsColumnNames.SAMPLE_INTERNAL_ID
     )
     mean_quality_score_q30: float = Field(
-        ..., alias=BclConvertQualityMetricsColumnNames.MEAN_QUALITY_SCORE_Q30.value
+        ..., alias=BclConvertQualityMetricsColumnNames.MEAN_QUALITY_SCORE_Q30
     )
     q30_bases_percent: float = Field(
-        ..., alias=BclConvertQualityMetricsColumnNames.Q30_BASES_PERCENT.value
+        ..., alias=BclConvertQualityMetricsColumnNames.Q30_BASES_PERCENT
     )
 
 
 class BclConvertDemuxMetrics(BaseModel):
     """Model for the BCL Convert demultiplexing metrics."""
 
-    lane: int = Field(..., alias=BclConvertDemuxMetricsColumnNames.LANE.value)
-    sample_internal_id: str = Field(
-        ..., alias=BclConvertDemuxMetricsColumnNames.SAMPLE_INTERNAL_ID.value
-    )
-    read_pair_count: int = Field(..., alias=BclConvertDemuxMetricsColumnNames.READ_PAIR_COUNT.value)
+    lane: int = Field(..., alias=BclConvertDemuxMetricsColumnNames.LANE)
+    sample_internal_id: str = Field(..., alias=BclConvertDemuxMetricsColumnNames.SAMPLE_INTERNAL_ID)
+    read_pair_count: int = Field(..., alias=BclConvertDemuxMetricsColumnNames.READ_PAIR_COUNT)

cg/apps/slurm/slurm_api.py

@@ -30,7 +30,7 @@ def set_dry_run(self, dry_run: bool) -> None:
     @staticmethod
     def generate_sbatch_content(sbatch_parameters: Sbatch) -> str:
         """Take a parameters object and generate a string with sbatch information."""
-        if hasattr(sbatch_parameters, Slurm.PARTITION.value):
+        if hasattr(sbatch_parameters, Slurm.PARTITION):
             sbatch_header: str = SlurmAPI.generate_dragen_sbatch_header(
                 sbatch_parameters=sbatch_parameters
             )

cg/cli/clean.py

@@ -24,7 +24,7 @@
 )
 from cg.constants import EXIT_FAIL, EXIT_SUCCESS
 from cg.constants.constants import DRY_RUN, SKIP_CONFIRMATION, Pipeline
-from cg.constants.housekeeper_tags import ALIGNMENT_FILE_TAGS, ScoutTag
+from cg.constants.housekeeper_tags import AlignmentFileTag, ScoutTag
 from cg.exc import CleanFlowCellFailedError
 from cg.meta.clean.api import CleanAPI
 from cg.meta.clean.clean_flow_cells import CleanFlowCellAPI
@@ -81,7 +81,7 @@ def hk_alignment_files(
 ) -> None:
     """Clean up alignment files in Housekeeper bundle."""
     housekeeper_api: HousekeeperAPI = context.housekeeper_api
-    for tag in ALIGNMENT_FILE_TAGS:
+    for tag in AlignmentFileTag.file_tags():
         tag_files = set(housekeeper_api.get_files(bundle=bundle, tags=[tag]))
 
         if not tag_files:
@@ -121,7 +121,7 @@ def scout_finished_cases(
     """Clean up of solved and archived Scout cases."""
     scout_api: ScoutAPI = context.obj.scout_api
     bundles: list[str] = []
-    for status in [ScoutTag.ARCHIVED.value, ScoutTag.SOLVED.value]:
+    for status in [ScoutTag.ARCHIVED, ScoutTag.SOLVED]:
         cases: list[ScoutExportCase] = scout_api.get_cases(status=status, reruns=False)
         cases_added: int = 0
         for case in cases:

cg/cli/compress/helpers.py

@@ -10,7 +10,7 @@
 
 from cg.apps.housekeeper.hk import HousekeeperAPI
 from cg.constants.compression import CRUNCHY_MIN_GB_PER_PROCESS, MAX_READS_PER_GB
-from cg.constants.slurm import Slurm
+from cg.constants.slurm import SlurmProcessing
 from cg.exc import CaseNotFoundError
 from cg.meta.compress import CompressAPI
 from cg.meta.compress.files import get_spring_paths
@@ -51,7 +51,7 @@ def get_fastq_individuals(store: Store, case_id: str = None) -> Iterator[str]:
 def set_memory_according_to_reads(
     sample_id: str, sample_reads: int | None = None, sample_process_mem: int | None = None
 ) -> int | None:
-    """Set SLURM sample process memory depending on number of sample reads if sample_process_mem is not set."""
+    """Set SLURM sample process memory depending on the number of sample reads if sample_process_mem is not set."""
     if sample_process_mem:
         return sample_process_mem
     if not sample_reads:
@@ -60,9 +60,9 @@ def set_memory_according_to_reads(
     sample_process_mem: int = ceil((sample_reads / MAX_READS_PER_GB))
     if sample_process_mem < CRUNCHY_MIN_GB_PER_PROCESS:
         return CRUNCHY_MIN_GB_PER_PROCESS
-    if CRUNCHY_MIN_GB_PER_PROCESS <= sample_process_mem < Slurm.MAX_NODE_MEMORY.value:
+    if CRUNCHY_MIN_GB_PER_PROCESS <= sample_process_mem < SlurmProcessing.MAX_NODE_MEMORY:
         return sample_process_mem
-    return Slurm.MAX_NODE_MEMORY.value
+    return SlurmProcessing.MAX_NODE_MEMORY
 
 
 def update_compress_api(

cg/cli/set/base.py

@@ -8,7 +8,7 @@
 
 from cg.cli.set.case import set_case
 from cg.cli.set.cases import set_cases
-from cg.constants import FLOWCELL_STATUS
+from cg.constants import FlowCellStatus
 from cg.exc import LimsDataError
 from cg.models.cg_config import CGConfig
 from cg.store import Store
@@ -283,7 +283,7 @@ def _update_comment(comment, obj):
 
 
 @set_cmd.command()
-@click.option("-s", "--status", type=click.Choice(FLOWCELL_STATUS))
+@click.option("-s", "--status", type=click.Choice(FlowCellStatus.statuses()))
 @click.argument("flow_cell_name")
 @click.pass_obj
 def flowcell(context: CGConfig, flow_cell_name: str, status: str | None):

cg/cli/set/case.py

@@ -3,7 +3,8 @@
 
 import click
 
-from cg.constants import CASE_ACTIONS, DataDelivery, Pipeline, Priority
+from cg.constants import DataDelivery, Pipeline, Priority
+from cg.constants.constants import CaseActions
 from cg.models.cg_config import CGConfig
 from cg.store import Store
 from cg.store.models import Case, Customer, Panel
@@ -13,7 +14,7 @@
 
 
 @click.command("case")
-@click.option("-a", "--action", type=click.Choice(CASE_ACTIONS), help="update case action")
+@click.option("-a", "--action", type=click.Choice(CaseActions.actions()), help="update case action")
 @click.option("-c", "--customer-id", type=click.STRING, help="update customer")
 @click.option(
     "-d",

cg/cli/set/cases.py

@@ -3,7 +3,8 @@
 import click
 
 from cg.cli.set.case import set_case
-from cg.constants import CASE_ACTIONS, Priority
+from cg.constants import Priority
+from cg.constants.constants import CaseActions
 from cg.store import Store
 from cg.store.models import Case, Sample
 from cg.utils.click.EnumChoice import EnumChoice
@@ -40,7 +41,7 @@ def _get_cases(identifiers: click.Tuple([str, str]), store: Store) -> list[Case]
     help="Give an identifier on sample and the value to use it with, e.g. --sample-identifier "
     "name Prov52",
 )
-@click.option("-a", "--action", type=click.Choice(CASE_ACTIONS), help="update case action")
+@click.option("-a", "--action", type=click.Choice(CaseActions.actions()), help="update case action")
 @click.option("-c", "--customer-id", type=click.STRING, help="update customer")
 @click.option("-g", "--panel", "panel_abbreviations", multiple=True, help="update gene panels")
 @click.option(

cg/cli/upload/clinical_delivery.py

@@ -6,10 +6,9 @@
 import click
 
 from cg.apps.tb import TrailblazerAPI
-from cg.constants import EXIT_FAIL, EXIT_SUCCESS, Pipeline
+from cg.constants import EXIT_FAIL, EXIT_SUCCESS, Pipeline, Priority
 from cg.constants.constants import DRY_RUN
 from cg.constants.delivery import PIPELINE_ANALYSIS_TAG_MAP
-from cg.constants.priority import PRIORITY_TO_SLURM_QOS
 from cg.constants.tb import AnalysisTypes
 from cg.meta.deliver import DeliverAPI
 from cg.meta.rsync import RsyncAPI
@@ -70,7 +69,7 @@ def upload_clinical_delivery(context: click.Context, case_id: str, dry_run: bool
             analysis_type=AnalysisTypes.OTHER,
             config_path=rsync_api.trailblazer_config_path.as_posix(),
             out_dir=rsync_api.log_dir.as_posix(),
-            slurm_quality_of_service=PRIORITY_TO_SLURM_QOS[case.priority],
+            slurm_quality_of_service=Priority.priority_to_slurm_qos().get(case.priority),
             data_analysis=Pipeline.RSYNC,
             ticket=case.latest_ticket,
         )

cg/cli/workflow/rnafusion/options.py

@@ -13,7 +13,7 @@
 OPTION_STRANDEDNESS = click.option(
     "--strandedness",
     type=str,
-    default=Strandedness.REVERSE.value,
+    default=Strandedness.REVERSE,
     show_default=True,
     help="Strandedness: forward, unstranded or reverse (default)",
 )

cg/constants/__init__.py

@@ -1,15 +1,17 @@
 """Import all constants here for easy access"""
 
-from cg.constants.compression import FASTQ_DELTA, FASTQ_FIRST_READ_SUFFIX, FASTQ_SECOND_READ_SUFFIX
+from cg.constants.compression import (
+    FASTQ_DELTA,
+    FASTQ_FIRST_READ_SUFFIX,
+    FASTQ_SECOND_READ_SUFFIX,
+)
 from cg.constants.constants import (
     CAPTUREKIT_CANCER_OPTIONS,
     CAPTUREKIT_OPTIONS,
-    CASE_ACTIONS,
     COLLABORATORS,
     COMBOS,
     CONTAINER_OPTIONS,
     DEFAULT_CAPTURE_KIT,
-    FLOWCELL_STATUS,
     PREP_CATEGORIES,
     SEX_OPTIONS,
     STATUS_OPTIONS,
@@ -18,7 +20,11 @@
     FlowCellStatus,
 )
 from cg.constants.gene_panel import GenePanelMasterList
-from cg.constants.housekeeper_tags import HK_FASTQ_TAGS, HK_MULTIQC_HTML_TAG, SequencingFileTag
+from cg.constants.housekeeper_tags import (
+    HK_FASTQ_TAGS,
+    HK_MULTIQC_HTML_TAG,
+    SequencingFileTag,
+)
 from cg.constants.paths import TMP_DIR
 from cg.constants.priority import Priority
 from cg.constants.process import EXIT_FAIL, EXIT_SUCCESS