Added workflows to register CLEM preprocessing results

pyproject.toml

@@ -97,9 +97,11 @@ murfey = "murfey.client:run"
 "password" = "murfey.server.api.auth:password_token_validation"
 [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"
+[project.entry-points."murfey.workflows.clem"]
+"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 @@
 
         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 @@
 
         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

@@ -5,6 +5,7 @@
 import math
 import os
 import subprocess
+import sys
 import time
 from datetime import datetime
 from functools import partial, singledispatch
@@ -61,6 +62,12 @@
 from murfey.util.spa_params import default_spa_parameters
 from murfey.util.state import global_state
 
+# Import entry_points depending on Python version
+if sys.version_info < (3, 10):
+    from importlib_metadata import EntryPoint, entry_points
+else:
+    from importlib.metadata import EntryPoint, entry_points
+
 try:
     from importlib.resources import files  # type: ignore
 except ImportError:
@@ -2964,6 +2971,29 @@
             if _transport_object:
                 _transport_object.transport.ack(header)
             return None
+        elif message["register"] in (
+            "register_lif_preprocessing_result",
+            "register_tiff_preprocessing_result",
+        ):
+            murfey_workflows = list(
+                entry_points().select(
+                    group="murfey.workflows.clem", name=message["register"]
+                )
+            )
+            # Run the workflow if a match is found
+            if len(murfey_workflows) > 0:
+                workflow: EntryPoint = murfey_workflows[0]
+                workflow.load()(
+                    message=message,
+                    db=murfey_db,
+                )
+                if _transport_object:
+                    _transport_object.transport.ack(header)
+            # Nack message if no workflow found for message
+            else:
+                if _transport_object:
+                    _transport_object.transport.nack(header)
+            return None
         if _transport_object:
             _transport_object.transport.nack(header, requeue=False)
         return None

src/murfey/server/api/clem.py

@@ -22,7 +22,7 @@
     CLEMTIFFFile,
 )
 from murfey.util.db import Session as MurfeySession
-from murfey.util.models import TiffSeriesInfo
+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:
@@ -631,51 +631,79 @@
 """
 
 
-@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"
+        group="murfey.workflows.clem", name="process_raw_lifs"
     )
 
     # Use entry point if found
-    if len(murfey_workflows) == 1:
+    if murfey_workflows:
+
+        # Get instrument name from the database
+        # This is needed in order to load the correct config file
+        row: MurfeySession = db.exec(
+            select(MurfeySession).where(MurfeySession.id == session_id)
+        ).one()
+        instrument_name = row.instrument_name
+
+        # Pass arguments along to the correct workflow
         workflow: EntryPoint = list(murfey_workflows)[0]
         workflow.load()(
-            # Match the arguments found in murfey.workflows.lif_to_stack
+            # Match the arguments found in murfey.workflows.process_raw_lifs
             file=lif_file,
             root_folder="images",
+            session_id=session_id,
+            instrument_name=instrument_name,
             messenger=_transport_object,
         )
         return True
+
     # Raise error if Murfey workflow not found
     else:
         raise RuntimeError("The relevant Murfey workflow was not found")
 
 
-@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"
+        group="murfey.workflows.clem", name="process_raw_tiffs"
     )
 
     # Use entry point if found
     if murfey_workflows:
+
+        # Load instrument name from database
+        row: MurfeySession = db.exec(
+            select(MurfeySession).where(MurfeySession.id == session_id)
+        ).one()
+        instrument_name = row.instrument_name
+
+        # Pass arguments to correct workflow
         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
+            # Match the arguments found in murfey.workflows.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
+
     # Raise error if Murfey workflow not found
     else:
         raise RuntimeError("The relevant Murfey workflow was not found")

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 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
 ===

src/murfey/workflows/clem/process_raw_lifs.py

@@ -0,0 +1,70 @@
+"""
+Script to allow Murfey to submit the LIF-to-STACK job to the cluster.
+The recipe referred to here is stored on GitLab.
+"""
+
+from pathlib import Path
+from typing import Optional
+
+from murfey.util.config import get_machine_config
+
+try:
+    from murfey.server.ispyb import TransportManager  # Session
+except AttributeError:
+    pass  # Ignore if ISPyB credentials environment variable not set
+
+
+def zocalo_cluster_request(
+    file: Path,
+    root_folder: str,
+    session_id: int,  # Provided by the client via the API endpoint
+    instrument_name: str,  # Acquired by looking up the Session table
+    messenger: Optional[TransportManager] = None,
+):
+    if messenger:
+        # Use file path parts to construct parameters
+        path_parts = list((file.parent / file.stem).parts)
+        # Replace leading "/" in Unix paths
+        path_parts[0] = "" if path_parts[0] == "/" else path_parts[0]
+        try:
+            # Find the position of the root folder in the list
+            root_index = [p.lower() for p in path_parts].index(root_folder.lower())
+        except ValueError:
+            raise Exception(
+                f"Unable to find the root folder {root_folder!r} in the file path {file!r}"
+            )
+
+        # Construct the session and job name
+        session_dir = "/".join(path_parts[:root_index])
+        job_name = "--".join(
+            [p.replace(" ", "_") if " " in p else p for p in path_parts][
+                root_index + 1 :
+            ]
+        )
+
+        # Load machine config to get the feedback queue
+        machine_config = get_machine_config()
+        feedback_queue = machine_config[instrument_name].feedback_queue
+
+        # Send the message
+        #   The keys under "parameters" will populate all the matching fields in {}
+        #   in the processing recipe
+        messenger.send(
+            "processing_recipe",
+            {
+                "recipes": ["clem-lif-to-stack"],
+                "parameters": {
+                    # Job parameters
+                    "lif_file": f"{str(file)}",
+                    "root_folder": root_folder,
+                    # Other recipe parameters
+                    "session_dir": f"{str(session_dir)}",
+                    "session_id": session_id,
+                    "job_name": job_name,
+                    "feedback_queue": feedback_queue,
+                },
+            },
+            new_connection=True,
+        )
+    else:
+        raise Exception("Unable to find transport manager")