CLEM workflow fixes (#784)

* Fixed broken logic when registering align-and-merge processing results
* Do not pass align-and-merge processing parameters via Murfey; set them directly in the recipe
* Fixed broken tests and added test for the align-and-merge registration workflow

Author: tieneupin

src/murfey/util/processing_params.py

@@ -1,7 +1,6 @@
 from datetime import datetime
 from functools import lru_cache
 from pathlib import Path
-from typing import Literal, Optional
 
 from pydantic import BaseModel
 from werkzeug.utils import secure_filename
@@ -60,12 +59,6 @@ class CLEMProcessingParameters(BaseModel):
     # Atlas vs GridSquare registration threshold
     atlas_threshold: float = 0.0015  # in m
 
-    # Image alignment and merging-specific parameters
-    crop_to_n_frames: Optional[int] = 50
-    align_self: Literal["enabled", ""] = "enabled"
-    flatten: Literal["mean", "min", "max", ""] = "mean"
-    align_across: Literal["enabled", ""] = "enabled"
-
 
 default_clem_processing_parameters = CLEMProcessingParameters()
 

src/murfey/workflows/clem/align_and_merge.py

@@ -6,7 +6,7 @@
 from __future__ import annotations
 
 from pathlib import Path
-from typing import Literal, Optional
+from typing import Optional
 
 from murfey.util.config import get_machine_config
 
@@ -24,11 +24,6 @@ def run(
     series_name: str,
     images: list[Path],
     metadata: Path,
-    # Optional processing parameters
-    crop_to_n_frames: Optional[int] = None,
-    align_self: Literal["enabled", ""] = "",
-    flatten: Literal["mean", "min", "max", ""] = "mean",
-    align_across: Literal["enabled", ""] = "",
     # Optional session parameters
     messenger: Optional[TransportManager] = None,
 ):
@@ -65,10 +60,6 @@ def run(
                 "series_name": series_name,
                 "images": [str(file) for file in images],
                 "metadata": str(metadata),
-                "crop_to_n_frames": crop_to_n_frames,
-                "align_self": align_self,
-                "flatten": flatten,
-                "align_across": align_across,
                 # Other recipe parameters
                 "session_dir": str(session_dir),
                 "session_id": session_id,

src/murfey/workflows/clem/register_align_and_merge_results.py

@@ -4,10 +4,11 @@
 import logging
 import traceback
 from ast import literal_eval
+from functools import cached_property
 from pathlib import Path
 from typing import Optional
 
-from pydantic import BaseModel, field_validator
+from pydantic import BaseModel, computed_field, field_validator
 from sqlmodel import Session, select
 
 from murfey.util.db import ImagingSite
@@ -25,6 +26,27 @@ class AlignAndMergeResult(BaseModel):
     thumbnail: Optional[Path] = None
     thumbnail_size: Optional[tuple[int, int]] = None
 
+    # Valid Pydantic decorator not supported by MyPy
+    @computed_field  # type: ignore
+    @cached_property
+    def is_denoised(self) -> bool:
+        """
+        The "_Lng_LVCC" suffix appended to a CLEM dataset's position name indicates
+        that it's a denoised image set of the same position. These results should
+        override or supersede the original ones once they're available.
+        """
+        return "_Lng_LVCC" in self.series_name
+
+    # Valid Pydantic decorator not supported by MyPy
+    @computed_field  # type: ignore
+    @cached_property
+    def site_name(self) -> str:
+        """
+        Extract just the name of the site by removing the "_Lng_LVCC" suffix from
+        the series name.
+        """
+        return self.series_name.replace("_Lng_LVCC", "")
+
     @field_validator("image_stacks", mode="before")
     @classmethod
     def parse_stringified_list(cls, value):
@@ -78,26 +100,36 @@ def run(message: dict, murfey_db: Session) -> dict[str, bool]:
 
     # Outer try-finally block for tidying up database-related section of function
     try:
-        # Register items in database if not already present
         try:
-            if not (
-                clem_img_series := murfey_db.exec(
-                    select(ImagingSite)
-                    .where(ImagingSite.session_id == session_id)
-                    .where(ImagingSite.site_name == result.series_name)
-                ).one_or_none()
-            ):
-                clem_img_series = ImagingSite(
-                    session_id=session_id, series_name=result.series_name
+            clem_img_site = murfey_db.exec(
+                select(ImagingSite)
+                .where(ImagingSite.session_id == session_id)
+                .where(ImagingSite.site_name == result.site_name)
+            ).one()
+
+            # Update the stored entry only if the incoming one matches it
+            if clem_img_site.image_path is not None and (
+                # Denoised dataset results should be registered regardless
+                result.is_denoised
+                or (
+                    # Raw dataset result should only be considered
+                    # If the current entry is also a raw dataset
+                    not result.is_denoised
+                    and "_Lng_LVCC" not in clem_img_site.image_path
                 )
-            clem_img_series.composite_created = True
-            murfey_db.add(clem_img_series)
-            murfey_db.commit()
+            ):
+                clem_img_site.composite_created = True
+                murfey_db.add(clem_img_site)
+                murfey_db.commit()
 
-            logger.info(
-                "Align-and-merge processing result registered for "
-                f"{result.series_name!r} series"
-            )
+                logger.info(
+                    "Align-and-merge processing result registered for "
+                    f"{result.series_name!r} series"
+                )
+            else:
+                logger.info(
+                    "Skipping database registration as incoming result doesn't match stored entry"
+                )
 
         except Exception:
             logger.error(

src/murfey/workflows/clem/register_preprocessing_results.py

@@ -658,10 +658,6 @@ def run(message: dict, murfey_db: Session) -> dict[str, bool]:
                     series_name=result.series_name,
                     images=image_combo,
                     metadata=result.metadata,
-                    crop_to_n_frames=processing_params.crop_to_n_frames,
-                    align_self=processing_params.align_self,
-                    flatten=processing_params.flatten,
-                    align_across=processing_params.align_across,
                     messenger=_transport_object,
                 )
             except Exception:

tests/workflows/clem/test_align_and_merge.py

@@ -1,8 +1,8 @@
 from pathlib import Path
-from typing import Literal
-from unittest.mock import MagicMock, patch
+from unittest.mock import MagicMock
 
 import pytest
+from pytest_mock import MockerFixture
 
 from murfey.server.ispyb import TransportManager
 from murfey.util.config import MachineConfig
@@ -23,12 +23,6 @@
 ]
 feedback_queue = "murfey_feedback"
 
-# Align and merge settings
-crop_to_n_frames = 20
-align_self: Literal["enabled", ""] = "enabled"
-flatten: Literal["mean", "min", "max", ""] = "max"
-align_across: Literal["enabled", ""] = "enabled"
-
 
 @pytest.fixture
 def processed_dir(tmp_path: Path):
@@ -62,9 +56,8 @@ def metadata(processed_dir: Path):
     return metadata
 
 
-@patch("murfey.workflows.clem.align_and_merge.get_machine_config")
 def test_run(
-    mock_get_machine_config,
+    mocker: MockerFixture,
     image_stacks: list[Path],
     metadata: Path,
     processed_dir: Path,
@@ -81,6 +74,9 @@ def test_run(
     # Construct a mock MachineConfig object for use within the function
     mock_machine_config = MagicMock(spec=MachineConfig)
     mock_machine_config.processed_directory_name = processed_folder
+    mock_get_machine_config = mocker.patch(
+        "murfey.workflows.clem.align_and_merge.get_machine_config"
+    )
     mock_get_machine_config.return_value = {
         instrument_name: mock_machine_config,
     }
@@ -92,10 +88,6 @@ def test_run(
         series_name=series_name_long,
         images=image_stacks,
         metadata=metadata,
-        crop_to_n_frames=crop_to_n_frames,
-        align_self=align_self,
-        flatten=flatten,
-        align_across=align_across,
         messenger=mock_transport,
     )
 
@@ -107,10 +99,6 @@ def test_run(
             "series_name": series_name_long,
             "images": [str(file) for file in image_stacks],
             "metadata": str(metadata),
-            "crop_to_n_frames": crop_to_n_frames,
-            "align_self": align_self,
-            "flatten": flatten,
-            "align_across": align_across,
             # Other recipe parameters
             "session_dir": str(processed_dir.parent),
             "session_id": session_id,

tests/workflows/clem/test_register_align_and_merge_results.py

@@ -0,0 +1,108 @@
+from pathlib import Path
+
+import pytest
+from sqlmodel import Session as SQLModelSession, select
+
+import murfey.util.db as MurfeyDB
+from murfey.workflows.clem.register_align_and_merge_results import run
+from tests.conftest import ExampleVisit, get_or_create_db_entry
+
+session_id = ExampleVisit.murfey_session_id + 1
+visit_name = f"{ExampleVisit.proposal_code}{ExampleVisit.proposal_number}-{ExampleVisit.visit_number}"
+processed_dir_name = "processed"
+project_name = "Grid_1"
+
+
+@pytest.mark.parametrize(
+    "test_params",
+    [  # Registered data | Incoming data
+        ["_Lng_LVCC", ""],
+        ["_Lng_LVCC", "_Lng_LVCC"],
+        ["", ""],
+        ["", "_Lng_LVCC"],
+    ],
+)
+def test_run(
+    test_params: tuple[str, str], murfey_db_session: SQLModelSession, tmp_path: Path
+):
+    # Unpack test params
+    registered_type, incoming_type = test_params
+
+    # Create a session to insert for this test
+    murfey_session: MurfeyDB.Session = get_or_create_db_entry(
+        murfey_db_session,
+        MurfeyDB.Session,
+        lookup_kwargs={
+            "id": session_id,
+            "name": visit_name,
+            "visit": visit_name,
+            "instrument_name": ExampleVisit.instrument_name,
+        },
+    )
+
+    # Create an ImagingSite entry using the existing values
+    registered_position_name = "Position_1" + registered_type
+    image_path = (
+        tmp_path
+        / visit_name
+        / "processed"
+        / project_name
+        / "TileScan_1"
+        / registered_position_name
+        / "*.tiff"
+    )
+    registered_series_name = f"{project_name}--TileScan_1--{registered_position_name}"
+    site_name = registered_series_name.rstrip(registered_type)
+    get_or_create_db_entry(
+        murfey_db_session,
+        MurfeyDB.ImagingSite,
+        lookup_kwargs={
+            "session_id": murfey_session.id,
+            "site_name": site_name,
+            "image_path": str(image_path),
+        },
+    )
+
+    # Create the incoming message
+    incoming_position_name = "Position_1" + incoming_type
+    incoming_series_name = f"{project_name}--TileScan_1--{incoming_position_name}"
+    # The site names should match
+    assert site_name == incoming_series_name.rstrip(incoming_type)
+
+    message = {
+        "session_id": murfey_session.id,
+        "result": {
+            "series_name": incoming_series_name,
+            "image_stacks": [],
+            "align_self": None,
+            "flatten": None,
+            "align_across": None,
+            "output_file": tmp_path / "dummy",
+            "thumbnail": None,
+            "thumbnail_size": None,
+        },
+    }
+
+    # Run the function and check that the expected values were created
+    result = run(message, murfey_db_session)
+    assert result["success"]
+
+    imaging_site = murfey_db_session.exec(
+        select(MurfeyDB.ImagingSite)
+        .where(MurfeyDB.ImagingSite.session_id == murfey_session.id)
+        .where(
+            MurfeyDB.ImagingSite.site_name == incoming_series_name.rstrip(incoming_type)
+        )
+    ).one()
+
+    # Check that 'composite_created' is updated correctly
+    if (
+        # Both data types match
+        (registered_type and incoming_type)
+        or (not registered_type and not incoming_type)
+        # Registered type is raw data, while incoming is denoised
+        or (not registered_type and incoming_type)
+    ):
+        assert imaging_site.composite_created
+    else:
+        assert not imaging_site.composite_created