Skip to content

Commit

Permalink
Skip RC QC EPP updates (#470)(minor)
Browse files Browse the repository at this point in the history
### Added
- new option to skip error messages to the copy EPP sample_to_artifact.py
- new measurement and input artifact flags for the calculate EPP calculate_amount_ng.py
  • Loading branch information
Karl-Svard authored Jan 19, 2024
1 parent 2447e99 commit eb361bd
Show file tree
Hide file tree
Showing 2 changed files with 65 additions and 37 deletions.
24 changes: 15 additions & 9 deletions cg_lims/EPPs/udf/calculate/calculate_amount_ng.py
Original file line number Diff line number Diff line change
@@ -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__)

Expand All @@ -15,30 +16,35 @@
@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,
amount_udf: str,
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
because the lab uses 0 or 3 ul in the initial qc measurements."""

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)
Expand All @@ -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:
Expand Down
78 changes: 50 additions & 28 deletions cg_lims/EPPs/udf/copy/sample_to_artifact.py
Original file line number Diff line number Diff line change
@@ -1,75 +1,97 @@
#!/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:
udf = udf_type(udf)
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)

0 comments on commit eb361bd

Please sign in to comment.