add new EPP for calculating available material for seq reload

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)

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)

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__)
 

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