Refactor SMRT Cell pooling calculator to comply with the changes in SMRT Link 25.1

cg_lims/EPPs/udf/calculate/revio_pooling.py

@@ -1,6 +1,6 @@
 import logging
 import sys
-from typing import List
+from typing import Any, List
 
 import click
 from cg_lims import options
@@ -21,32 +21,26 @@ def get_targeted_pooling_concentration(process: Process, udf_name: str) -> float
     return final_concentration
 
 
-def get_total_pooling_volume(process: Process, udf_name: str) -> float:
+def get_total_pooling_volume(
+    process: Process, artifact: Artifact, volume_udf_name: str, nr_smrt_cells_udf: str
+) -> float:
     """Return the total pooling volume from a process."""
-    total_volume: float = process.udf.get(udf_name)
+    volume_per_smrt_cell: float = process.udf.get(volume_udf_name)
+    nr_smrt_cells: int = get_numeric_artifact_udf(artifact=artifact, udf_name=nr_smrt_cells_udf)
+    total_volume: float = volume_per_smrt_cell * nr_smrt_cells
     if not total_volume or total_volume == 0:
         raise MissingUDFsError("You need to specify a pool volume above 0 ul.")
     return total_volume
 
 
-def get_sample_concentration(artifact: Artifact, udf_name: str) -> float:
-    """Return the total pooling volume from a process."""
-    concentration: float = artifact.udf.get(udf_name)
-    if not concentration or concentration == 0:
+def get_numeric_artifact_udf(artifact: Artifact, udf_name: str) -> Any:
+    """Return the numeric UDF value from a given artifact"""
+    value: Any = artifact.udf.get(udf_name)
+    if not value or value == 0:
         raise MissingUDFsError(
-            f"Sample {get_one_sample_from_artifact(artifact=artifact)} is missing an input concentration!"
+            f"Artifact {artifact.name} is missing a value for the UDF {udf_name}!"
         )
-    return concentration
-
-
-def get_sample_fragment_size(artifact: Artifact, udf_name: str) -> int:
-    """Return the total pooling volume from a process."""
-    size: int = artifact.udf.get(udf_name)
-    if not size or size == 0:
-        raise MissingUDFsError(
-            f"Sample {get_one_sample_from_artifact(artifact=artifact)} is missing an average fragment size!"
-        )
-    return size
+    return value
 
 
 def convert_ngul_to_pm(ngul_conc: float, average_size: int) -> float:
@@ -73,19 +67,21 @@ def set_pooling_volumes(
     pool_artifact: Artifact,
     target_concentration: float,
     target_volume: float,
+    control_volume: float,
     size_udf: str,
     input_concentration_udf: str,
     sample_volume_udf: str,
     total_volume_udf: str,
     buffer_volume_udf: str,
+    control_volume_udf: str,
 ) -> None:
     """Set the aliquot volumes needed for the pooling."""
     input_artifacts: List[Artifact] = pool_artifact.input_artifact_list()
     number_of_samples: int = get_number_of_samples_in_pool(artifact=pool_artifact)
     total_sample_volume: float = 0
     for input_artifact in input_artifacts:
-        size: int = get_sample_fragment_size(artifact=input_artifact, udf_name=size_udf)
-        input_concentration_ngul: float = get_sample_concentration(
+        size: int = get_numeric_artifact_udf(artifact=input_artifact, udf_name=size_udf)
+        input_concentration_ngul: float = get_numeric_artifact_udf(
             artifact=input_artifact, udf_name=input_concentration_udf
         )
         input_concentration_pm: float = convert_ngul_to_pm(
@@ -103,6 +99,7 @@ def set_pooling_volumes(
     buffer_volume: float = target_volume - total_sample_volume
     pool_artifact.udf[total_volume_udf] = total_sample_volume
     pool_artifact.udf[buffer_volume_udf] = buffer_volume
+    pool_artifact.udf[control_volume_udf] = control_volume
     pool_artifact.put()
 
 
@@ -114,6 +111,8 @@ def set_pooling_volumes(
 @options.size_udf()
 @options.target_volume_udf()
 @options.target_concentration_udf()
+@options.sequencing_container_udf()
+@options.control_volume_udf()
 @click.pass_context
 def revio_pooling(
     ctx: click.Context,
@@ -124,6 +123,8 @@ def revio_pooling(
     size_udf: str,
     target_volume_udf: str,
     target_concentration_udf: str,
+    sequencing_container_udf: str,
+    control_volume_udf: str,
 ):
     """Calculate and set the pooling aliquots needed for pooling SMRTbell libraries."""
 
@@ -133,20 +134,31 @@ def revio_pooling(
 
     try:
         artifacts: List[Artifact] = get_artifacts(process=process)
-        target_volume: float = get_total_pooling_volume(process=process, udf_name=target_volume_udf)
         target_concentration: float = get_targeted_pooling_concentration(
             process=process, udf_name=target_concentration_udf
         )
         for artifact in artifacts:
+            target_volume: float = get_total_pooling_volume(
+                process=process,
+                artifact=artifact,
+                volume_udf_name=target_volume_udf,
+                nr_smrt_cells_udf=sequencing_container_udf,
+            )
+            # 1ul diluted control is always loaded per SMRT Cell
+            control_volume: float = get_numeric_artifact_udf(
+                artifact=artifact, udf_name=sequencing_container_udf
+            )
             set_pooling_volumes(
                 pool_artifact=artifact,
                 target_volume=target_volume,
+                control_volume=control_volume,
                 target_concentration=target_concentration,
                 size_udf=size_udf,
                 input_concentration_udf=concentration_udf,
                 sample_volume_udf=volume_udf,
                 total_volume_udf=total_volume_udf,
                 buffer_volume_udf=buffer_udf,
+                control_volume_udf=control_volume_udf,
             )
 
         message: str = "Pooling volumes have been calculated for all artifacts!"

cg_lims/options.py

@@ -763,3 +763,15 @@ def value_fields(
     help: str = "Value fields in file",
 ) -> click.option:
     return click.option("-vf", "--value-fields", required=True, multiple=True, help=help)
+
+
+def sequencing_container_udf(
+    help: str = "UDF name for values connected to a particular sequencing container.",
+) -> click.option:
+    return click.option("-sc", "--sequencing-container-udf", required=True, help=help)
+
+
+def control_volume_udf(
+    help: str = "UDF name used to get the control volume.",
+) -> click.option:
+    return click.option("-cv", "--control-volume-udf", required=True, help=help)