From a8fc7be03c9ed214d5efa9b7edbd0b8c88028de7 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Karl=20Sv=C3=A4rd?= <karl.svard@scilifelab.se>
Date: Wed, 15 Nov 2023 11:48:05 +0100
Subject: [PATCH] add new epp for setting seq settings

---
 .../EPPs/udf/set/set_sequencing_settings.py   | 84 +++++++++++++++++++
 1 file changed, 84 insertions(+)
 create mode 100644 cg_lims/EPPs/udf/set/set_sequencing_settings.py

diff --git a/cg_lims/EPPs/udf/set/set_sequencing_settings.py b/cg_lims/EPPs/udf/set/set_sequencing_settings.py
new file mode 100644
index 00000000..a207353f
--- /dev/null
+++ b/cg_lims/EPPs/udf/set/set_sequencing_settings.py
@@ -0,0 +1,84 @@
+import logging
+import sys
+import click
+
+from typing import List, Optional
+from genologics.lims import Artifact, Process
+from cg_lims.exceptions import LimsError, MissingUDFsError
+from cg_lims.get.artifacts import get_artifacts
+
+from cg_lims.EPPs.udf.calculate.constants import FlowCellTypes
+from cg_lims.EPPs.udf.set.constants import DefaultReadLength, DefaultIndexLength
+
+LOG = logging.getLogger(__name__)
+
+DEFAULT_INDEX_LENGTHS = {
+    FlowCellTypes.FLOW_CELL_10B: DefaultIndexLength.FLOW_CELL_10B,
+    FlowCellTypes.FLOW_CELL_15B: DefaultIndexLength.FLOW_CELL_15B,
+    FlowCellTypes.FLOW_CELL_25B: DefaultIndexLength.FLOW_CELL_25B,
+}
+
+DEFAULT_READ_LENGTHS = {
+    FlowCellTypes.FLOW_CELL_10B: DefaultReadLength.FLOW_CELL_10B,
+    FlowCellTypes.FLOW_CELL_15B: DefaultReadLength.FLOW_CELL_15B,
+    FlowCellTypes.FLOW_CELL_25B: DefaultReadLength.FLOW_CELL_25B,
+}
+
+
+def get_library_tube_strip(process: Process) -> str:
+    """Return the Library Tube Strip ID from a process."""
+    library_tube_strip: str = process.udf.get("Library Tube Strip ID")
+    if not library_tube_strip:
+        LOG.error(f"Process {process.id} is missing UDF 'Library Tube Strip ID'")
+        raise MissingUDFsError(f"UDF 'Library Tube Strip ID' missing from previous step.")
+    return library_tube_strip
+
+
+def get_flow_cell_type(process: Process) -> str:
+    """Return the Flow Cell Type from a process."""
+    flow_cell_type: str = process.udf.get("Flow Cell Type")
+    if not flow_cell_type:
+        LOG.error(f"Process {process.id} is missing UDF 'Flow Cell Type'")
+        raise MissingUDFsError(f"UDF 'Flow Cell Type' missing from previous step.")
+    return flow_cell_type
+
+
+def set_process_udfs(process: Process, parent_process: Process) -> None:
+    """Set Prepare for Sequencing (NovaSeq X) process UDFs."""
+    library_tube_strip: str = get_library_tube_strip(process=parent_process)
+    flow_cell_type: str = get_flow_cell_type(process=parent_process)
+    process.udf["Library Tube Strip ID"] = library_tube_strip
+    process.udf["Read 1 Cycles"] = DEFAULT_READ_LENGTHS[flow_cell_type]
+    process.udf["Read 2 Cycles"] = DEFAULT_READ_LENGTHS[flow_cell_type]
+    process.udf["Index Read 1"] = DEFAULT_INDEX_LENGTHS[flow_cell_type]
+    process.udf["Index Read 2"] = DEFAULT_INDEX_LENGTHS[flow_cell_type]
+    process.put()
+
+
+def get_parent_process(process: Process) -> Optional[Process]:
+    """Get the parent process of another process, assuming all input artifacts come from the same step."""
+    input_artifacts: List[Artifact] = get_artifacts(process=process, input=True)
+    if not input_artifacts:
+        LOG.info(f"No input artifacts found for process {process.id}.")
+        return None
+    return input_artifacts[0].parent_process
+
+
+@click.command()
+@click.pass_context
+def set_sequencing_settings(ctx):
+    """Sets the settings required for sequencing of NovaSeq X flow cells."""
+
+    LOG.info(f"Running {ctx.command_path} with params: {ctx.params}")
+
+    process: Process = ctx.obj["process"]
+
+    try:
+        parent_process: Process = get_parent_process(process=process)
+        set_process_udfs(process=process, parent_process=parent_process)
+        message: str = "Sequencing settings have been successfully set."
+        LOG.info(message)
+        click.echo(message)
+    except LimsError as e:
+        LOG.error(e.message)
+        sys.exit(e.message)