diff --git a/cg_lims/EPPs/udf/set/base.py b/cg_lims/EPPs/udf/set/base.py index 7149b9b6..93eed7b4 100644 --- a/cg_lims/EPPs/udf/set/base.py +++ b/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) diff --git a/cg_lims/EPPs/udf/set/replace_flow_cell_output_path.py b/cg_lims/EPPs/udf/set/replace_flow_cell_output_path.py new file mode 100644 index 00000000..1bf83d24 --- /dev/null +++ b/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: + - \\\clinicaldata\Runs\ + - X:\\ + - //cg-nas.scilifelab.se/cg_data/seqdata/20230824_LH00217_0004_A225CW7LT3 + + They are replaced with: \\130.237.80.51\Runs\ + """ + 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)