diff --git a/cg_lims/EPPs/udf/calculate/calculate_amount_ng.py b/cg_lims/EPPs/udf/calculate/calculate_amount_ng.py index d2a025c1..6737b02d 100644 --- a/cg_lims/EPPs/udf/calculate/calculate_amount_ng.py +++ b/cg_lims/EPPs/udf/calculate/calculate_amount_ng.py @@ -1,11 +1,12 @@ 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 genologics.entities import Artifact, Process LOG = logging.getLogger(__name__) @@ -15,6 +16,8 @@ @options.amount_udf_option() @options.volume_udf_option() @options.subtract_volume_option() +@options.measurement() +@options.input() @click.pass_context def calculate_amount_ng( ctx: click.Context, @@ -22,6 +25,8 @@ def calculate_amount_ng( volume_udf: str, concentration_udf: str, subtract_volume: str, + measurement: bool = False, + input: bool = False, ): """Calculates and auto-fills the quantities of DNA in sample from concentration and volume measurements. The volume is subtracted by either 0 or 3 in the calculations. This is @@ -29,16 +34,17 @@ def calculate_amount_ng( LOG.info(f"Running {ctx.command_path} with params: {ctx.params}") - process = ctx.obj["process"] - lims = ctx.obj["lims"] + process: Process = ctx.obj["process"] try: - artifacts = get_artifacts(process=process, measurement=True) - missing_udfs_count = 0 - artifacts_with_missing_udf = [] + artifacts: List[Artifact] = get_artifacts( + process=process, measurement=measurement, input=input + ) + missing_udfs_count: int = 0 + artifacts_with_missing_udf: List = [] for artifact in artifacts: - vol = artifact.udf.get(volume_udf) - conc = artifact.udf.get(concentration_udf) + vol: float = artifact.udf.get(volume_udf) + conc: float = artifact.udf.get(concentration_udf) if None in [conc, vol]: missing_udfs_count += 1 artifacts_with_missing_udf.append(artifact.id) @@ -51,7 +57,7 @@ def calculate_amount_ng( f"Udf missing for {missing_udfs_count} artifact(s): " f"{','.join(artifacts_with_missing_udf)}. " ) - message = "Amounts have been calculated for all artifacts." + message: str = "Amounts have been calculated for all artifacts." LOG.info(message) click.echo(message) except LimsError as e: diff --git a/cg_lims/EPPs/udf/copy/sample_to_artifact.py b/cg_lims/EPPs/udf/copy/sample_to_artifact.py index 09797480..94dd2dd1 100644 --- a/cg_lims/EPPs/udf/copy/sample_to_artifact.py +++ b/cg_lims/EPPs/udf/copy/sample_to_artifact.py @@ -1,36 +1,49 @@ #!/usr/bin/env python 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_udf_type +from genologics.entities import Artifact, Process, Sample +from genologics.lims import Lims LOG = logging.getLogger(__name__) -def udf_copy_sample_to_artifact(artifacts: list, sample_udf: str, art_udf: str, lims) -> None: +def udf_copy_sample_to_artifact( + artifacts: List[Artifact], + sample_udf: str, + artifact_udf: str, + lims: Lims, + ignore_fail: bool = False, +) -> None: """Function to copy sample udf to artifact. - For each artifact in the artifacts list, picking the first sample in the art.samples list. - copying the samp_udf from the sample to the art_udf on the artifact. + For each artifact in the artifacts list, picking the first sample in the artifact.samples list. + copying the sample_udf from the sample to the art_udf on the artifact. Warns if sample_udf is missing on a sample. Logs if a the sample list of a artifact contains more than one sample. Arguments: artifacts: list of artifacts to copy from - art_udf: artifact udf to get - sample_udf: sample udf to set""" - failed_udfs = 0 - passed_udfs = 0 - udf_type = get_udf_type(lims=lims, udf_name=art_udf, attach_to_name=artifacts[0].output_type) - for art in artifacts: - if len(art.samples) > 1: + artifact_udf: artifact udf to get + sample_udf: sample udf to set + lims: genologics Lims object connected to the database + ignore_fail: bool option to mute warnings""" + failed_udfs: int = 0 + passed_udfs: int = 0 + udf_type = get_udf_type( + lims=lims, udf_name=artifact_udf, attach_to_name=artifacts[0].output_type + ) + for artifact in artifacts: + if len(artifact.samples) > 1: LOG.info("More than one sample per artifact, picking the first one") - sample = art.samples[0] + sample: Sample = artifact.samples[0] udf = sample.udf.get(sample_udf) if not isinstance(udf, udf_type): try: @@ -38,38 +51,47 @@ def udf_copy_sample_to_artifact(artifacts: list, sample_udf: str, art_udf: str, except: failed_udfs += 1 LOG.error( - f"Udf: {sample_udf} missing on sample {sample.id} or it could not be converted to type {udf_type}" + f"UDF: {sample_udf} missing on sample {sample.id} or it could not be converted to type {udf_type}" ) continue - art.udf[art_udf] = udf - art.put() + artifact.udf[artifact_udf] = udf + artifact.put() LOG.info( - f"copied udf {sample_udf} from sample {sample.id} to artifact: {art.id}, udf: {art_udf}" + f"copied UDF {sample_udf} from sample {sample.id} to artifact: {artifact.id}, UDF: {artifact_udf}" ) passed_udfs += 1 - if failed_udfs: + if failed_udfs and not ignore_fail: raise MissingUDFsError( - message=f"The udf '{sample_udf}' is missing for {failed_udfs} samples. Udfs were set on {passed_udfs} artifacts." + message=f"The UDF '{sample_udf}' is missing for {failed_udfs} samples. UDFs were set on {passed_udfs} artifacts." ) @click.command() -@options.sample_udf(help="Sample udf to set.") -@options.artifact_udf(help="Artifact udf to get.") -@options.measurement(help="Udfs will be set on measurements.") -@options.input(help="Udfs will be set on input artifacts.") +@options.sample_udf(help="Sample UDF to set.") +@options.artifact_udf(help="Artifact UDF to get.") +@options.measurement(help="UDFs will be set on measurements.") +@options.input(help="UDFs will be set on input artifacts.") +@options.ignore_fail(help="Script will not raise exception errors when not all UDFs are set.") @click.pass_context -def sample_to_artifact(ctx, sample_udf, artifact_udf, measurement, input): - """Script to copy artifact udf to sample udf""" +def sample_to_artifact(ctx, sample_udf, artifact_udf, measurement, input, ignore_fail): + """Script to copy artifact UDF to sample UDF""" - process = ctx.obj["process"] - lims = ctx.obj["lims"] + process: Process = ctx.obj["process"] + lims: Lims = ctx.obj["lims"] try: - artifacts = get_artifacts(process=process, input=input, measurement=measurement) - udf_copy_sample_to_artifact(artifacts, sample_udf, artifact_udf, lims) - click.echo("Udfs have been set on all samples.") + artifacts: List[Artifact] = get_artifacts( + process=process, input=input, measurement=measurement + ) + udf_copy_sample_to_artifact( + artifacts=artifacts, + sample_udf=sample_udf, + artifact_udf=artifact_udf, + lims=lims, + ignore_fail=ignore_fail, + ) + click.echo("UDFs have been set on all samples.") except LimsError as e: sys.exit(e.message)