diff --git a/cg_lims/EPPs/udf/calculate/aliquot_amount.py b/cg_lims/EPPs/udf/calculate/aliquot_amount.py new file mode 100644 index 00000000..d3c1d0ee --- /dev/null +++ b/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 + * . + + 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) diff --git a/cg_lims/EPPs/udf/calculate/base.py b/cg_lims/EPPs/udf/calculate/base.py index 27030625..087152e3 100644 --- a/cg_lims/EPPs/udf/calculate/base.py +++ b/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) diff --git a/cg_lims/EPPs/udf/calculate/calculate_buffer.py b/cg_lims/EPPs/udf/calculate/calculate_buffer.py index 7e524e08..bc15dc66 100644 --- a/cg_lims/EPPs/udf/calculate/calculate_buffer.py +++ b/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,6 +61,7 @@ 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, @@ -67,6 +69,7 @@ def volume_buffer( 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) diff --git a/cg_lims/EPPs/udf/calculate/twist_aliquot_amount.py b/cg_lims/EPPs/udf/calculate/twist_aliquot_amount.py deleted file mode 100644 index f6d86bd3..00000000 --- a/cg_lims/EPPs/udf/calculate/twist_aliquot_amount.py +++ /dev/null @@ -1,73 +0,0 @@ -import logging -import sys -from typing import List - -import click -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 - -LOG = logging.getLogger(__name__) - - -# The maximum amount taken into the prep is MAXIMUM_SAMPLE_AMOUNT. -# Any amount below this can be used in the prep if the total amount is limited. -MAXIMUM_SAMPLE_AMOUNT = 250 - - -def get_possible_input_amount(artifact: Artifact) -> float: - """Return the maximum input amount possible for a sample, depending on the total volume required.""" - process = artifact.parent_process - sample_volume = artifact.udf.get("Volume (ul)") - total_volume = process.udf.get("Total Volume (ul)") - max_volume = min(sample_volume, total_volume) - concentration = artifact.udf.get("Concentration") - return max_volume * concentration - - -def get_maximum_input_for_aliquot(artifact: Artifact) -> float: - """Return the maximum allowed input amount for the specified artifact.""" - possible_input_amount = get_possible_input_amount(artifact=artifact) - max_input_amount = 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]): - """The maximum amount taken into the prep is decided by calculating the minimum between MAXIMUM_SAMPLE_AMOUNT and - * . - - Any amount below this can be used in the prep if the total amount or sample volume is limited. - """ - - missing_udfs = 0 - for artifact in artifacts: - if not artifact.udf.get("Concentration") or not artifact.udf.get("Volume (ul)"): - missing_udfs += 1 - continue - maximum_amount = get_maximum_input_for_aliquot(artifact=artifact) - artifact.udf["Amount needed (ng)"] = maximum_amount - artifact.put() - - if missing_udfs: - raise MissingUDFsError(f"UDF missing for {missing_udfs} samples") - - -@click.command() -@click.pass_context -def twist_aliquot_amount(ctx): - """Calculates amount needed for samples.""" - - LOG.info(f"Running {ctx.command_path} with params: {ctx.params}") - - process = ctx.obj["process"] - - try: - artifacts = get_artifacts(process=process, input=False) - set_amount_needed(artifacts) - message = "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) diff --git a/cg_lims/options.py b/cg_lims/options.py index 6d8cdfb0..374c8b9a 100644 --- a/cg_lims/options.py +++ b/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)