Watchmaker EPP updates (#536)(minor)

### Added
- New flag to specify max volume in the calculate volume-buffer EPP

### Changed
- Renamed the TWIST aliquot amount EPP to just "aliquot-amount"
- Also generalized it by moving all hard-coded UDF names and constants to click options instead

cg_lims/EPPs/udf/calculate/aliquot_amount.py

@@ -0,0 +1,116 @@
+import logging
+import sys
+from typing import List
+
+import click
+from cg_lims import options
+from cg_lims.exceptions import LimsError, MissingUDFsError
+from cg_lims.get.artifacts import get_artifacts
+from cg_lims.get.udfs import get_maximum_amount
+from genologics.entities import Artifact, Process
+
+LOG = logging.getLogger(__name__)
+
+
+def get_possible_input_amount(
+    artifact: Artifact, sample_volume_udf: str, total_volume_udf: str, concentration_udf: str
+) -> float:
+    """Return the maximum input amount possible for a sample, depending on the total volume required."""
+    process: Process = artifact.parent_process
+    sample_volume: float = artifact.udf.get(sample_volume_udf)
+    total_volume: float = process.udf.get(total_volume_udf)
+    max_volume: float = min(sample_volume, total_volume)
+    concentration: float = artifact.udf.get(concentration_udf)
+    return max_volume * concentration
+
+
+def get_maximum_input_for_aliquot(
+    artifact: Artifact,
+    sample_volume_udf: str,
+    total_volume_udf: str,
+    concentration_udf: str,
+    maximum_sample_amount: float,
+) -> float:
+    """Return the maximum allowed input amount for the specified artifact."""
+    possible_input_amount: float = get_possible_input_amount(
+        artifact=artifact,
+        sample_volume_udf=sample_volume_udf,
+        total_volume_udf=total_volume_udf,
+        concentration_udf=concentration_udf,
+    )
+    max_input_amount: float = get_maximum_amount(
+        artifact=artifact, default_amount=maximum_sample_amount
+    )
+    return min(possible_input_amount, max_input_amount)
+
+
+def set_amount_needed(
+    artifacts: List[Artifact],
+    sample_volume_udf: str,
+    total_volume_udf: str,
+    concentration_udf: str,
+    amount_udf: str,
+    maximum_sample_amount: float,
+) -> None:
+    """The maximum amount taken into the prep is decided by calculating the minimum between maximum_sample_amount and
+    <the sample concentration> * <the total volume>.
+
+    Any amount below this can be used in the prep if the total amount or sample volume is limited.
+    """
+
+    missing_udfs: int = 0
+    for artifact in artifacts:
+        if not artifact.udf.get(concentration_udf) or not artifact.udf.get(sample_volume_udf):
+            missing_udfs += 1
+            continue
+        maximum_amount: float = get_maximum_input_for_aliquot(
+            artifact=artifact,
+            sample_volume_udf=sample_volume_udf,
+            total_volume_udf=total_volume_udf,
+            concentration_udf=concentration_udf,
+            maximum_sample_amount=maximum_sample_amount,
+        )
+        artifact.udf[amount_udf] = maximum_amount
+        artifact.put()
+
+    if missing_udfs:
+        raise MissingUDFsError(f"UDF missing for {missing_udfs} samples")
+
+
+@click.command()
+@options.volume_udf(help="Sample volume artifact UDF name.")
+@options.total_volume_udf(help="Total volume process UDF name.")
+@options.concentration_udf(help="Sample concentration artifact UDF name.")
+@options.amount_ng_udf(help="Sample amount (ng) artifact UDF name.")
+@options.maximum_amount(help="The maximum input amount of the prep.")
+@click.pass_context
+def aliquot_amount(
+    ctx,
+    volume_udf: str,
+    total_volume_udf: str,
+    concentration_udf: str,
+    amount_ng_udf: str,
+    max_amount: str,
+):
+    """Calculates amount needed for samples."""
+
+    LOG.info(f"Running {ctx.command_path} with params: {ctx.params}")
+
+    process: Process = ctx.obj["process"]
+
+    try:
+        artifacts: List[Artifact] = get_artifacts(process=process, input=False)
+        set_amount_needed(
+            artifacts=artifacts,
+            sample_volume_udf=volume_udf,
+            total_volume_udf=total_volume_udf,
+            concentration_udf=concentration_udf,
+            amount_udf=amount_ng_udf,
+            maximum_sample_amount=int(max_amount),
+        )
+        message: str = "Amount needed has been calculated for all samples."
+        LOG.info(message)
+        click.echo(message)
+    except LimsError as e:
+        LOG.error(e.message)
+        sys.exit(e.message)

cg_lims/EPPs/udf/calculate/base.py

@@ -2,6 +2,7 @@
 
 import click
 from cg_lims.EPPs.udf.calculate.adjust_missing_reads import adjust_missing_reads
+from cg_lims.EPPs.udf.calculate.aliquot_amount import aliquot_amount
 from cg_lims.EPPs.udf.calculate.aliquot_volume import aliquot_volume
 from cg_lims.EPPs.udf.calculate.calculate_amount_ng import calculate_amount_ng
 from cg_lims.EPPs.udf.calculate.calculate_amount_ng_fmol import calculate_amount_ng_fmol
@@ -28,7 +29,6 @@
 from cg_lims.EPPs.udf.calculate.ont_sequencing_reload import ont_available_sequencing_reload
 from cg_lims.EPPs.udf.calculate.qpcr_concentration import qpcr_concentration
 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
 from cg_lims.EPPs.udf.calculate.twist_get_volumes_from_buffer import get_volumes_from_buffer
 
 # commands
@@ -44,7 +44,7 @@ def calculate(ctx):
 
 
 calculate.add_command(twist_pool)
-calculate.add_command(twist_aliquot_amount)
+calculate.add_command(aliquot_amount)
 calculate.add_command(aliquot_volume)
 calculate.add_command(twist_qc_amount)
 calculate.add_command(get_volumes_from_buffer)

cg_lims/EPPs/udf/calculate/calculate_buffer.py

@@ -24,6 +24,7 @@ def calculate_volumes(
     volume_udf: str,
     buffer_udf: str,
     sample_volume_limit: float,
+    maximum_volume: float,
 ):
     """Calculates buffer volume and total volume"""
 
@@ -39,7 +40,7 @@ def calculate_volumes(
             sample_volume=sample_volume, sample_volume_limit=sample_volume_limit
         )
         total_volume = buffer_volume + sample_volume
-        if total_volume > 100:
+        if total_volume > maximum_volume:
             high_volume_warning = True
         artifact.udf[buffer_udf] = buffer_volume
         artifact.udf[total_volume_udf] = total_volume
@@ -50,7 +51,7 @@ def calculate_volumes(
             f'Udf "Sample Volume (ul)" missing for {missing_udfs} out of {len(artifacts)} samples '
         )
     if high_volume_warning:
-        warning_message += "Total volume higher than 100 ul for some samples!"
+        warning_message += f"Total volume higher than {maximum_volume} ul for some samples!"
     if warning_message:
         raise MissingUDFsError(message=warning_message)
 
@@ -60,13 +61,15 @@ def calculate_volumes(
 @options.volume_udf()
 @options.buffer_udf()
 @options.sample_volume_limit()
+@options.maximum_volume()
 @click.pass_context
 def volume_buffer(
     context: click.Context,
     total_volume_udf: str,
     volume_udf: str,
     buffer_udf: str,
     sample_volume_limit: float,
+    max_volume: float,
 ):
     """Buffer volume calculation."""
 
@@ -82,6 +85,7 @@ def volume_buffer(
             volume_udf=volume_udf,
             buffer_udf=buffer_udf,
             sample_volume_limit=sample_volume_limit,
+            maximum_volume=int(max_volume),
         )
         message = "Volumes have been calculated."
         LOG.info(message)

cg_lims/options.py

@@ -699,3 +699,15 @@ def reset_virus_reads(
         multiple=False,
         help=help,
     )
+
+
+def maximum_amount(
+    help: str = "Maximum amount",
+) -> click.option:
+    return click.option("--max-amount", required=True, help=help)
+
+
+def maximum_volume(
+    help: str = "Maximum volume",
+) -> click.option:
+    return click.option("--max-volume", required=True, help=help)