From 09713279ba75668ecd14abc25fdacc43941cadb6 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Karl=20Sv=C3=A4rd?= <60181709+Karl-Svard@users.noreply.github.com> Date: Mon, 14 Oct 2024 15:08:26 +0200 Subject: [PATCH] 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 | 110 +++++++++++++++--- cg_lims/exceptions.py | 6 + cg_lims/options.py | 6 + .../EPPs/udf/calculate/test_aliquot_volume.py | 54 ++++++++- 4 files changed, 157 insertions(+), 19 deletions(-) diff --git a/cg_lims/EPPs/udf/calculate/aliquot_volume.py b/cg_lims/EPPs/udf/calculate/aliquot_volume.py index b4472c43..909145d3 100644 --- a/cg_lims/EPPs/udf/calculate/aliquot_volume.py +++ b/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 in the 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 in the 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 .""" - 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,6 +118,9 @@ 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)'" ) @@ -58,7 +128,7 @@ def calculate_volumes( 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: diff --git a/cg_lims/exceptions.py b/cg_lims/exceptions.py index 17af7985..56ebe053 100644 --- a/cg_lims/exceptions.py +++ b/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""" diff --git a/cg_lims/options.py b/cg_lims/options.py index 374c8b9a..5973560f 100644 --- a/cg_lims/options.py +++ b/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) diff --git a/tests/EPPs/udf/calculate/test_aliquot_volume.py b/tests/EPPs/udf/calculate/test_aliquot_volume.py index 317f9bf1..fd48e57e 100644 --- a/tests/EPPs/udf/calculate/test_aliquot_volume.py +++ b/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)' + # - An artifact with values for the UDFs 'Concentration' and 'Amount needed (ul)' + 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