Merge branch 'master' into hashing

Author: kedhammar

VERSIONLOG.md

@@ -1,9 +1,43 @@
 # Scilifelab_epps Version Log
 
-## 20250403.1
+## 20250509.1
 
 Create project automation to apply sample antibodies from CSV to UDFs.
 
+## 20250411.2
+
+Change SQL query for finding sample-label mappings in a pool to use pool artifact backtracking rather than using a submitted sample global query.
+
+## 20250411.1
+
+Introduce EPP for generic UDF calculations.
+
+## 20250410.3
+
+Change get_epp_user query to account for differet epp triggers
+
+## 20250410.2
+
+Patch last PR, don't include file slot artifacts in logic for parsing demux artifacts.
+
+## 20250410.1
+
+Introduce new EPP to fix bugged demux artifacts with multiple samples in the name by using sample-label linkage to identify the "correct" sample and rename the demux artifact accordingly.
+Demux EPP: Add logic block to handle demux artifacts being tied to multiple samples, in which case try to choose a single sample matching the name of the demux artifact.
+Utility function for mapping samples to labels in pool: Move into main module
+
+## 20250408.1
+
+Set up fall-back for get_epp_user() failing in wrapper.
+
+## 20250405.1
+
+Convert list of artifacts to set to get unique values to avoid repeating lines in running notes
+
+## 20250431.1
+
+Update EPP wrapper to send log messages to stdout.
+
 ## 20250328.1
 
 Add script to add running notes from the step Load to Flowcell (NovaSeqXPlus) v1.0.

scilifelab_epps/epp.py

@@ -16,6 +16,8 @@
 from shutil import copy
 from time import localtime, strftime
 
+import psycopg2
+import yaml
 from genologics.config import MAIN_LOG
 from genologics.entities import Artifact, Process
 from genologics.lims import Lims
@@ -568,3 +570,78 @@ def upload_file(
     if remove:
         os.remove(file_path)
         logging.info(f"'{file_path}' removed from local filesystem.")
+
+
+def get_pool_sample_label_mapping(pool: Artifact) -> dict[str, str]:
+    """Given a pool artifact containing labeled samples, use database queries to
+    build a dictionary mapping each sample name to its reagent label.
+    """
+    with open("/opt/gls/clarity/users/glsai/config/genosqlrc.yaml") as f:
+        config = yaml.safe_load(f)
+
+    # Setup DB connection
+    connection = psycopg2.connect(
+        user=config["username"],
+        host=config["url"],
+        database=config["db"],
+        password=config["password"],
+    )
+    cursor = connection.cursor()
+
+    """Supply a pool artifact ID and a sample name:
+    1. Find the ancestor artifacts of the pool artifact.
+    2. Filter for derived sample artifacts
+    3. Filter for artifacts sharing a name with the target sample
+    4. Filter for artifacts with reagent labels
+    """
+    query = """--sql
+        SELECT
+            DISTINCT rl.name
+        FROM
+            -- Table mapping artifact IDs to ancestor artifact IDs
+            artifact_ancestor_map aam
+            -- Join artifact information on ancestor artifact IDs
+            JOIN artifact parent ON aam.ancestorartifactid = parent.artifactid
+            -- Join reagent label information on ancestor artifact IDs
+            LEFT JOIN artifact_label_map alm ON parent.artifactid = alm.artifactid
+            LEFT JOIN reagentlabel rl ON rl.labelid = alm.labelid
+        WHERE
+            -- The pool artifact ID is used to find its ancestors
+            aam.artifactid = {}
+            -- Filter for derived sample artifacts
+            AND parent.artifacttypeid = 2
+            -- Filter for artifacts sharing a name with the target sample
+            AND parent.name = '{}'
+            -- Filter for artifacts with reagent labels
+            AND rl.name IS NOT NULL;
+    """
+
+    errors = False
+    sample2label = {}
+    pool_db_id = int(pool.id.split("-")[1])
+    for sample in pool.samples:
+        try:
+            cursor.execute(query.format(pool_db_id, sample.name))
+            query_results = cursor.fetchall()
+
+            assert len(query_results) != 0, (
+                f"No reagent labels found for sample '{sample.name}'."
+            )
+            assert len(query_results) == 1, (
+                f"Multiple reagent labels found for sample '{sample.name}'."
+            )
+
+            label = query_results[0][0]
+            sample2label[sample.name] = label
+        except AssertionError as e:
+            logging.error(str(e), exc_info=True)
+            logging.warning(f"Skipping sample '{sample.name}' due to error.")
+            errors = True
+            continue
+
+    if errors:
+        raise AssertionError(
+            "Errors occurred when linking samples and indices. Please report this error."
+        )
+    else:
+        return sample2label

scilifelab_epps/utils/get_epp_user.py

@@ -15,17 +15,22 @@ def get_epp_user(lims, procid):
     with open("/opt/gls/clarity/users/glsai/config/genosqlrc.yaml") as f:
         config = yaml.safe_load(f)
 
-    query = (
-        "SELECT ps.researcherid "
-        "FROM principals ps "
-        "JOIN externalprogramstatus eps ON ps.principalid = eps.ownerid "
-        "JOIN auditchangelog acl ON acl.rowpk = CAST(eps.externalprogramstatusid AS TEXT) "
-        "AND acl.tablename = 'externalprogramstatus' "
-        "JOIN auditeventlog ael ON ael.eventid = acl.eventid "
-        "WHERE ael.eventtype = 'EPP_EXECUTE' "
-        f"AND eps.processid = {procid.split('-')[1]} "
-        "AND eps.status = 'RUNNING';"
-    )
+    # When an epp is launched, 2 events are created in auditeventlog, one for event (i.e. advancing in a step,
+    # clicking the button and so on) and one for consumption of the next EPP request by the API. Both of them map to the
+    # the same rowpk in auditchangelog. This rowpk maps to the externalprogramstatusid in the table externalprogramstatus.
+    # So by checking for 'EPP_CONSUME_NEXT_REQUEST' and the process thats currently running we should be able to get
+    # the user that launched the EPP.
+    query = f"""--sql
+        SELECT ps.researcherid
+        FROM principals ps
+        JOIN externalprogramstatus eps ON ps.principalid = eps.ownerid
+        JOIN auditchangelog acl ON acl.rowpk = CAST(eps.externalprogramstatusid AS TEXT)
+        AND acl.tablename = 'externalprogramstatus'
+        JOIN auditeventlog ael ON ael.eventid = acl.eventid
+        WHERE ael.eventtype ='EPP_CONSUME_NEXT_REQUEST'
+        AND eps.processid = {procid.split("-")[1]}
+        AND eps.status = 'RUNNING';
+     """
 
     with psycopg2.connect(
         user=config["username"],

scilifelab_epps/utils/udf_tools.py

@@ -224,7 +224,7 @@ def fetch_last(
                         logging.info(
                             f"Found target UDF '{target_udf}'"
                             + f" with value '{current_art.udf[target_udf]}'"
-                            + f" in process {steps_visited[-1]}"
+                            + f" in process {steps_visited[-1] if pp else ''}"
                             + f" {'output' if pp else 'input'}"
                             + f" artifact '{current_art.name}' ({current_art.id})"
                         )

scilifelab_epps/wrapper.py

@@ -5,10 +5,29 @@
 import sys
 
 from genologics.config import BASEURI, PASSWORD, USERNAME
-from genologics.entities import Process
+from genologics.entities import Process, Researcher
 from genologics.lims import Lims
 
 from scilifelab_epps.epp import upload_file
+from scilifelab_epps.utils.get_epp_user import get_epp_user
+
+
+class TrackingRootLogger(logging.RootLogger):
+    """A root logger that tracks whether any errors or warnings have been emitted."""
+
+    def __init__(self, level):
+        """Initialize with the given level and errors_or_warnings set to False."""
+        super().__init__(level)
+        self.errors_or_warnings = False
+
+    def handle(self, record):
+        """Handle a log record and track if it's an error or warning."""
+        # Check if the record is a warning or error before handling it
+        if record.levelno >= logging.WARNING:
+            self.errors_or_warnings = True
+
+        # Let the parent class handle the record normally
+        return super().handle(record)
 
 
 def epp_decorator(script_path: str, timestamp: str):
@@ -28,29 +47,55 @@ def epp_wrapper(args):
             lims.check_version()
             process = Process(lims, id=args.pid)
 
+            # Get EPP user
+            try:
+                epp_user: Researcher = get_epp_user(lims, args.pid)
+            except ValueError:
+                epp_user = None
+
             # Name log file
             log_filename: str = (
                 "_".join(
                     [
                         script_name,
                         process.id,
                         timestamp,
-                        process.technician.name.replace(" ", ""),
+                        (epp_user or process.technician).name.replace(" ", ""),
                     ]
                 )
                 + ".log"
             )
 
             # Set up logging
-            logging.basicConfig(
-                filename=log_filename,
-                filemode="w",
-                format="%(levelname)s: %(message)s",
-                level=logging.INFO,
-            )
+
+            # Set custom subclass as root logger
+            logger = TrackingRootLogger(level=logging.INFO)
+            logging.root = logger
+
+            # Clear any existing handlers (to avoid duplicates)
+            for handler in logger.handlers[:]:
+                logger.removeHandler(handler)
+
+            # Create file handler
+            file_handler = logging.FileHandler(log_filename, mode="w")
+            file_handler.setLevel(logging.INFO)
+            file_handler.setFormatter(logging.Formatter("%(levelname)s: %(message)s"))
+
+            # Create stdout handler
+            stdout_handler = logging.StreamHandler(sys.stdout)
+            stdout_handler.setLevel(logging.INFO)
+            stdout_handler.setFormatter(logging.Formatter("%(levelname)s: %(message)s"))
+
+            # Add both handlers to logger
+            logger.addHandler(file_handler)
+            logger.addHandler(stdout_handler)
 
             # Start logging
-            logging.info(f"Script '{script_name}' started at {timestamp}.")
+            if not epp_user:
+                logging.warning("No EPP user found for process ID {args.pid}.")
+            logging.info(
+                f"Script '{script_name}' started at {timestamp} by {(epp_user.name if epp_user else 'unknown')}."
+            )
             logging.info(
                 f"Launched in step '{process.type.name}' ({process.id}) opened by {process.technician.name}."
             )
@@ -80,23 +125,23 @@ def epp_wrapper(args):
 
             # On script success
             else:
-                logging.info("Script completed successfully.")
+                logging.info("Script finished successfully. Uploading log file.")
                 logging.shutdown()
                 upload_file(
                     file_path=log_filename,
                     file_slot=args.log,
                     process=process,
                     lims=lims,
+                    remove=True,
                 )
                 # Check log for errors and warnings
-                log_content = open(log_filename).read()
-                os.remove(log_filename)
-                if "ERROR:" in log_content or "WARNING:" in log_content:
+                if logger.errors_or_warnings:
                     sys.stderr.write(
                         "Script finished successfully, but log contains errors or warnings, please have a look."
                     )
                     sys.exit(2)
                 else:
+                    sys.stdout.write("Script finished successfully.")
                     sys.exit(0)
 
         return epp_wrapper