Add new EPP for calculating available reload amounts (#488)(minor)

### Added
- new EPP for calculating and setting the sample amounts available for the reload of an ONT run
- new function in cg_lims/get/udfs.py for fetching UDF values from analytes

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/calculate_amount_ng.py

@@ -14,7 +14,7 @@
 @click.command()
 @options.concentration_udf_option()
 @options.amount_udf_option()
-@options.volume_udf_option()
+@options.volume_udf()
 @options.subtract_volume_option()
 @options.measurement()
 @options.input()

cg_lims/EPPs/udf/calculate/ont_sequencing_reload.py

@@ -0,0 +1,94 @@
+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_concentration_after_prep(lims: Lims, sample_id: str) -> float:
+    """Return sample concentration after ONT library prep."""
+    return get_analyte_udf(
+        lims=lims, sample_id=sample_id, process_types=["ONT Start Sequencing"], udf="Concentration"
+    )
+
+
+def get_total_amount_after_prep(lims: Lims, sample_id: str) -> float:
+    """Return total sample amount (fmol) after ONT library prep."""
+    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 the original loading amount of the flow cell."""
+    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]:
+    """Return any potential reloading amounts of the flow cell."""
+    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:
+            reload_amount = result_file.udf.get("Reload Amount (fmol)")
+            if not reload_amount:
+                reload_amount = 0
+            amounts.append(reload_amount)
+    return amounts
+
+
+def get_available_amount(lims: Lims, sample_id: str) -> float:
+    """Return the available sample amount (fmol) after the original loading and any potential reloads."""
+    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_and_conc(lims: Lims, artifact: Artifact) -> None:
+    """Set the available amount and concentration of the sample."""
+    sample_id = get_one_sample_from_artifact(artifact=artifact).id
+    concentration = get_concentration_after_prep(lims=lims, sample_id=sample_id)
+    available_amount = get_available_amount(lims=lims, sample_id=sample_id)
+    artifact.udf["Concentration"] = concentration
+    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_and_conc(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/updated_sample_volume.py

@@ -112,7 +112,7 @@ def set_volumes_from_process(
 @click.command()
 @options.process_types(help="The process type names from where you want to copy the UDF from.")
 @options.sample_artifact(help="Use this flag if you want to copy udf from original artifact")
-@options.process_udf(
+@options.process_udf_optional(
     help="Optionally fetch the volume from a process UDF. Default is otherwise 'Volume (ul)' on the artifact level."
 )
 @options.subtract_volume(help="Subtracts volume taken from sample.")

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 an analyte, given: sample ID, parent process type, UDF name."""
+    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

cg_lims/options.py

@@ -77,6 +77,12 @@ def process_udf(
     return click.option("-pudf", "--process-udf", required=True, help=help)
 
 
+def process_udf_optional(
+    help: str = "Process UDF name. Not required to specify.",
+) -> click.option:
+    return click.option("-pudf", "--process-udf", required=False, help=help)
+
+
 def file_placeholder(
     help: str = "File.",
 ) -> click.option:
@@ -270,7 +276,7 @@ def amount_udf_option(
 def volume_udf_option(
     help: str = "String of UDF used to get volume value",
 ) -> click.option:
-    return click.option("--volume-udf", required=True, help=help)
+    return click.option("--volume-udf", required=False, help=help)
 
 
 def concentration_udf_option(
@@ -390,7 +396,7 @@ def add_volume(
 ) -> click.option:
     return click.option("--add-volume", required=False, default=0, help=help)
 
-  
+
 def amount_fmol_udf(
     help: str = "String of UDF used to get amount (fmol)",
 ) -> click.option: