Process TIFF files for cryo-CLEM workflow (#295)

Added function, API endpoint, command line command, and cluster submission script for creating image stacks from individual TIFF files.

Author: tieneupin

pyproject.toml

@@ -30,6 +30,7 @@ classifiers = [
     "Programming Language :: Python :: 3.12",
 ]
 dependencies = [
+    "defusedxml", # For safely parsing XML files
     "pydantic<2", # Pip hops between installing v2.7 or v1.10 depending on which of the additional dependencies are requested
     "requests",
     "rich",
@@ -85,9 +86,11 @@ murfey = "murfey.client:run"
 "murfey.simulate" = "murfey.cli.dummy:run"
 "murfey.spa_inject" = "murfey.cli.inject_spa_processing:run"
 "murfey.spa_ispyb_entries" = "murfey.cli.spa_ispyb_messages:run"
+"murfey.tiff_to_stack" = "murfey.cli.tiff_to_stack:run"
 "murfey.transfer" = "murfey.cli.transfer:run"
 [project.entry-points."murfey.workflows"]
 "lif_to_tiff" = "murfey.workflows.lif_to_tiff:zocalo_cluster_request"
+"tiff_to_stack" = "murfey.workflows.tiff_to_stack:zocalo_cluster_request"
 
 [tool.setuptools]
 package-dir = {"" = "src"}

src/murfey/cli/lif_to_tiff.py

@@ -1,20 +1,24 @@
 import argparse
 from pathlib import Path
 
-from murfey.util.lif import convert_lif_to_tiff
+from murfey.util.clem import convert_lif_to_tiff
 
 
 def run():
-    parser = argparse.ArgumentParser(description="Convert LIF to TIFF")
+    parser = argparse.ArgumentParser(
+        description="Convert LIF files into TIFF image stacks"
+    )
 
     parser.add_argument(
-        nargs=1, dest="lif_path", help="Path to LIF file for conversion"
+        dest="lif_path",
+        type=str,
+        help="Path to LIF file for conversion",
     )
     parser.add_argument(
         "--root-dir",
         default="images",
         type=str,
-        help="Top subdirectory that LIF files are stored in. Used to determine destination of TIFFs",
+        help="Top subdirectory that LIF files are stored in. Used to determine destination of the created TIFF image stacks",
     )
     parser.add_argument(
         "-n", "--num-procs", default=1, type=int, help="Number of processes"
@@ -23,7 +27,7 @@ def run():
     args = parser.parse_args()
 
     convert_lif_to_tiff(
-        Path(args.lif_path),
+        file=Path(args.lif_path),
         root_folder=args.root_dir,
         number_of_processes=args.num_procs,
     )

src/murfey/cli/tiff_to_stack.py

@@ -0,0 +1,63 @@
+import argparse
+from pathlib import Path
+
+from murfey.util.clem import convert_tiff_to_stack
+
+
+def run():
+    # Create an argument parser
+    parser = argparse.ArgumentParser(
+        description="Convert individual TIFF files into image stacks"
+    )
+    # Path to single TIFF file from series (Mandatory)
+    parser.add_argument(
+        dest="tiff_path",
+        type=str,
+        help="Path to any one of the TIFF files from the series to be processed",
+    )
+    # Root directory (Optional)
+    parser.add_argument(
+        "--root-dir",
+        default="images",
+        type=str,
+        help="Top subdirectory that raw TIFF files are stored in. Used to determine destination of the created image stacks",
+    )
+    # Path to metadata file (Optional)
+    parser.add_argument(
+        "--metadata",
+        default=None,
+        type=str,
+        help="Path to the XLIF file associated with this dataset. If not provided, the script will use relative file paths to find what it thinks is the appropriate file",
+    )
+    # Parse the arguments
+    args = parser.parse_args()
+
+    # Convert to correct object types
+    tiff_file = Path(args.tiff_path)
+    # Generate list from the single file provided
+    tiff_list = [
+        f.resolve()
+        for f in tiff_file.parent.glob("./*")
+        if f.suffix in {".tif", ".tiff"}
+        and f.stem.startswith(tiff_file.stem.split("--")[0])
+    ]
+    # Sort by series, then channel, then frame
+    tiff_list.sort(
+        key=lambda e: (
+            e.stem.split("--")[0],
+            e.stem.split("--")[2],
+            e.stem.split("--")[1],
+        )
+    )
+
+    # Resolve for metadata argument
+    if not args.metadata:
+        metadata = None
+    else:
+        metadata = Path(args.metadata)
+
+    convert_tiff_to_stack(
+        tiff_list=tiff_list,
+        root_folder=args.root_dir,
+        metadata_file=metadata,
+    )

src/murfey/client/contexts/clem.py

@@ -10,11 +10,14 @@
 import logging
 from datetime import datetime
 from pathlib import Path
-from typing import Optional
+from typing import Dict, List, Optional
+
+from defusedxml.ElementTree import parse
 
 from murfey.client.context import Context
 from murfey.client.instance_environment import MurfeyInstanceEnvironment
-from murfey.util import capture_post, get_machine_config
+from murfey.util import capture_post, get_machine_config, sanitise
+from murfey.util.clem import xml
 
 # Create logger object
 logger = logging.getLogger("murfey.client.contexts.clem")
@@ -50,10 +53,20 @@ def _get_source(
     return None
 
 
+# WORK IN PROGRESS
+# Will need to add context for TIFF files associated with CLEM
 class CLEMContext(Context):
     def __init__(self, acquisition_software: str, basepath: Path):
         super().__init__("CLEM", acquisition_software)
         self._basepath = basepath
+        # CLEM contexts for "auto-save" acquisition mode
+        self._tiff_series: Dict[str, List[str]] = {}  # Series name : List of TIFF paths
+        self._tiff_timestamps: Dict[str, List[float]] = {}  # Series name: Timestamps
+        self._tiff_sizes: Dict[str, List[int]] = {}  # Series name: File sizes
+        self._series_metadata: Dict[str, str] = {}  # Series name: Metadata file path
+        self._metadata_timestamp: Dict[str, float] = {}  # Series name: Timestamp
+        self._metadata_size: Dict[str, int] = {}  # Series name: File size
+        self._files_in_series: Dict[str, int] = {}  # Series name : Total TIFFs
 
     def post_transfer(
         self,
@@ -65,7 +78,126 @@ def post_transfer(
         super().post_transfer(
             transferred_file, role=role, environment=environment, **kwargs
         )
-        # Check if file is a LIF file
+
+        # Process files generated by "auto-save" acquisition mode
+        # These include TIF/TIFF and XLIF files
+        if transferred_file.suffix in (".tif", ".tiff", ".xlif"):
+            # Type checking to satisfy MyPy
+            if not environment:
+                logger.warning("No environment passed in")
+                return True
+
+            # Location of the file on the client PC
+            source = _get_source(transferred_file, environment)
+            # Type checking to satisfy MyPy
+            if not source:
+                logger.warning(f"No source found for file {transferred_file}")
+                return True
+
+            # Get the Path on the DLS file system
+            file_path = _file_transferred_to(
+                environment=environment,
+                source=source,
+                file_path=transferred_file,
+            )
+            if not file_path:
+                logger.warning(
+                    f"File associated with {sanitise(str(transferred_file))} not found on the storage system"
+                )
+                return False
+
+            # Process TIF/TIFF files
+            if any(transferred_file.suffix == s for s in [".tif", ".tiff"]):
+                # Files should be named "PositionX--ZXX--CXX.tif" by default
+                if not len(transferred_file.stem.split("--")) == 3:
+                    logger.warning(
+                        "This TIFF file is likely not part of the CLEM workflow"
+                    )
+                    return False  # Not sure if None, False, or True is most appropriate
+
+                # Get series name from file name
+                series_name = "/".join(
+                    [*file_path.parent.parts[-2:], file_path.stem.split("--")[0]]
+                )  # The previous 2 parent directories should be unique enough
+
+                # Create key-value pairs containing empty list if not already present
+                if series_name not in self._tiff_series.keys():
+                    self._tiff_series[series_name] = []
+                if series_name not in self._tiff_sizes.keys():
+                    self._tiff_sizes[series_name] = []
+                if series_name not in self._tiff_timestamps.keys():
+                    self._tiff_timestamps[series_name] = []
+                # Append information to list
+                self._tiff_series[series_name].append(str(file_path))
+                self._tiff_sizes[series_name].append(transferred_file.stat().st_size)
+                self._tiff_timestamps[series_name].append(
+                    transferred_file.stat().st_ctime
+                )
+
+            # Process XLIF files
+            if transferred_file.suffix == ".xlif":
+
+                # XLIF files don't have the "--ZXX--CXX" additions in the file name
+                # But they have "/Metadata/" as the immediate parent
+                series_name = "/".join(
+                    [*file_path.parent.parent.parts[-2:], file_path.stem]
+                )  # The previous 2 parent directories should be unique enough
+
+                # Extract metadata to get the expected size of the series
+                metadata = parse(file_path).getroot()
+                metadata = xml.get_image_elements(metadata)[0]
+
+                # Get channel and dimension information
+                channels = metadata.findall(
+                    "Data/Image/ImageDescription/Channels/ChannelDescription"
+                )
+                dimensions = metadata.findall(
+                    "Data/Image/ImageDescription/Dimensions/DimensionDescription"
+                )
+
+                # Calculate expected number of files for this series
+                num_channels = len(channels)
+                num_frames = (
+                    int(dimensions[2].attrib["NumberOfElements"])
+                    if len(dimensions) > 2
+                    else 1
+                )
+                num_files = num_channels * num_frames
+
+                # Update dictionary entries
+                self._files_in_series[series_name] = num_files
+                self._series_metadata[series_name] = str(file_path)
+                self._metadata_size[series_name] = transferred_file.stat().st_size
+                self._metadata_timestamp[series_name] = transferred_file.stat().st_ctime
+
+            # Post message if all files for the associated series have been collected
+            if (
+                len(self._tiff_series[series_name])
+                == self._files_in_series[series_name]
+            ):
+
+                # Construct URL for Murfey server to communicate with
+                url = f"{str(environment.url.geturl())}/sessions/{environment.murfey_session}/tiff_to_stack"
+                if not url:
+                    logger.warning("No URL found for the environment")
+                    return True
+
+                # Post the message and log any errors that arise
+                capture_post(
+                    url,
+                    json={
+                        "series_name": series_name,
+                        "tiff_files": self._tiff_series[series_name],
+                        "tiff_sizes": self._tiff_sizes[series_name],
+                        "tiff_timestamps": self._tiff_timestamps[series_name],
+                        "series_metadata": self._series_metadata[series_name],
+                        "metadata_size": self._metadata_size[series_name],
+                        "metadata_timestamp": self._metadata_timestamp[series_name],
+                        "description": "",
+                    },
+                )
+
+        # Process LIF files
         if transferred_file.suffix == ".lif":
             # Type checking to satisfy MyPy
             if not environment:
@@ -83,7 +215,7 @@ def post_transfer(
             url = f"{str(environment.url.geturl())}/sessions/{environment.murfey_session}/lif_to_tiff"
             # Type checking to satisfy MyPy
             if not url:
-                logger.warning("No url found for the environment")
+                logger.warning("No URL found for the environment")
                 return True
 
             # Get the Path on the DLS file system

src/murfey/server/clem/api.py

@@ -5,15 +5,14 @@
 from fastapi import APIRouter
 
 from murfey.server import _transport_object
-from murfey.util.lif import convert_lif_to_tiff
-from murfey.util.models import LifFileInfo
+from murfey.util.clem import convert_lif_to_tiff, convert_tiff_to_stack
+from murfey.util.models import LifFileInfo, TiffSeriesInfo
 
 # Create APIRouter class object
 router = APIRouter()
 
 
-# Allow function to be seen as an endpoint by the router
-@router.post("/sessions/{session_id}/lif_to_tiff")
+@router.post("/sessions/{session_id}/lif_to_tiff")  # API posts to this URL
 def lif_to_tiff(
     session_id: int,  # Used by the decorator
     lif_info: LifFileInfo,
@@ -23,10 +22,38 @@ def lif_to_tiff(
     )
     if murfey_workflows:
         murfey_workflows[0].load()(
-            file=lif_info.name, root_folder="images", messenger=_transport_object
+            # Match the arguments found in murfey.workflows.lif_to_tiff
+            file=lif_info.name,
+            root_folder="images",
+            messenger=_transport_object,
         )
     else:
         convert_lif_to_tiff(
             file=lif_info.name,
             root_folder="images",
         )
+
+
+# WORK IN PROGRESS
+@router.post("/sessions/{session_id}/tiff_to_stack")
+def tiff_to_stack(
+    session_id: int,  # Used by the decorator
+    tiff_info: TiffSeriesInfo,
+):
+    murfey_workflows = importlib.metadata.entry_points().select(
+        group="murfey.workflows", name="tiff_to_stack"
+    )
+    if murfey_workflows:
+        murfey_workflows[0].load()(
+            # Match the arguments found in murfey.workflows.tiff_to_stack
+            file=tiff_info.tiff_files,
+            root_folder="images",
+            metadata=tiff_info.series_metadata,
+            messenger=_transport_object,
+        )
+    else:
+        convert_tiff_to_stack(
+            tiff_list=tiff_info.tiff_files,
+            root_folder="images",
+            metadata_file=tiff_info.series_metadata,
+        )