Add low volume warning to aliquot volume EPP (#547)(minor)

### Added
- New exception for low volumes, LowVolumeError
- New Click option for minimum volume threshold, minimum_volume

### Changed
- Restructured aliquot_volume.py EPP

cg_lims/EPPs/udf/calculate/aliquot_volume.py

@@ -4,13 +4,62 @@
 
 import click
 from cg_lims import options
-from cg_lims.exceptions import LimsError, MissingUDFsError
+from cg_lims.exceptions import LimsError, LowVolumeError, MissingUDFsError
 from cg_lims.get.artifacts import get_artifacts
+from cg_lims.get.samples import get_one_sample_from_artifact
 from genologics.entities import Artifact, Process
 
 LOG = logging.getLogger(__name__)
 
 
+def get_maximum_volume(process: Process, udf_name: str) -> float:
+    """Return the maximum volume specified by the UDF <udf_name> in the process <process>"""
+    max_volume: float = process.udf.get(udf_name)
+    if not max_volume:
+        raise MissingUDFsError(f"Process udf missing:{udf_name}")
+    return max_volume
+
+
+def get_artifact_udf(artifact: Artifact, udf_name: str) -> float:
+    """Return the concentration specified by the UDF <udf_name> in the artifact <artifact>"""
+    concentration: float = artifact.udf.get(udf_name)
+    if not concentration:
+        LOG.warning(
+            f" UDF '{udf_name}' is missing for sample {get_one_sample_from_artifact(artifact=artifact).id}."
+        )
+    return concentration
+
+
+def calculate_sample_volume(amount: float, concentration: float) -> float:
+    """Calculate the volume of a sample given the amount needed and concentration."""
+    return amount / concentration
+
+
+def calculate_buffer_volume(max_volume: float, sample_volume: float) -> float:
+    """Calculate the buffer volume of a sample given the max total volume and sample volume."""
+    return max_volume - sample_volume
+
+
+def set_volumes(
+    artifact: Artifact,
+    sample_volume_udf: str,
+    buffer_volume_udf: str,
+    sample_volume: float,
+    buffer_volume: float,
+) -> None:
+    """Set sample UDFs."""
+    artifact.udf[sample_volume_udf] = sample_volume
+    artifact.udf[buffer_volume_udf] = buffer_volume
+    artifact.put()
+
+
+def volumes_below_threshold(
+    minimum_volume: float, sample_volume: float, buffer_volume: float
+) -> bool:
+    """Check if volume aliquots are below the given threshold."""
+    return sample_volume < minimum_volume or buffer_volume < minimum_volume
+
+
 def calculate_volumes(
     artifacts: List[Artifact],
     process: Process,
@@ -19,30 +68,48 @@ def calculate_volumes(
     buffer_volume_udf: str,
     total_volume_udf: str,
     amount_needed_udf: str,
+    minimum_limit: float,
 ):
     """Calculates volumes for diluting samples. The total volume differ depending on
     type of sample. It is given by the process udf <total_volume_udf>."""
 
-    max_volume = process.udf.get(total_volume_udf)
-    if not max_volume:
-        raise MissingUDFsError(f"Process udf missing:{total_volume_udf}")
-
-    missing_udfs = 0
+    max_volume: float = get_maximum_volume(process=process, udf_name=total_volume_udf)
+    samples_missing_udfs: List[str] = []
+    samples_below_threshold: List[str] = []
 
     for art in artifacts:
-        concentration = art.udf.get(concentration_udf)
-        amount = art.udf.get(amount_needed_udf)
+        concentration: float = get_artifact_udf(artifact=art, udf_name=concentration_udf)
+        amount: float = get_artifact_udf(artifact=art, udf_name=amount_needed_udf)
+
         if not amount or not concentration:
-            missing_udfs += 1
+            samples_missing_udfs.append(get_one_sample_from_artifact(artifact=art).id)
             continue
 
-        art.udf[sample_volume_udf] = amount / float(concentration)
-        art.udf[buffer_volume_udf] = max_volume - art.udf[sample_volume_udf]
-        art.put()
+        sample_volume: float = calculate_sample_volume(amount=amount, concentration=concentration)
+        buffer_volume: float = calculate_buffer_volume(
+            max_volume=max_volume, sample_volume=sample_volume
+        )
+        if volumes_below_threshold(
+            minimum_volume=minimum_limit, sample_volume=sample_volume, buffer_volume=buffer_volume
+        ):
+            samples_below_threshold.append(get_one_sample_from_artifact(artifact=art).id)
 
-    if missing_udfs:
+        set_volumes(
+            artifact=art,
+            sample_volume_udf=sample_volume_udf,
+            buffer_volume_udf=buffer_volume_udf,
+            sample_volume=sample_volume,
+            buffer_volume=buffer_volume,
+        )
+
+    if samples_missing_udfs:
         raise MissingUDFsError(
-            f"UDFs '{concentration_udf}' and/or '{amount_needed_udf}' missing for {missing_udfs} samples"
+            f"Error: {len(samples_missing_udfs)} sample(s) are missing values for either '{concentration_udf}' or '{amount_needed_udf}' - {', '.join(samples_missing_udfs)}"
+        )
+
+    if samples_below_threshold:
+        raise LowVolumeError(
+            f"Warning: {len(samples_below_threshold)} sample(s) have aliquot volumes below {minimum_limit} µl - {', '.join(samples_below_threshold)}"
         )
 
 
@@ -51,14 +118,17 @@ def calculate_volumes(
 @options.volume_udf(help="Name of the sample volume artifact UDF")
 @options.buffer_udf(help="Name of the buffer volume artifact UDF")
 @options.total_volume_udf(help="Name of the total volume process UDF")
+@options.minimum_volume(
+    help="The minimum volume (ul) allowed without sending a warning to the user. Default is 0."
+)
 @options.amount_ng_udf(
     help="Use if you want to overwrite the default UDF name 'Amount needed (ng)'"
 )
 @options.measurement(
     help="UDFs will be calculated and set on measurement artifacts. Use in QC steps."
 )
 @options.input(
-    help="UDFs will be calculated ans set on input artifacts. Use non-output generating steps."
+    help="UDFs will be calculated and set on input artifacts. Use non-output generating steps."
 )
 @click.pass_context
 def aliquot_volume(
@@ -67,6 +137,7 @@ def aliquot_volume(
     volume_udf: str,
     buffer_udf: str,
     total_volume_udf: str,
+    min_volume: str = 0,
     amount_ng_udf: str = "Amount needed (ng)",
     measurement: bool = False,
     input: bool = False,
@@ -75,10 +146,12 @@ def aliquot_volume(
 
     LOG.info(f"Running {ctx.command_path} with params: {ctx.params}")
 
-    process = ctx.obj["process"]
+    process: Process = ctx.obj["process"]
 
     try:
-        artifacts = get_artifacts(process=process, input=input, measurement=measurement)
+        artifacts: List[Artifact] = get_artifacts(
+            process=process, input=input, measurement=measurement
+        )
         calculate_volumes(
             artifacts=artifacts,
             process=process,
@@ -87,8 +160,9 @@ def aliquot_volume(
             buffer_volume_udf=buffer_udf,
             total_volume_udf=total_volume_udf,
             amount_needed_udf=amount_ng_udf,
+            minimum_limit=float(min_volume),
         )
-        message = "Volumes have been calculated for all samples."
+        message: str = "Volumes have been calculated for all samples."
         LOG.info(message)
         click.echo(message)
     except LimsError as e:

cg_lims/exceptions.py

@@ -77,6 +77,12 @@ class LowAmountError(LimsError):
     pass
 
 
+class LowVolumeError(LimsError):
+    """Raise when amount is low."""
+
+    pass
+
+
 class FailingQCError(LimsError):
     """Raise when qc fails"""
 

cg_lims/options.py

@@ -711,3 +711,9 @@ def maximum_volume(
     help: str = "Maximum volume",
 ) -> click.option:
     return click.option("--max-volume", required=True, help=help)
+
+
+def minimum_volume(
+    help: str = "Minimum volume",
+) -> click.option:
+    return click.option("--min-volume", required=False, default=0, help=help)

tests/EPPs/udf/calculate/test_aliquot_volume.py

@@ -1,6 +1,6 @@
 import pytest
 from cg_lims.EPPs.udf.calculate.aliquot_volume import calculate_volumes
-from cg_lims.exceptions import MissingUDFsError
+from cg_lims.exceptions import LowVolumeError, MissingUDFsError
 from genologics.entities import Artifact, Process
 from genologics.lims import Lims
 from tests.conftest import server
@@ -46,6 +46,7 @@ def test_calculate_volumes(
         buffer_volume_udf="Volume H2O (ul)",
         total_volume_udf="Total Volume (ul)",
         amount_needed_udf="Amount needed (ng)",
+        minimum_limit=0,
     )
 
     # THEN the correct values are calculated for the artifact UDFs 'Volume H2O (ul)' and 'Sample Volume (ul)'
@@ -89,6 +90,7 @@ def test_calculate_volumes_missing_artifact_udf(lims: Lims, udf_name: str):
             buffer_volume_udf="Volume H2O (ul)",
             total_volume_udf="Total Volume (ul)",
             amount_needed_udf="Amount needed (ng)",
+            minimum_limit=0,
         )
     assert artifact_2.udf["Sample Volume (ul)"] == 20
     assert artifact_2.udf["Volume H2O (ul)"] == 30
@@ -118,4 +120,54 @@ def test_calculate_volumes_missing_process_udf(lims: Lims):
             buffer_volume_udf="Volume H2O (ul)",
             total_volume_udf="Total Volume (ul)",
             amount_needed_udf="Amount needed (ng)",
+            minimum_limit=0,
         )
+
+
+@pytest.mark.parametrize(
+    "total_volume,concentration,amount,sample_vol,water_vol",
+    [
+        (50, 10, 200, 20, 30),
+        (30, 10, 200, 20, 10),
+        (50, 2, 100, 50, 0),
+        (30, 2, 10, 5, 25),
+    ],
+)
+def test_calculate_volumes_below_threshold(
+    lims: Lims,
+    total_volume: float,
+    concentration: float,
+    amount: float,
+    sample_vol: float,
+    water_vol: float,
+):
+    # GIVEN:
+    # - A process with the UDF 'Total Volume (ul)' <total_volume>
+    # - An artifact with values for the UDFs 'Concentration' <concentration> and 'Amount needed (ul)' <amount>
+    server("flat_tests")
+
+    process = Process(lims=lims, id="24-196211")
+    process.udf["Total Volume (ul)"] = total_volume
+    process.put()
+
+    artifact_1 = Artifact(lims=lims, id="1")
+    artifact_1.udf["Concentration"] = concentration
+    artifact_1.udf["Amount needed (ng)"] = amount
+    artifact_1.put()
+
+    # WHEN calculating the aliquot sample and water volumes with a minimum volume threshold above them
+    # THEN MissingUDFsError is being raised and the correct values are calculated
+    with pytest.raises(LowVolumeError):
+        calculate_volumes(
+            artifacts=[artifact_1],
+            process=process,
+            concentration_udf="Concentration",
+            sample_volume_udf="Sample Volume (ul)",
+            buffer_volume_udf="Volume H2O (ul)",
+            total_volume_udf="Total Volume (ul)",
+            amount_needed_udf="Amount needed (ng)",
+            minimum_limit=500,
+        )
+
+    assert artifact_1.udf["Sample Volume (ul)"] == sample_vol
+    assert artifact_1.udf["Volume H2O (ul)"] == water_vol