add cg workflow nallo panel

cg/cli/workflow/nallo/base.py

@@ -4,13 +4,16 @@
 
 import rich_click as click
 
-from cg.cli.utils import CLICK_CONTEXT_SETTINGS
+from cg.cli.utils import CLICK_CONTEXT_SETTINGS, echo_lines
+from cg.cli.workflow.commands import ARGUMENT_CASE_ID
 
 from cg.cli.workflow.nf_analysis import config_case, run, start
+from cg.constants.cli_options import DRY_RUN
 
 from cg.constants.constants import MetaApis
 from cg.meta.workflow.analysis import AnalysisAPI
 from cg.meta.workflow.nallo import NalloAnalysisAPI
+from cg.models.cg_config import CGConfig
 
 LOG = logging.getLogger(__name__)
 
@@ -26,3 +29,20 @@ def nallo(context: click.Context) -> None:
 nallo.add_command(config_case)
 nallo.add_command(run)
 nallo.add_command(start)
+
+
+@nallo.command("panel")
+@DRY_RUN
+@ARGUMENT_CASE_ID
+@click.pass_obj
+def panel(context: CGConfig, case_id: str, dry_run: bool) -> None:
+    """Write aggregated gene panel file exported from Scout."""
+
+    analysis_api: NalloAnalysisAPI = context.meta_apis["analysis_api"]
+    analysis_api.status_db.verify_case_exists(case_internal_id=case_id)
+
+    bed_lines: list[str] = analysis_api.get_gene_panel(case_id=case_id)
+    if dry_run:
+        echo_lines(lines=bed_lines)
+        return
+    analysis_api.write_panel(case_id=case_id, content=bed_lines)

cg/meta/workflow/nallo.py

@@ -2,6 +2,7 @@
 
 import logging
 from cg.constants import Workflow
+from cg.constants.constants import GenomeVersion
 from cg.constants.subject import PlinkPhenotypeStatus, PlinkSex
 from cg.meta.workflow.nf_analysis import NfAnalysisAPI
 from cg.models.cg_config import CGConfig
@@ -91,3 +92,12 @@ def get_built_workflow_parameters(self, case_id: str) -> NalloParameters:
             input=self.get_sample_sheet_path(case_id=case_id),
             outdir=outdir,
         )
+
+    @property
+    def is_gene_panel_required(self) -> bool:
+        """Return True if a gene panel needs to be created using information in StatusDB and exporting it from Scout."""
+        return True
+
+    def get_genome_build(self, case_id: str) -> GenomeVersion:
+        """Return reference genome for a Nallo case. Currently fixed for hg38."""
+        return GenomeVersion.HG38

tests/cli/workflow/nallo/test_cli_nallo_compound_commands.py

@@ -0,0 +1,55 @@
+from pathlib import Path
+
+from click.testing import CliRunner
+from cg.cli.workflow.nallo.base import panel
+from cg.constants.scout import ScoutExportFileName
+from cg.io.txt import read_txt
+from cg.meta.workflow.nallo import NalloAnalysisAPI
+from cg.models.cg_config import CGConfig
+
+
+def test_panel_dry_run(
+    nallo_case_id: str,
+    cli_runner: CliRunner,
+    nallo_context: CGConfig,
+    scout_panel_output: str,
+    mocker,
+):
+    # GIVEN a case
+
+    # GIVEN that, the Scout command writes the panel to stdout
+    mocker.patch.object(NalloAnalysisAPI, "get_gene_panel", return_value=scout_panel_output)
+
+    result = cli_runner.invoke(panel, [nallo_case_id, "--dry-run"], obj=nallo_context)
+    # THEN the output should contain the output from Scout
+    actual_output = "".join(result.stdout.strip().split())
+    expected_output = "".join(scout_panel_output.strip().split())
+
+    assert actual_output == expected_output
+
+
+def test_panel_file_is_written(
+    nallo_case_id: str,
+    cli_runner: CliRunner,
+    nallo_context: CGConfig,
+    scout_panel_output: str,
+    mocker,
+):
+    # GIVEN an analysis API
+    analysis_api: NalloAnalysisAPI = nallo_context.meta_apis["analysis_api"]
+
+    # GIVEN a case
+
+    # GIVEN that, the Scout command writes the panel to stdout
+    mocker.patch.object(NalloAnalysisAPI, "get_gene_panel", return_value=scout_panel_output)
+
+    cli_runner.invoke(panel, [nallo_case_id], obj=nallo_context)
+
+    panel_file = Path(analysis_api.root, nallo_case_id, ScoutExportFileName.PANELS)
+
+    # THEN the file should exist
+    assert panel_file.exists()
+
+    # THEN the file should contain the output from Scout
+    file_content: str = read_txt(file_path=panel_file, read_to_string=True)
+    assert "".join(file_content.strip().split()) == "".join(scout_panel_output.strip().split())