diff --git a/cg_lims/EPPs/udf/calculate/base.py b/cg_lims/EPPs/udf/calculate/base.py index 087152e3..89d19d6b 100644 --- a/cg_lims/EPPs/udf/calculate/base.py +++ b/cg_lims/EPPs/udf/calculate/base.py @@ -21,6 +21,7 @@ from cg_lims.EPPs.udf.calculate.calculate_water_volume_rna import calculate_water_volume_rna from cg_lims.EPPs.udf.calculate.get_missing_reads import get_missing_reads from cg_lims.EPPs.udf.calculate.library_normalization import library_normalization +from cg_lims.EPPs.udf.calculate.library_normalization_revio import library_normalization_revio from cg_lims.EPPs.udf.calculate.maf_calculate_volume import maf_calculate_volume from cg_lims.EPPs.udf.calculate.molar_concentration import molar_concentration from cg_lims.EPPs.udf.calculate.novaseq_x_denaturation import novaseq_x_denaturation @@ -62,6 +63,7 @@ def calculate(ctx): calculate.add_command(calculate_average_size_and_set_qc) calculate.add_command(novaseq_x_volumes) calculate.add_command(library_normalization) +calculate.add_command(library_normalization_revio) calculate.add_command(novaseq_x_denaturation) calculate.add_command(qpcr_concentration) calculate.add_command(calculate_saphyr_concentration) diff --git a/cg_lims/EPPs/udf/calculate/library_normalization.py b/cg_lims/EPPs/udf/calculate/library_normalization.py index e766f013..9888c2cc 100644 --- a/cg_lims/EPPs/udf/calculate/library_normalization.py +++ b/cg_lims/EPPs/udf/calculate/library_normalization.py @@ -6,33 +6,13 @@ from cg_lims import options from cg_lims.exceptions import InvalidValueError, LimsError, MissingValueError from cg_lims.get.artifacts import get_artifacts -from cg_lims.get.udfs import get_udf +from cg_lims.get.udfs import get_final_concentration, get_artifact_concentration, get_total_volume, get_process_total_volume from genologics.entities import Artifact, Process LOG = logging.getLogger(__name__) failed_samples = [] -def get_final_concentration(process: Process, final_concentration_udf: str) -> float: - """Return final concentration value from process.""" - return float(get_udf(entity=process, udf=final_concentration_udf)) - - -def get_artifact_concentration(artifact: Artifact, concentration_udf: str) -> float: - """Return concentration value from artifact.""" - return float(get_udf(entity=artifact, udf=concentration_udf)) - - -def get_total_volume(artifact: Artifact, total_volume_udf: str) -> float: - """Return total volume value from artifact.""" - return float(get_udf(entity=artifact, udf=total_volume_udf)) - - -def get_process_total_volume(process: Process, total_volume_udf: str) -> Optional[float]: - """Return total volume value from process.""" - return process.udf.get(total_volume_udf) - - def calculate_sample_volume( final_concentration: float, total_volume: float, sample_concentration: float, artifact: Artifact ) -> float: diff --git a/cg_lims/EPPs/udf/calculate/library_normalization_revio.py b/cg_lims/EPPs/udf/calculate/library_normalization_revio.py new file mode 100644 index 00000000..d016a632 --- /dev/null +++ b/cg_lims/EPPs/udf/calculate/library_normalization_revio.py @@ -0,0 +1,119 @@ +import logging +import sys +from typing import List, Optional + +import click +from cg_lims import options +from cg_lims.exceptions import InvalidValueError, LimsError, MissingValueError +from cg_lims.get.artifacts import get_artifacts +from cg_lims.get.udfs import get_final_concentration, get_artifact_concentration, get_artifact_volume +from genologics.entities import Artifact, Process + +LOG = logging.getLogger(__name__) +failed_samples = [] + + +def calculate_total_volume( + final_concentration: float, + sample_volume: float, + sample_concentration: float, + artifact: Artifact, +) -> float: + """Calculate and return the total volume needed to reach the desired final concentration.""" + if final_concentration > sample_concentration: + error_message: str = ( + f"The final concentration ({final_concentration} ng/ul) is higher than the original one" + f" ({sample_concentration} ng/ul) for sample {artifact.samples[0].id}. No dilution needed." + ) + LOG.error(error_message) + global failed_samples + failed_samples.append(artifact.name) + return sample_volume + return (sample_volume * sample_concentration) / final_concentration + + +def calculate_buffer_volume( + total_volume: float, sample_volume +) -> float: + """Calculate and return the buffer volume to dilute the sample with to the desired total volume""" + return total_volume - sample_volume + + +def set_artifact_volumes( + artifacts: List[Artifact], + final_concentration: float, + total_volume_udf: Optional[str], + sample_volume_udf: str, + buffer_volume_udf: str, + concentration_udf: str, +) -> None: + """Set volume UDFs on artifact level, given a list of artifacts, final concentration, and UDF names.""" + for artifact in artifacts: + sample_concentration: float = get_artifact_concentration( + artifact=artifact, concentration_udf=concentration_udf + ) + sample_volume: float = get_artifact_volume( + artifact=artifact, + sample_volume_udf=sample_volume_udf + ) + total_volume: float = calculate_total_volume( + final_concentration=final_concentration, + artifact=artifact, + sample_volume=sample_volume, + sample_concentration=sample_concentration + ) + buffer_volume: float = calculate_buffer_volume( + total_volume=total_volume, sample_volume=sample_volume + ) + artifact.udf[buffer_volume_udf] = buffer_volume + artifact.udf[total_volume_udf] = total_volume + artifact.put() + + +@click.command() +@options.input() +@options.sample_udf(help="Name of sample volume UDF.") +@options.buffer_udf(help="Name of buffer volume UDF.") +@options.concentration_udf(help="Name of sample concentration UDF.") +@options.final_concentration_udf(help="Name of final target concentration UDF.") +@options.total_volume_udf( + help="Name of total volume UDF on sample level. Note: Can't be combined with the process level alternative." +) +@click.pass_context +def library_normalization_revio( + ctx: click.Context, + input: bool, + sample_udf: str, + buffer_udf: str, + concentration_udf: str, + final_concentration_udf: str, + total_volume_udf: Optional[str] = None, +) -> None: + """Calculate the volumes for dilution for a set concentration with the total available sample volume""" + + LOG.info(f"Running {ctx.command_path} with params: {ctx.params}") + process: Process = ctx.obj["process"] + artifacts: List[Artifact] = get_artifacts(process=process, input=input) + try: + final_concentration: float = get_final_concentration( + process=process, final_concentration_udf=final_concentration_udf + ) + set_artifact_volumes( + artifacts=artifacts, + final_concentration=final_concentration, + total_volume_udf=total_volume_udf, + sample_volume_udf=sample_udf, + buffer_volume_udf=buffer_udf, + concentration_udf=concentration_udf, + ) + if failed_samples: + failed_samples_string = ", ".join(failed_samples) + error_message = f"The following artifacts had a lower concentration than targeted: {failed_samples_string}" + LOG.info(error_message) + raise InvalidValueError(error_message) + message: str = "Volumes were successfully calculated." + LOG.info(message) + click.echo(message) + except LimsError as e: + LOG.error(e.message) + sys.exit(e.message) diff --git a/cg_lims/get/udfs.py b/cg_lims/get/udfs.py index f6ad00d7..1edb4dd2 100644 --- a/cg_lims/get/udfs.py +++ b/cg_lims/get/udfs.py @@ -5,7 +5,7 @@ from cg_lims.exceptions import MissingUDFsError from cg_lims.get.artifacts import get_latest_analyte -from genologics.entities import Artifact, Entity +from genologics.entities import Artifact, Entity, Process from genologics.lims import Lims LOG = logging.getLogger(__name__) @@ -71,3 +71,28 @@ def get_analyte_udf(lims: Lims, sample_id: str, process_types: List[str], udf: s f"Couldn't find UDF '{udf}' for artifacts of sample {sample_id} generated by the step {process_types}" ) return value + + +def get_final_concentration(process: Process, final_concentration_udf: str) -> float: + """Return final concentration value from process.""" + return float(get_udf(entity=process, udf=final_concentration_udf)) + + +def get_artifact_concentration(artifact: Artifact, concentration_udf: str) -> float: + """Return concentration value from artifact.""" + return float(get_udf(entity=artifact, udf=concentration_udf)) + + +def get_artifact_volume(artifact: Artifact, sample_volume_udf: str) -> float: + """Return volume value from artifact.""" + return float(get_udf(entity=artifact, udf=sample_volume_udf)) + + +def get_total_volume(artifact: Artifact, total_volume_udf: str) -> float: + """Return total volume value from artifact.""" + return float(get_udf(entity=artifact, udf=total_volume_udf)) + + +def get_process_total_volume(process: Process, total_volume_udf: str) -> Optional[float]: + """Return total volume value from process.""" + return process.udf.get(total_volume_udf) \ No newline at end of file