add new EPP for replacing flow cell output path

cg_lims/EPPs/udf/set/base.py

@@ -5,6 +5,7 @@
 from cg_lims.EPPs.udf.set.set_method import method_document
 from cg_lims.EPPs.udf.set.set_barcode import assign_barcode
 from cg_lims.EPPs.udf.set.set_sequencing_settings import set_sequencing_settings
+from cg_lims.EPPs.udf.set.replace_flow_cell_output_path import replace_flow_cell_output_path
 
 
 @click.group(invoke_without_command=True)
@@ -19,3 +20,4 @@ def set(context: click.Context):
 set.add_command(method_document)
 set.add_command(assign_barcode)
 set.add_command(set_sequencing_settings)
+set.add_command(replace_flow_cell_output_path)

cg_lims/EPPs/udf/set/replace_flow_cell_output_path.py

@@ -0,0 +1,63 @@
+import logging
+import sys
+import click
+
+from genologics.lims import Process
+from cg_lims.exceptions import LimsError, MissingUDFsError
+
+LOG = logging.getLogger(__name__)
+
+
+def get_output_path(process: Process) -> str:
+    """Return the output path UDF of the given process."""
+    path: str = process.udf.get("Output Folder")
+    if not path:
+        LOG.error(f"Process {process.id} is missing UDF 'Output Folder'")
+        raise MissingUDFsError(f"UDF: 'Output Folder' is missing from the step.")
+    return path
+
+
+def convert_output_path(path: str) -> str:
+    """Return a corrected output path if needed. Can currently convert paths of the types:
+    - \\<some novaseq windows network path>\clinicaldata\Runs\<run folder>
+    - X:\<run folder>\
+    - //cg-nas.scilifelab.se/cg_data/seqdata/20230824_LH00217_0004_A225CW7LT3
+
+    They are replaced with: \\130.237.80.51\Runs\<run folder>
+    """
+    new_path = path
+    if "X:\\" in path:
+        new_path = "\\\\130.237.80.51\\Runs" + path.split(":")[1]
+    elif "\\clinicaldata\\Runs\\" in path:
+        new_path = "\\\\130.237.80.51\\Runs" + path.split("Runs")[1]
+    elif "cg-nas.scilifelab.se/cg_data/seqdata/" in path:
+        new_path = "\\\\130.237.80.51\\Runs\\" + path.split("seqdata/")[1]
+    LOG.info(f"Original path '{path}' has been replaced with '{new_path}'.")
+    return new_path
+
+
+def replace_output_path(process: Process) -> None:
+    """Replaces the output path of a process."""
+    current_path = get_output_path(process=process)
+    converted_path = convert_output_path(path=current_path)
+    process.udf["Output Folder"] = converted_path
+    process.put()
+
+
+@click.command()
+@click.pass_context
+def replace_flow_cell_output_path(ctx):
+    """Replaces the output path of flow cells."""
+
+    LOG.info(f"Running {ctx.command_path} with params: {ctx.params}")
+
+    process: Process = ctx.obj["process"]
+
+    try:
+        replace_output_path(process=process)
+        message: str = "Output path has been successfully updated."
+        LOG.info(message)
+        click.echo(message)
+    except LimsError as e:
+        LOG.error(e.message)
+        sys.exit(e.message)