From ff412b0b85052b98b7149861688e5891d62d10e8 Mon Sep 17 00:00:00 2001 From: kedhammar Date: Fri, 20 Dec 2024 12:57:58 +0100 Subject: [PATCH] live testing dev --- scilifelab_epps/utils/udf_tools.py | 43 ++++++++++++++++++++++-------- scripts/fetch_last_known_field.py | 43 +++++++++++++++++++++--------- 2 files changed, 62 insertions(+), 24 deletions(-) diff --git a/scilifelab_epps/utils/udf_tools.py b/scilifelab_epps/utils/udf_tools.py index 278b196d..22ff69e4 100644 --- a/scilifelab_epps/utils/udf_tools.py +++ b/scilifelab_epps/utils/udf_tools.py @@ -136,19 +136,36 @@ def list_udfs(art: Artifact) -> list: def fetch_last( currentStep: Process, - art_tuple: tuple, target_udfs: str | list, + art_tuple: tuple = None, + art: Artifact = None, use_current=True, print_history=False, on_fail=AssertionError, ): """Recursively look for target UDF. - Target UDF can be supplied as a string, or as a prioritized list of strings. + Arguments: + + - "art_tuple": step input-output tuple. Mutually exclusive use with "art". + + - "art": step artifact, either input or output. Mutually exclusive use with "art_tuple". + + - "target_udfs": can be supplied as a string, or as a + prioritized list of strings. - If "print_history" == True, will return both the target metric and the lookup history as a string. + - "use_current": if true, will return the target metric + if found in the current step. + + - "print_history": if true, will return both the target + metric and the lookup history as a string. """ + assert art_tuple or art, "One of function args 'art_tuple' and 'art' are required." + assert not ( + art_tuple and art + ), "Function args 'art_tuple' and 'art' are mutually exclusive." + # Convert to list, to enable iteration if isinstance(target_udfs, str): target_udfs = [target_udfs] @@ -158,15 +175,19 @@ def fetch_last( while True: history.append({"Step name": currentStep.type.name, "Step ID": currentStep.id}) - # Try to grab input and output articles, if possible - try: - input_art = art_tuple[0]["uri"] - except: - input_art = None - try: - output_art = art_tuple[1]["uri"] - except: + if len(history) == 1 and not art_tuple: + # Handle the case of having an art instead of an art_tuple in the original step + input_art = art output_art = None + else: + try: + input_art = art_tuple[0]["uri"] + except: + input_art = None + try: + output_art = art_tuple[1]["uri"] + except: + output_art = None if len(history) == 1 and use_current is not True: # If we are in the original step and "use_current" is false, skip diff --git a/scripts/fetch_last_known_field.py b/scripts/fetch_last_known_field.py index fc59ab49..dae82767 100644 --- a/scripts/fetch_last_known_field.py +++ b/scripts/fetch_last_known_field.py @@ -4,7 +4,7 @@ from datetime import datetime as dt from genologics.config import BASEURI, PASSWORD, USERNAME -from genologics.entities import Process +from genologics.entities import Artifact, Process from genologics.lims import Lims from scilifelab_epps.utils import udf_tools @@ -30,41 +30,58 @@ def main(args): lims = Lims(BASEURI, USERNAME, PASSWORD) process = Process(lims, id=args.pid) + # Get the target UDF from the step field target_udf = process.udf.get(args.step_udf, None) - if target_udf is None or target_udf == "None": - logging.error(f"No target UDF supplied from step field '{args.step_udf}'") + assert ( + target_udf is not None or target_udf != "None" + ), f"No target UDF supplied from step field '{args.step_udf}'" - no_outputs = udf_tools.no_outputs(process) + # Check whether process has output artifacts, not the case for e.g. QC steps + no_outputs: bool = udf_tools.no_outputs(process) + # Load input artifacts + arts_in: list[Artifact] = [ + art for art in process.all_inputs() if art.type == "Analyte" + ] + + # Find target output artifacts, if any if no_outputs: logging.info("Step has no output artifacts. Assigning to input artifact.") + else: + art_tuples: list[tuple[dict]] = process.input_output_maps + art_in2out: dict[Process.Artifact : Process.Artifact] = { + i["uri"]: o["uri"] + for i, o in art_tuples + if i["uri"].type == "Analyte" and o["uri"].type == "Analyte" + } - # TODO need to tweak this script and possible the traceback function to handle both - # TODO aggregate QC and regular steps - art_tuples = udf_tools.get_art_tuples(process) # TODO this returns [] - for art_tuple in art_tuples: - target_artifact = art_tuple[0]["uri"] if no_outputs else art_tuple[1]["uri"] + for art_in in arts_in: + if no_outputs: + target_artifact = art_in + else: + target_artifact = art_in2out[art_in] logging.info( - f"Looking for last recorded UDF '{target_udf}' of sample '{target_artifact.name}'..." + f"Looking for last recorded UDF '{target_udf}' of {'input' if no_outputs else 'output'} artifact '{target_artifact.name}'..." ) udf_value, udf_history = udf_tools.fetch_last( currentStep=process, - art_tuple=art_tuple, + art=art_in, target_udfs=target_udf, use_current=False, print_history=True, on_fail=None, ) if udf_value: + logging.info(f"Found target UDF '{target_udf}' with value '{udf_value}'") logging.info(f"Traceback:\n{udf_history}") target_artifact.udf[target_udf] = udf_value target_artifact.put() logging.info( - f"Updated UDF '{target_udf}' for '{art_tuple[1]['uri'].name}' to '{udf_value}'" + f"Updated UDF '{target_udf}' for '{art_in.name}' to '{udf_value}'" ) else: logging.warning( - f"Could not traceback UDF '{target_udf}' for '{art_tuple[1]['uri'].name}'" + f"Could not traceback UDF '{target_udf}' for '{art_in.name}'" ) logging.info(f"Traceback:\n{udf_history}")