add nallo cg workflow config-case (#4098)

### Added
- cg workflow nallo config-case

cg/cli/workflow/nallo/base.py

@@ -5,6 +5,9 @@
 import rich_click as click
 
 from cg.cli.utils import CLICK_CONTEXT_SETTINGS
+
+from cg.cli.workflow.nf_analysis import config_case
+
 from cg.constants.constants import MetaApis
 from cg.meta.workflow.analysis import AnalysisAPI
 from cg.meta.workflow.nallo import NalloAnalysisAPI
@@ -18,3 +21,6 @@ def nallo(context: click.Context) -> None:
     """GMS/Nallo analysis workflow."""
     AnalysisAPI.get_help(context)
     context.obj.meta_apis[MetaApis.ANALYSIS_API] = NalloAnalysisAPI(config=context.obj)
+
+
+nallo.add_command(config_case)

cg/meta/workflow/nallo.py

@@ -2,8 +2,12 @@
 
 import logging
 from cg.constants import Workflow
+from cg.constants.subject import PlinkPhenotypeStatus, PlinkSex
 from cg.meta.workflow.nf_analysis import NfAnalysisAPI
 from cg.models.cg_config import CGConfig
+from cg.models.nallo.nallo import NalloSampleSheetHeaders, NalloSampleSheetEntry, NalloParameters
+from cg.store.models import CaseSample
+from pathlib import Path
 
 LOG = logging.getLogger(__name__)
 
@@ -18,3 +22,72 @@ def __init__(
         workflow: Workflow = Workflow.NALLO,
     ):
         super().__init__(config=config, workflow=workflow)
+        self.root_dir: str = config.nallo.root
+        self.workflow_bin_path: str = config.nallo.workflow_bin_path
+        self.profile: str = config.nallo.profile
+        self.conda_env: str = config.nallo.conda_env
+        self.conda_binary: str = config.nallo.conda_binary
+        self.platform: str = config.nallo.platform
+        self.params: str = config.nallo.params
+        self.workflow_config_path: str = config.nallo.config
+        self.resources: str = config.nallo.resources
+        self.tower_binary_path: str = config.tower_binary_path
+        self.tower_workflow: str = config.nallo.tower_workflow
+        self.account: str = config.nallo.slurm.account
+        self.email: str = config.nallo.slurm.mail_user
+        self.compute_env_base: str = config.nallo.compute_env
+        self.revision: str = config.nallo.revision
+        self.nextflow_binary_path: str = config.nallo.binary_path
+
+    @property
+    def sample_sheet_headers(self) -> list[str]:
+        """Headers for sample sheet."""
+        return NalloSampleSheetHeaders.list()
+
+    def get_sample_sheet_content_per_sample(self, case_sample: CaseSample) -> list[list[str]]:
+        """Collect and format information required to build a sample sheet for a single sample."""
+        read_file_paths = self.get_bam_read_file_paths(sample=case_sample.sample)
+        sample_sheet_entries = []
+
+        for bam_path in read_file_paths:
+            sample_sheet_entry = NalloSampleSheetEntry(
+                project=case_sample.case.internal_id,
+                sample=case_sample.sample.internal_id,
+                read_file=Path(bam_path),
+                family_id=case_sample.case.internal_id,
+                paternal_id=case_sample.get_paternal_sample_id or "0",
+                maternal_id=case_sample.get_maternal_sample_id or "0",
+                sex=self.get_sex_code(case_sample.sample.sex),
+                phenotype=self.get_phenotype_code(case_sample.status),
+            )
+            sample_sheet_entries.extend(sample_sheet_entry.reformat_sample_content)
+        return sample_sheet_entries
+
+    @staticmethod
+    def get_phenotype_code(phenotype: str) -> int:
+        """Return Nallo phenotype code."""
+        LOG.debug("Translate phenotype to integer code")
+        try:
+            code = PlinkPhenotypeStatus[phenotype.upper()]
+        except KeyError:
+            raise ValueError(f"{phenotype} is not a valid phenotype")
+        return code
+
+    @staticmethod
+    def get_sex_code(sex: str) -> int:
+        """Return Nallo sex code."""
+        LOG.debug("Translate sex to integer code")
+        try:
+            code = PlinkSex[sex.upper()]
+        except KeyError:
+            raise ValueError(f"{sex} is not a valid sex")
+        return code
+
+    def get_built_workflow_parameters(self, case_id: str) -> NalloParameters:
+        """Return parameters."""
+        outdir = self.get_case_path(case_id=case_id)
+
+        return NalloParameters(
+            input=self.get_sample_sheet_path(case_id=case_id),
+            outdir=outdir,
+        )

cg/meta/workflow/nf_analysis.py

@@ -18,6 +18,7 @@
     WorkflowManager,
 )
 from cg.constants.gene_panel import GenePanelGenomeBuild
+from cg.constants.housekeeper_tags import AlignmentFileTag
 from cg.constants.nextflow import NFX_WORK_DIR
 from cg.constants.nf_analysis import NfTowerStatus
 from cg.constants.tb import AnalysisStatus
@@ -252,6 +253,15 @@ def get_paired_read_paths(self, sample=Sample) -> tuple[list[str], list[str]]:
         )
         return fastq_forward_read_paths, fastq_reverse_read_paths
 
+    def get_bam_read_file_paths(self, sample=Sample) -> list[Path]:
+        """Gather BAM file path for a sample based on the BAM tag."""
+        return [
+            Path(hk_file.full_path)
+            for hk_file in self.housekeeper_api.files(
+                bundle=sample.internal_id, tags={AlignmentFileTag.BAM}
+            )
+        ]
+
     def get_sample_sheet_content_per_sample(self, case_sample: CaseSample) -> list[list[str]]:
         """Collect and format information required to build a sample sheet for a single sample."""
         raise NotImplementedError

cg/models/cg_config.py

@@ -211,6 +211,24 @@ class MipConfig(BaseModel):
     script: str
 
 
+class NalloConfig(CommonAppConfig):
+    binary_path: str | None = None
+    compute_env: str
+    conda_binary: str | None = None
+    conda_env: str
+    platform: str
+    params: str
+    config: str
+    resources: str
+    launch_directory: str
+    workflow_bin_path: str
+    profile: str
+    revision: str
+    root: str
+    slurm: SlurmConfig
+    tower_workflow: str
+
+
 class RarediseaseConfig(CommonAppConfig):
     binary_path: str | None = None
     compute_env: str
@@ -442,6 +460,7 @@ class CGConfig(BaseModel):
     mip_rd_dna: MipConfig | None = Field(None, alias="mip-rd-dna")
     mip_rd_rna: MipConfig | None = Field(None, alias="mip-rd-rna")
     mutant: MutantConfig | None = None
+    nallo: NalloConfig | None = None
     raredisease: RarediseaseConfig | None = None
     rnafusion: RnafusionConfig | None = None
     statina: StatinaConfig | None = None

cg/models/nallo/nallo.py

@@ -0,0 +1,63 @@
+from enum import StrEnum
+from pathlib import Path
+
+from pydantic import BaseModel, field_validator
+
+from cg.exc import NfSampleSheetError
+from cg.models.nf_analysis import WorkflowParameters
+
+
+class NalloSampleSheetEntry(BaseModel):
+    """Nallo sample model is used when building the sample sheet."""
+
+    project: str
+    sample: str
+    read_file: Path
+    family_id: str
+    paternal_id: str
+    maternal_id: str
+    sex: int
+    phenotype: int
+
+    @property
+    def reformat_sample_content(self) -> list[list[str]]:
+        """Reformat sample sheet content as a list of lists, where each list represents a line in the final file."""
+        return [
+            [
+                self.project,
+                self.sample,
+                self.read_file,
+                self.family_id,
+                self.paternal_id,
+                self.maternal_id,
+                self.sex,
+                self.phenotype,
+            ]
+        ]
+
+    @field_validator("read_file")
+    @classmethod
+    def read_file_exists(cls, bam_path: Path) -> Path:
+        """Verify that bam files exist."""
+        if not bam_path.is_file():
+            raise NfSampleSheetError(f"Bam file does not exist: {str(bam_path)}")
+        return bam_path
+
+
+class NalloSampleSheetHeaders(StrEnum):
+    project: str = "project"
+    sample: str = "sample"
+    file: str = "file"
+    family_id: str = "family_id"
+    paternal_id: str = "paternal_id"
+    maternal_id: str = "maternal_id"
+    sex: str = "sex"
+    phenotype: str = "phenotype"
+
+    @classmethod
+    def list(cls) -> list[str]:
+        return list(map(lambda header: header.value, cls))
+
+
+class NalloParameters(WorkflowParameters):
+    """Model for Nallo parameters."""

tests/cli/workflow/nf_analysis/test_cli_config_case.py

@@ -24,7 +24,7 @@
 
 @pytest.mark.parametrize(
     "workflow",
-    NEXTFLOW_WORKFLOWS,
+    NEXTFLOW_WORKFLOWS + [Workflow.NALLO],
 )
 def test_config_case_without_options(
     cli_runner: CliRunner, workflow: Workflow, request: FixtureRequest