diff --git a/cg_lims/EPPs/udf/calculate/revio_pooling.py b/cg_lims/EPPs/udf/calculate/revio_pooling.py index 556dc3a9..1ac90969 100644 --- a/cg_lims/EPPs/udf/calculate/revio_pooling.py +++ b/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!" diff --git a/cg_lims/options.py b/cg_lims/options.py index 3c6d7319..ccc0a1e5 100644 --- a/cg_lims/options.py +++ b/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)