From c9d56b10cdbdb9ea7e942e2f3b2f9df948771dd2 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Karl=20Sv=C3=A4rd?= Date: Wed, 6 Mar 2024 08:36:42 +0100 Subject: [PATCH] add new EPP for calculating available material for seq reload --- cg_lims/EPPs/udf/calculate/base.py | 2 + .../udf/calculate/ont_sequencing_reload.py | 82 +++++++++++++++++++ .../udf/set/set_ont_sequencing_settings.py | 2 +- cg_lims/get/udfs.py | 14 +++- 4 files changed, 98 insertions(+), 2 deletions(-) create mode 100644 cg_lims/EPPs/udf/calculate/ont_sequencing_reload.py diff --git a/cg_lims/EPPs/udf/calculate/base.py b/cg_lims/EPPs/udf/calculate/base.py index 3164512e..e7c61fa5 100644 --- a/cg_lims/EPPs/udf/calculate/base.py +++ b/cg_lims/EPPs/udf/calculate/base.py @@ -22,6 +22,7 @@ from cg_lims.EPPs.udf.calculate.novaseq_x_denaturation import novaseq_x_denaturation from cg_lims.EPPs.udf.calculate.novaseq_x_volumes import novaseq_x_volumes from cg_lims.EPPs.udf.calculate.ont_aliquot_volume import ont_aliquot_volume +from cg_lims.EPPs.udf.calculate.ont_sequencing_reload import ont_available_sequencing_reload from cg_lims.EPPs.udf.calculate.pool_normalization import pool_normalization from cg_lims.EPPs.udf.calculate.sum_missing_reads_in_pool import missing_reads_in_pool from cg_lims.EPPs.udf.calculate.twist_aliquot_amount import twist_aliquot_amount @@ -60,3 +61,4 @@ def calculate(ctx): calculate.add_command(pool_normalization) calculate.add_command(novaseq_x_denaturation) calculate.add_command(ont_aliquot_volume) +calculate.add_command(ont_available_sequencing_reload) diff --git a/cg_lims/EPPs/udf/calculate/ont_sequencing_reload.py b/cg_lims/EPPs/udf/calculate/ont_sequencing_reload.py new file mode 100644 index 00000000..3a706134 --- /dev/null +++ b/cg_lims/EPPs/udf/calculate/ont_sequencing_reload.py @@ -0,0 +1,82 @@ +import logging +import sys +from typing import List + +import click +from cg_lims.exceptions import LimsError +from cg_lims.get.artifacts import get_artifacts +from cg_lims.get.samples import get_one_sample_from_artifact +from cg_lims.get.udfs import get_analyte_udf +from genologics.entities import Artifact, Process +from genologics.lims import Lims + +LOG = logging.getLogger(__name__) + + +def get_total_amount_after_prep(lims: Lims, sample_id: str) -> float: + """""" + return get_analyte_udf( + lims=lims, sample_id=sample_id, process_types=["ONT Start Sequencing"], udf="Amount (fmol)" + ) + + +def get_loading_amount(lims: Lims, sample_id: str) -> float: + """""" + return get_analyte_udf( + lims=lims, + sample_id=sample_id, + process_types=["ONT Start Sequencing"], + udf="Loading Amount (fmol)", + ) + + +def get_reload_amounts(lims: Lims, sample_id: str) -> List[float]: + """""" + result_files = lims.get_artifacts( + process_type="ONT Sequencing and Reloading v2", samplelimsid=sample_id, type="ResultFile" + ) + amounts = [] + for result_file in result_files: + if result_file.name == "EPP Log": + continue + else: + amounts.append(result_file.udf.get("Reload Amount (fmol)")) + return amounts + + +def get_available_amount(lims: Lims, artifact: Artifact) -> float: + """""" + sample_id = get_one_sample_from_artifact(artifact=artifact).id + original_amount = get_total_amount_after_prep(lims=lims, sample_id=sample_id) + loading_amount = get_loading_amount(lims=lims, sample_id=sample_id) + total_reload_amount = sum(get_reload_amounts(lims=lims, sample_id=sample_id)) + return original_amount - loading_amount - total_reload_amount + + +def set_available_amount(lims: Lims, artifact: Artifact) -> None: + """""" + available_amount = get_available_amount(lims=lims, artifact=artifact) + artifact.udf["Available Amount (fmol)"] = available_amount + artifact.put() + + +@click.command() +@click.pass_context +def ont_available_sequencing_reload(ctx): + """Calculates and sets the amount of material available for reload of ONT runs.""" + + LOG.info(f"Running {ctx.command_path} with params: {ctx.params}") + + lims: Lims = ctx.obj["lims"] + process: Process = ctx.obj["process"] + + try: + artifacts = get_artifacts(process=process, measurement=True) + for artifact in artifacts: + set_available_amount(lims=lims, artifact=artifact) + message: str = "Available amounts have been successfully calculated and set." + LOG.info(message) + click.echo(message) + except LimsError as e: + LOG.error(e.message) + sys.exit(e.message) diff --git a/cg_lims/EPPs/udf/set/set_ont_sequencing_settings.py b/cg_lims/EPPs/udf/set/set_ont_sequencing_settings.py index 3d922728..cb11e0c8 100644 --- a/cg_lims/EPPs/udf/set/set_ont_sequencing_settings.py +++ b/cg_lims/EPPs/udf/set/set_ont_sequencing_settings.py @@ -5,7 +5,7 @@ import click from cg_lims.exceptions import LimsError from cg_lims.get.artifacts import get_artifacts -from genologics.lims import Process +from genologics.entities import Process LOG = logging.getLogger(__name__) diff --git a/cg_lims/get/udfs.py b/cg_lims/get/udfs.py index 465cac57..21f8d4e0 100644 --- a/cg_lims/get/udfs.py +++ b/cg_lims/get/udfs.py @@ -1,9 +1,10 @@ import logging from datetime import date from enum import Enum -from typing import Optional +from typing import List, Optional from cg_lims.exceptions import MissingUDFsError +from cg_lims.get.artifacts import get_latest_analyte from genologics.entities import Artifact, Entity from genologics.lims import Lims @@ -59,3 +60,14 @@ def get_maximum_amount(artifact: Artifact, default_amount: float) -> float: if maximum_amount: return maximum_amount return default_amount + + +def get_analyte_udf(lims: Lims, sample_id: str, process_types: List[str], udf: str) -> float: + """Return the UDF value of the analyte generated for the given sample by the given process type.""" + artifact = get_latest_analyte(lims=lims, process_types=process_types, sample_id=sample_id) + value = artifact.udf.get(udf) + if not value: + raise MissingUDFsError( + f"Couldn't find UDF '{udf}' for artifacts of sample {sample_id} generated by the step {process_types}" + ) + return value