Added workflows to register CLEM preprocessing results (#396)

Murfey can now recognise messages from the LIF-to-image stack and TIFF-to-image stack processes, and register the files and metadata associated with that batch job in the PostgreSQL database accordingly.

Author: tieneupin

pyproject.toml

@@ -98,8 +98,10 @@ murfey = "murfey.client:run"
 [project.entry-points."murfey.config.extraction"]
 "murfey_machine" = "murfey.util.config:get_extended_machine_config"
 [project.entry-points."murfey.workflows"]
-"lif_to_stack" = "murfey.workflows.lif_to_stack:zocalo_cluster_request"
-"tiff_to_stack" = "murfey.workflows.tiff_to_stack:zocalo_cluster_request"
+"process_raw_lifs" = "murfey.workflows.clem.process_raw_lifs:zocalo_cluster_request"
+"process_raw_tiffs" = "murfey.workflows.clem.process_raw_tiffs:zocalo_cluster_request"
+"register_lif_preprocessing_result" = "murfey.workflows.clem.register_preprocessing_results:register_lif_preprocessing_result"
+"register_tiff_preprocessing_result" = "murfey.workflows.clem.register_preprocessing_results:register_tiff_preprocessing_result"
 
 [tool.setuptools]
 package-dir = {"" = "src"}

src/murfey/client/contexts/clem.py

@@ -386,7 +386,7 @@ def process_lif_file(
 
         try:
             # Construct the URL to post the request to
-            url = f"{str(environment.url.geturl())}/sessions/{environment.murfey_session}/lif_to_stack?lif_file={quote(str(lif_file), safe='')}"
+            url = f"{str(environment.url.geturl())}/sessions/{environment.murfey_session}/clem/preprocessing/process_raw_lifs?lif_file={quote(str(lif_file), safe='')}"
             # Validate
             if not url:
                 logger.error(
@@ -442,7 +442,7 @@ def process_tiff_series(
 
         try:
             # Construct URL for Murfey server to communicate with
-            url = f"{str(environment.url.geturl())}/sessions/{environment.murfey_session}/tiff_to_stack"
+            url = f"{str(environment.url.geturl())}/sessions/{environment.murfey_session}/clem/preprocessing/process_raw_tiffs"
             if not url:
                 logger.error(
                     "URL could not be constructed from the environment and file path"

src/murfey/server/__init__.py

@@ -56,6 +56,9 @@
     from murfey.server.ispyb import TransportManager  # Session
 except AttributeError:
     pass
+from backports.entry_points_selectable import entry_points
+from importlib_metadata import EntryPoint  # For type hinting only
+
 import murfey.util.db as db
 from murfey.util import LogFilter
 from murfey.util.spa_params import default_spa_parameters
@@ -2964,6 +2967,26 @@ def feedback_callback(header: dict, message: dict) -> None:
             if _transport_object:
                 _transport_object.transport.ack(header)
             return None
+        elif (
+            message["register"] in entry_points().select(group="murfey.workflows").names
+        ):
+            # Run the workflow if a match is found
+            workflow: EntryPoint = list(  # Returns a list of either 1 or 0
+                entry_points().select(
+                    group="murfey.workflows", name=message["register"]
+                )
+            )[0]
+            result = workflow.load()(
+                message=message,
+                db=murfey_db,
+            )
+            if _transport_object:
+                if result:
+                    _transport_object.transport.ack(header)
+                else:
+                    # Send it directly to DLQ without trying to rerun it
+                    _transport_object.transport.nack(header, requeue=False)
+            return None
         if _transport_object:
             _transport_object.transport.nack(header, requeue=False)
         return None

src/murfey/server/api/clem.py

@@ -1,12 +1,13 @@
 from __future__ import annotations
 
 import re
-import sys
 import traceback
+from importlib.metadata import EntryPoint  # type hinting only
 from logging import getLogger
 from pathlib import Path
 from typing import Optional, Type, Union
 
+from backports.entry_points_selectable import entry_points
 from fastapi import APIRouter
 from sqlalchemy.exc import NoResultFound
 from sqlmodel import Session, select
@@ -22,13 +23,7 @@
     CLEMTIFFFile,
 )
 from murfey.util.db import Session as MurfeySession
-from murfey.util.models import TiffSeriesInfo
-
-# Use backport from importlib_metadata for Python <3.10
-if sys.version_info.major == 3 and sys.version_info.minor < 10:
-    from importlib_metadata import EntryPoint, entry_points
-else:
-    from importlib.metadata import EntryPoint, entry_points
+from murfey.util.models import TIFFSeriesInfo
 
 # Set up logger
 logger = getLogger("murfey.server.api.clem")
@@ -81,23 +76,15 @@ def validate_and_sanitise(
     machine_config = get_machine_config(instrument_name=instrument_name)[
         instrument_name
     ]
-    rsync_basepath = machine_config.rsync_basepath
-    try:
-        base_path = list(rsync_basepath.parents)[-2].as_posix()
-    except IndexError:
-        # Print to troubleshoot
-        logger.warning(f"Base path {rsync_basepath!r} is too short")
-        base_path = rsync_basepath.as_posix()
-    except Exception:
-        raise Exception("Unexpected exception occurred when loading the file base path")
+    base_path = machine_config.rsync_basepath.as_posix()
 
     # Check that full file path doesn't contain unallowed characters
-    # Currently allows only:
-    # - words (alphanumerics and "_"; \w),
-    # - spaces (\s),
-    # - periods,
-    # - dashes,
-    # - forward slashes ("/")
+    #   Currently allows only:
+    #   - words (alphanumerics and "_"; \w),
+    #   - spaces (\s),
+    #   - periods,
+    #   - dashes,
+    #   - forward slashes ("/")
     if bool(re.fullmatch(r"^[\w\s\.\-/]+$", str(full_path))) is False:
         raise ValueError(f"Unallowed characters present in {file}")
 
@@ -631,51 +618,68 @@ def register_image_stack(
 """
 
 
-@router.post("/sessions/{session_id}/lif_to_stack")  # API posts to this URL
-def lif_to_stack(
+@router.post(
+    "/sessions/{session_id}/clem/preprocessing/process_raw_lifs"
+)  # API posts to this URL
+def process_raw_lifs(
     session_id: int,  # Used by the decorator
     lif_file: Path,
+    db: Session = murfey_db,
 ):
-    # Get command line entry point
-    murfey_workflows = entry_points().select(
-        group="murfey.workflows", name="lif_to_stack"
-    )
-
-    # Use entry point if found
-    if len(murfey_workflows) == 1:
-        workflow: EntryPoint = list(murfey_workflows)[0]
-        workflow.load()(
-            # Match the arguments found in murfey.workflows.lif_to_stack
-            file=lif_file,
-            root_folder="images",
-            messenger=_transport_object,
-        )
-        return True
-    # Raise error if Murfey workflow not found
-    else:
+    try:
+        # Try and load relevant Murfey workflow
+        workflow: EntryPoint = list(
+            entry_points().select(group="murfey.workflows", name="process_raw_lifs")
+        )[0]
+    except IndexError:
         raise RuntimeError("The relevant Murfey workflow was not found")
 
+    # Get instrument name from the database to load the correct config file
+    session_row: MurfeySession = db.exec(
+        select(MurfeySession).where(MurfeySession.id == session_id)
+    ).one()
+    instrument_name = session_row.instrument_name
+
+    # Pass arguments along to the correct workflow
+    workflow.load()(
+        # Match the arguments found in murfey.workflows.clem.process_raw_lifs
+        file=lif_file,
+        root_folder="images",
+        session_id=session_id,
+        instrument_name=instrument_name,
+        messenger=_transport_object,
+    )
+    return True
+
 
-@router.post("/sessions/{session_id}/tiff_to_stack")
-def tiff_to_stack(
+@router.post("/sessions/{session_id}/clem/preprocessing/process_raw_tiffs")
+def process_raw_tiffs(
     session_id: int,  # Used by the decorator
-    tiff_info: TiffSeriesInfo,
+    tiff_info: TIFFSeriesInfo,
+    db: Session = murfey_db,
 ):
-    # Get command line entry point
-    murfey_workflows = entry_points().select(
-        group="murfey.workflows", name="tiff_to_stack"
-    )
-
-    # Use entry point if found
-    if murfey_workflows:
-        workflow: EntryPoint = list(murfey_workflows)[0]
-        workflow.load()(
-            # Match the arguments found in murfey.workflows.tiff_to_stack
-            file=tiff_info.tiff_files[0],  # Pass it only one file from the list
-            root_folder="images",
-            metadata=tiff_info.series_metadata,
-            messenger=_transport_object,
-        )
-    # Raise error if Murfey workflow not found
-    else:
+    try:
+        # Try and load relevant Murfey workflow
+        workflow: EntryPoint = list(
+            entry_points().select(group="murfey.workflows", name="process_raw_tiffs")
+        )[0]
+    except IndexError:
         raise RuntimeError("The relevant Murfey workflow was not found")
+
+    # Get instrument name from the database to load the correct config file
+    session_row: MurfeySession = db.exec(
+        select(MurfeySession).where(MurfeySession.id == session_id)
+    ).one()
+    instrument_name = session_row.instrument_name
+
+    # Pass arguments to correct workflow
+    workflow.load()(
+        # Match the arguments found in murfey.workflows.clem.process_raw_tiffs
+        tiff_list=tiff_info.tiff_files,
+        root_folder="images",
+        session_id=session_id,
+        instrument_name=instrument_name,
+        metadata=tiff_info.series_metadata,
+        messenger=_transport_object,
+    )
+    return True

src/murfey/server/murfey_db.py

@@ -1,6 +1,5 @@
 from __future__ import annotations
 
-import os
 from functools import partial
 
 import yaml
@@ -11,8 +10,6 @@
 
 from murfey.util.config import Security, get_security_config
 
-instrument_name = os.getenv("BEAMLINE", "")
-
 
 def url(security_config: Security | None = None) -> str:
     security_config = security_config or get_security_config()

src/murfey/util/db.py

@@ -269,9 +269,12 @@ class CLEMImageSeries(SQLModel, table=True):  # type: ignore
     )  # One to many
 
     # Process checklist for series
-    images_aligned: bool = False  # Image stacks aligned to reference image
-    rgbs_created: bool = False  # Image stacks all colorised
-    composite_created: bool = False  # Composite flattened image created
+    number_of_members: int = (
+        0  # Expected number of image stacks belonging to this series
+    )
+    images_aligned: bool = False  # Have all members been aligned?
+    rgbs_created: bool = False  # Have all members been colourised?
+    composite_created: bool = False  # Has a composite image been created?
     composite_image: Optional[str] = None  # Full path to composite image
 
 

src/murfey/util/models.py

@@ -1,10 +1,11 @@
 from __future__ import annotations
 
+from ast import literal_eval
 from datetime import datetime
 from pathlib import Path
 from typing import Any, Dict, List, Optional
 
-from pydantic import BaseModel
+from pydantic import BaseModel, validator
 
 """
 General Models
@@ -154,12 +155,46 @@ class FractionationParameters(BaseModel):
 """
 
 
-class TiffSeriesInfo(BaseModel):
+class TIFFSeriesInfo(BaseModel):
     series_name: str
     tiff_files: List[Path]
     series_metadata: Path
 
 
+class LIFPreprocessingResult(BaseModel):
+    image_stack: Path
+    metadata: Path
+    series_name: str
+    channel: str
+    number_of_members: int
+    parent_lif: Path
+
+
+class TIFFPreprocessingResult(BaseModel):
+    image_stack: Path
+    metadata: Path
+    series_name: str
+    channel: str
+    number_of_members: int
+    parent_tiffs: list[Path]
+
+    @validator(
+        "parent_tiffs",
+        pre=True,
+    )
+    def parse_stringified_list(cls, value):
+        if isinstance(value, str):
+            try:
+                eval_result = literal_eval(value)
+                if isinstance(eval_result, list):
+                    parent_tiffs = [Path(p) for p in eval_result]
+                    return parent_tiffs
+            except (SyntaxError, ValueError):
+                raise ValueError("Unable to parse input")
+        # Return value as-is; if it fails, it fails
+        return value
+
+
 """
 FIB
 ===