Standardise CLEM workflow

src/murfey/server/api/workflow_clem.py

@@ -16,9 +16,10 @@
 import murfey.util.db as MurfeyDB
 from murfey.server import _transport_object
 from murfey.server.murfey_db import murfey_db
+from murfey.util import sanitise_path
 
 # Set up logger
-logger = getLogger("murfey.server.api.clem")
+logger = getLogger("murfey.server.api.workflow_clem")
 
 # Create APIRouter class object
 router = APIRouter(
@@ -27,38 +28,79 @@
 )
 
 
-class LifInfo(BaseModel):
+class LifFileInfo(BaseModel):
     lif_file: Path
 
 
 @router.post("/sessions/{session_id}/process_raw_lifs")  # API posts to this URL
 def process_raw_lifs(
     session_id: int,
-    lif_file: LifInfo,
-    db: Session = murfey_db,
+    lif_info: LifFileInfo,
+    murfey_db: Session = murfey_db,
 ):
+    if _transport_object is None:
+        logger.error("No TransportManager object was set up")
+        return False
+
+    # Load the visit name from the database
     try:
-        # Try and load relevant Murfey workflow
-        workflow: EntryPoint = list(
-            entry_points(group="murfey.workflows", name="clem.process_raw_lifs")
-        )[0]
-    except IndexError:
-        raise RuntimeError("The relevant Murfey workflow was not found")
+        murfey_session = murfey_db.exec(
+            select(MurfeyDB.Session).where(MurfeyDB.Session.id == session_id)
+        ).one()
+        visit_name = murfey_session.visit
+    except Exception as e:
+        logger.error("Error querying session information from database", exc_info=True)
+        print(e)
+        return False
 
-    # Get instrument name from the database to load the correct config file
-    session_row: MurfeyDB.Session = db.exec(
-        select(MurfeyDB.Session).where(MurfeyDB.Session.id == session_id)
-    ).one()
-    instrument_name = session_row.instrument_name
+    # Find the visit directory, the raw directory name, and the job name
+    try:
+        visit_idx = lif_info.lif_file.parts.index(visit_name)
+        visit_dir = Path(
+            "/".join(
+                ""
+                if part == "/"  # Replace root "/" with "" for Linux paths
+                else part
+                for part in lif_info.lif_file.parts[: visit_idx + 1]
+            )
+        )
+        raw_dir = lif_info.lif_file.parts[visit_idx + 1]
+        job_name = str(
+            (lif_info.lif_file.parent / lif_info.lif_file.stem).relative_to(
+                visit_dir.parent
+            )
+        )
+    except Exception:
+        logger.error(
+            "Could not determine the visit directory from LIF file "
+            f"{sanitise_path(lif_info.lif_file)}",
+            exc_info=True,
+        )
+        return False
 
-    # Pass arguments along to the correct workflow
-    workflow.load()(
-        # Match the arguments found in murfey.workflows.clem.process_raw_lifs
-        file=lif_file.lif_file,
-        root_folder="images",
-        session_id=session_id,
-        instrument_name=instrument_name,
-        messenger=_transport_object,
+    # Construct recipe and submit it for processing
+    recipe = {
+        "recipes": ["clem-process-raw-lifs"],
+        "parameters": {
+            # Job parameters
+            "lif_file": f"{str(lif_info.lif_file)}",
+            "root_folder": raw_dir,
+            # Other recipe parameters
+            "session_dir": f"{str(visit_dir)}",
+            "session_id": session_id,
+            "job_name": job_name,
+            "feedback_queue": _transport_object.feedback_queue,
+        },
+    }
+    logger.debug(
+        f"Submitting LIF processing request to {_transport_object.feedback_queue!r} "
+        "with the following recipe: \n"
+        f"{recipe}"
+    )
+    _transport_object.send(
+        queue="processing_recipe",
+        message=recipe,
+        new_connection=True,
     )
     return True
 
@@ -73,31 +115,77 @@ class TIFFSeriesInfo(BaseModel):
 def process_raw_tiffs(
     session_id: int,
     tiff_info: TIFFSeriesInfo,
-    db: Session = murfey_db,
+    murfey_db: Session = murfey_db,
 ):
+    if _transport_object is None:
+        logger.error("No TransportManager object was set up")
+        return False
+    if not tiff_info.tiff_files:
+        logger.error("No TIFF files were included in the incoming message")
+        return False
+
+    # Load the visit name from the database
     try:
-        # Try and load relevant Murfey workflow
-        workflow: EntryPoint = list(
-            entry_points(group="murfey.workflows", name="clem.process_raw_tiffs")
-        )[0]
-    except IndexError:
-        raise RuntimeError("The relevant Murfey workflow was not found")
+        murfey_session = murfey_db.exec(
+            select(MurfeyDB.Session).where(MurfeyDB.Session.id == session_id)
+        ).one()
+        visit_name = murfey_session.visit
+    except Exception as e:
+        logger.error("Error querying session information from database", exc_info=True)
+        print(e)
+        return False
 
-    # Get instrument name from the database to load the correct config file
-    session_row: MurfeyDB.Session = db.exec(
-        select(MurfeyDB.Session).where(MurfeyDB.Session.id == session_id)
-    ).one()
-    instrument_name = session_row.instrument_name
+    # Find the visit directory, the raw directory name, and the job name
+    try:
+        tiff_file = tiff_info.tiff_files[0]
+        visit_idx = tiff_file.parts.index(visit_name)
+        visit_dir = Path(
+            "/".join(
+                ""
+                if part == "/"  # Replace root "/" with "" for Linux paths
+                else part
+                for part in tiff_file.parts[: visit_idx + 1]
+            )
+        )
+        raw_dir = tiff_file.parts[visit_idx + 1]
+        job_name = str(
+            (tiff_file.parent / tiff_file.stem.split("--")[0]).relative_to(
+                visit_dir.parent
+            )
+        )
+    except Exception:
+        logger.error(
+            "Could not determine the visit directory from TIFF file "
+            f"{sanitise_path(tiff_file)}",
+            exc_info=True,
+        )
+        return False
 
-    # 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,
+    # Construct recipe and submit it for processing
+    recipe = {
+        "recipes": ["clem-process-raw-tiffs"],
+        "parameters": {
+            # Job parameters
+            "tiff_list": "null",
+            "tiff_file": f"{str(tiff_file)}",
+            "root_folder": raw_dir,
+            "metadata": f"{str(tiff_info.series_metadata)}",
+            # Other recipe parameters
+            "session_dir": f"{str(visit_dir)}",
+            "session_id": session_id,
+            "job_name": job_name,
+            "feedback_queue": _transport_object.feedback_queue,
+        },
+    }
+    logger.debug(
+        f"Submitting TIFF processing request to {_transport_object.feedback_queue!r} "
+        "with the following recipe: \n"
+        f"{recipe}"
+    )
+    _transport_object.send(
+        queue="processing_recipe",
+        message=recipe,
+        new_connection=True,
     )
     return True
 

src/murfey/workflows/clem/register_preprocessing_results.py

@@ -25,7 +25,6 @@
 from murfey.util.processing_params import (
     default_clem_processing_parameters as processing_params,
 )
-from murfey.workflows.clem.align_and_merge import run as run_align_and_merge
 
 logger = logging.getLogger("murfey.workflows.clem.register_preprocessing_results")
 
@@ -556,12 +555,14 @@ def _register_grid_square(
 
 
 def run(message: dict, murfey_db: Session) -> dict[str, bool]:
-    session_id: int = (
-        int(message["session_id"])
-        if not isinstance(message["session_id"], int)
-        else message["session_id"]
-    )
+    # Early exit if no TransportManager object is configured
+    if not _transport_object:
+        logger.error("No TransportManager object was set up")
+        return {"success": False, "requeue": False}
+
+    # Parse the incoming message
     try:
+        session_id = int(message["session_id"])
         if isinstance(message["result"], str):
             json_obj: dict = json.loads(message["result"])
             result = CLEMPreprocessingResult(**json_obj)
@@ -577,6 +578,10 @@ def run(message: dict, murfey_db: Session) -> dict[str, bool]:
             "Exception encountered when parsing TIFF preprocessing result: \n"
             f"{traceback.format_exc()}"
         )
+
+    # Check that output files were included
+    if not result.output_files:
+        logger.error("No files were provided in the incoming message")
         return {"success": False, "requeue": False}
 
     # Outer try-finally block for tidying up database-related section of function
@@ -586,6 +591,8 @@ def run(message: dict, murfey_db: Session) -> dict[str, bool]:
             murfey_session = murfey_db.exec(
                 select(MurfeyDB.Session).where(MurfeyDB.Session.id == session_id)
             ).one()
+            instrument_name = murfey_session.instrument_name
+            visit_name = murfey_session.visit
         except Exception:
             logger.error(
                 "Exception encountered when loading Murfey session information: \n",
@@ -610,8 +617,8 @@ def run(message: dict, murfey_db: Session) -> dict[str, bool]:
             # Register data collection group and atlas in ISPyB
             _register_dcg_and_atlas(
                 session_id=session_id,
-                instrument_name=murfey_session.instrument_name,
-                visit_name=murfey_session.visit,
+                instrument_name=instrument_name,
+                visit_name=visit_name,
                 imaging_site=clem_img_site,
                 murfey_db=murfey_db,
             )
@@ -660,15 +667,39 @@ def run(message: dict, murfey_db: Session) -> dict[str, bool]:
             )
 
         # Request for image alignment and processing for the requested combinations
+        try:
+            ref_file = list(result.output_files.values())[0]
+            visit_idx = ref_file.parts.index(visit_name)
+            visit_dir = Path(
+                "/".join(
+                    ""
+                    if part == "/"  # Replace root "/" with "" for Linux paths
+                    else part
+                    for part in ref_file.parts[: visit_idx + 1]
+                )
+            )
+        except Exception:
+            logger.error("Could not construct visit directory", exc_info=True)
+            return {"success": False, "requeue": False}
         for image_combo in image_combos_to_process:
             try:
-                run_align_and_merge(
-                    session_id=session_id,
-                    instrument_name=murfey_session.instrument_name,
-                    series_name=result.series_name,
-                    images=image_combo,
-                    metadata=result.metadata,
-                    messenger=_transport_object,
+                _transport_object.send(
+                    "processing_recipe",
+                    {
+                        "recipes": ["clem-align-and-merge"],
+                        "parameters": {
+                            # Job parameters
+                            "series_name": result.series_name,
+                            "images": [str(file) for file in image_combo],
+                            "metadata": str(result.metadata),
+                            # Other recipe parameters
+                            "session_dir": str(visit_dir),
+                            "session_id": session_id,
+                            "job_name": result.series_name,
+                            "feedback_queue": _transport_object.feedback_queue,
+                        },
+                    },
+                    new_connection=True,
                 )
             except Exception:
                 logger.error(