More CLEM fixes (#306)

Pull request to resolve some of the issues mentioned in issue #304.
* More robust sorting of TIFF files once the stack exceeds 100 images
* Better logic to distinguish between image series that start with same position number
* Skipped processing of binned "Position X_pmd_X--" TIFF files
* Skipped processing of "Postion X_histo" XLIF files
* Optimised rsync parser logic to read rsync output correctly
* Corrected incorrect variable in tiff_to_stack function being passed to zocalo cluster
---------
Co-authored-by: Daniel Hatton <[email protected]>

Author: tieneupin

src/murfey/cli/tiff_to_stack.py

@@ -39,16 +39,10 @@ def run():
         f.resolve()
         for f in tiff_file.parent.glob("./*")
         if f.suffix in {".tif", ".tiff"}
-        and f.stem.startswith(tiff_file.stem.split("--")[0])
+        # Handle cases where series start with the same position number,
+        # but deviate afterwards
+        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:

src/murfey/client/analyser.py

@@ -141,7 +141,7 @@ def _find_context(self, file_path: Path) -> bool:
                 self._context = CLEMContext("leica", self._basepath)
                 return True
 
-        # Checks for tomography and SPA workflows
+        # Tomography and SPA workflow checks
         split_file_name = file_path.name.split("_")
         if split_file_name:
             # Files starting with "FoilHole" belong to the SPA workflow
@@ -237,15 +237,18 @@ def post_transfer(self, transferred_file: Path):
                     transferred_file, role=self._role, environment=self._environment
                 )
         except Exception as e:
-            logger.error(
-                f"An exception was encountered posting the file for transfer: {e}"
-            )
+            logger.error(f"An exception was encountered post transfer: {e}")
 
     def _analyse(self):
         logger.info("Analyser thread started")
         mdoc_for_reading = None
         while not self._halt_thread:
             transferred_file = self.queue.get()
+            transferred_file = (
+                Path(transferred_file)
+                if isinstance(transferred_file, str)
+                else transferred_file
+            )
             if not transferred_file:
                 self._halt_thread = True
                 continue
@@ -341,9 +344,15 @@ def _analyse(self):
                                         ),
                                     }
                                 )
+
                 # If a file with a CLEM context is identified, immediately post it
                 elif isinstance(self._context, CLEMContext):
+                    logger.info(
+                        f"File {transferred_file.name!r} will be processed as part of CLEM workflow"
+                    )
                     self.post_transfer(transferred_file)
+
+                # Handle files with tomography and SPA context differently
                 elif not self._extension or self._unseen_xml:
                     self._find_extension(transferred_file)
                     if self._extension:

src/murfey/client/contexts/clem.py

@@ -3,7 +3,6 @@
 the CLEM workflow should be processed.
 """
 
-# import requests
 import logging
 from datetime import datetime
 from pathlib import Path
@@ -13,7 +12,7 @@
 
 from murfey.client.context import Context
 from murfey.client.instance_environment import MurfeyInstanceEnvironment
-from murfey.util import capture_post, get_machine_config, sanitise
+from murfey.util import capture_post, get_machine_config
 from murfey.util.clem.xml import get_image_elements
 
 # Create logger object
@@ -74,24 +73,27 @@ def post_transfer(
         environment: Optional[MurfeyInstanceEnvironment] = None,
         **kwargs,
     ) -> bool:
+
         super().post_transfer(
             transferred_file, role=role, environment=environment, **kwargs
         )
 
         # Process files generated by "auto-save" acquisition mode
         # These include TIF/TIFF and XLIF files
         if transferred_file.suffix in (".tif", ".tiff", ".xlif"):
+            logger.debug(f"File extension {transferred_file.suffix!r} detected")
+
             # Type checking to satisfy MyPy
             if not environment:
                 logger.warning("No environment passed in")
-                return True
+                return False
 
             # 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
+                return False
 
             # Get the Path on the DLS file system
             file_path = _file_transferred_to(
@@ -101,23 +103,54 @@ def post_transfer(
             )
             if not file_path:
                 logger.warning(
-                    f"File associated with {sanitise(str(transferred_file))} not found on the storage system"
+                    f"File {transferred_file.name!r} not found on the storage system"
                 )
                 return False
 
+            # Skip processing of binned "_pmd" image series
+            if "_pmd_" in transferred_file.stem:
+                logger.debug(
+                    f"File {transferred_file.name!r} belongs to the '_pmd_' series of binned images; skipping processing"
+                )
+                return True
+
             # Process TIF/TIFF files
             if transferred_file.suffix in (".tif", ".tiff"):
+                logger.debug("Detected a TIFF file")
+
                 # Files should be named "PositionX--ZXX--CXX.tif" by default
-                if not len(transferred_file.stem.split("--")) == 3:
+                # If Position is repeated, it will add an additional --00X to the end
+                if len(transferred_file.stem.split("--")) not in [3, 4]:
                     logger.warning(
-                        "This TIFF file is likely not part of the CLEM workflow"
+                        f"File {transferred_file.name!r} is likely not part of the CLEM workflow"
                     )
                     return False
 
-                # 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 a unique name for the series
+                # For standard file name
+                if len(transferred_file.stem.split("--")) == 3:
+                    series_name = "/".join(
+                        [
+                            *file_path.parent.parts[-2:],  # Upper 2 parent directories
+                            file_path.stem.split("--")[0],
+                        ]
+                    )
+                # When this a repeated position
+                elif len(transferred_file.stem.split("--")) == 4:
+                    series_name = "/".join(
+                        [
+                            *file_path.parent.parts[-2:],  # Upper 2 parent directories
+                            "--".join(file_path.stem.split("--")[i] for i in [0, -1]),
+                        ]
+                    )
+                else:
+                    logger.error(
+                        f"Series name could not be generated from file {transferred_file.name!r}"
+                    )
+                    return False
+                logger.debug(
+                    f"File {transferred_file.name!r} given the series identifier {series_name!r}"
+                )
 
                 # Create key-value pairs containing empty list if not already present
                 if series_name not in self._tiff_series.keys():
@@ -132,15 +165,28 @@ def post_transfer(
                 self._tiff_timestamps[series_name].append(
                     transferred_file.stat().st_ctime
                 )
+                logger.debug(
+                    f"Created TIFF file dictionary entries for {series_name!r}"
+                )
 
             # Process XLIF files
             if transferred_file.suffix == ".xlif":
+                logger.debug("Detected an XLIF file")
+
+                # Skip processing of "_histo" histogram XLIF files
+                if transferred_file.stem.endswith("_histo"):
+                    logger.debug(
+                        f"File {transferred_file.name!r} contains histogram metadata; skipping processing"
+                    )
 
                 # 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
+                logger.debug(
+                    f"File {transferred_file.name!r} given the series identifier {series_name!r}"
+                )
 
                 # Extract metadata to get the expected size of the series
                 metadata = parse(transferred_file).getroot()
@@ -162,18 +208,30 @@ def post_transfer(
                     else 1
                 )
                 num_files = num_channels * num_frames
+                logger.debug(
+                    f"Expected number of files in {series_name!r}: {num_files}"
+                )
 
                 # 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
+                logger.debug(f"Created dictionary entries for {series_name!r} metadata")
 
             # Post message if all files for the associated series have been collected
-            if len(self._tiff_series[series_name]) == self._files_in_series.get(
-                series_name, 0  # Return 0 if the key hasn't been generated yet
+            # .get(series_name, 0) returns 0 if no associated key is found
+            if len(self._tiff_series[series_name]) == 0:
+                logger.debug(f"TIFF series {series_name!r} not yet loaded")
+                return True
+            elif self._files_in_series.get(series_name, 0) == 0:
+                logger.debug(
+                    f"Metadata for TIFF series {series_name!r} not yet processed"
+                )
+                return True
+            elif len(self._tiff_series[series_name]) == self._files_in_series.get(
+                series_name, 0
             ):
-
                 # Construct URL for Murfey server to communicate with
                 url = f"{str(environment.url.geturl())}/sessions/{environment.murfey_session}/tiff_to_stack"
                 if not url:
@@ -194,6 +252,9 @@ def post_transfer(
                         "description": "",
                     },
                 )
+                return True
+            else:
+                logger.debug(f"TIFF series {series_name!r} is still being processed")
 
         # Process LIF files
         if transferred_file.suffix == ".lif":
@@ -233,4 +294,5 @@ def post_transfer(
                     "description": "",
                 },
             )
+            return True
         return True

src/murfey/client/rsync.py

@@ -1,3 +1,10 @@
+"""
+rsync is a Linux process that does fast, versatile remote (and local) file copying.
+This module allows Murfey to use rsync to facilitate its file transfers, and to
+parse the logs outputted by rsync in order confirm that the file has been correctly
+transferred.
+"""
+
 from __future__ import annotations
 
 import logging
@@ -251,22 +258,39 @@ def _fake_transfer(self, files: list[Path]) -> bool:
 
     def _transfer(self, files: list[Path]) -> bool:
         """
-        Actually transfer files in an rsync subprocess
+        Transfer files via an rsync sub-process, and parses the rsync stdout to verify
+        the success of the transfer.
         """
-        previously_transferred = self._files_transferred
 
-        next_file: RSyncerUpdate | None = None
+        # Set up initial variables
+        files = [f for f in files if f.is_file()]
+        previously_transferred = self._files_transferred
         transfer_success: set[Path] = set()
         successful_updates: list[RSyncerUpdate] = []
-
-        files = [f for f in files if f.is_file()]
+        next_file: RSyncerUpdate | None = None
 
         def parse_stdout(line: str):
+            """
+            Reads the stdout from rsync in order to verify the status of transferred
+            files and other things.
+            """
             nonlocal next_file
 
+            # Miscellaneous rsync stdout lines to skip
             if not line:
                 return
+            if line.startswith(("building file list", "created directory", "sending")):
+                return
+            if line.startswith("sent "):
+                # sent 6,676 bytes  received 397 bytes  4,715.33 bytes/sec
+                return
+            if line.startswith("total "):
+                # total size is 315,265,653  speedup is 44,573.12 (DRY RUN)
+                return
+            if line.startswith(("cd", ".d")):
+                return
 
+            # Lines with chr(13), \r, contain file transfer information
             if chr(13) in line:
                 # partial transfer
                 #
@@ -288,15 +312,8 @@ def parse_stdout(line: str):
                 successful_updates.append(next_file._replace(file_size=size_bytes))
                 next_file = None
                 return
-            if line.startswith(("building file list", "created directory", "sending")):
-                return
-            if line.startswith("sent "):
-                # sent 6,676 bytes  received 397 bytes  4,715.33 bytes/sec
-                return
-            if line.startswith("total "):
-                # total size is 315,265,653  speedup is 44,573.12 (DRY RUN)
-                return
 
+            # Lines starting with "*f" contain file transfer information
             if line.startswith((".f", ">f", "<f")):
                 # .d          ./
                 # .f          README.md
@@ -317,7 +334,9 @@ def parse_stdout(line: str):
                     self._files_transferred - previously_transferred
                 )
                 update = RSyncerUpdate(
-                    file_path=Path(line[12:].replace(" ", "")),
+                    file_path=Path(
+                        line[12:].rstrip()
+                    ),  # Remove trailing newlines, spaces, and carriage returns
                     file_size=0,
                     outcome=TransferResult.SUCCESS,
                     transfer_total=self._files_transferred - previously_transferred,
@@ -334,18 +353,17 @@ def parse_stdout(line: str):
                     next_file = update
                 return
 
-            if line.startswith(("cd", ".d")):
-                return
-
         def parse_stderr(line: str):
-            logger.warning(line)
+            logger.warning(f"rsync stderr: {line!r}")
 
-        relative_filenames = []
+        # Generate list of relative filenames for this batch of transferred files
+        relative_filenames: List[Path] = []
         for f in files:
             try:
                 relative_filenames.append(f.relative_to(self._basepath))
             except ValueError:
                 raise ValueError(f"File '{f}' is outside of {self._basepath}") from None
+
         if self._remove_files:
             if self._required_substrings_for_removal:
                 rsync_stdin_remove = b"\n".join(
@@ -372,6 +390,9 @@ def parse_stderr(line: str):
         else:
             rsync_stdin_remove = b""
             rsync_stdin = b"\n".join(os.fsencode(f) for f in relative_filenames)
+
+        # Create and run rsync subprocesses
+        # rsync commands to pass to subprocess
         rsync_cmd = [
             "rsync",
             "-iiv",
@@ -381,7 +402,6 @@ def parse_stderr(line: str):
             "--files-from=-",
             "-p",  # preserve permissions
         ]
-
         rsync_cmd.extend([".", self._remote])
 
         result: subprocess.CompletedProcess | None = None
@@ -415,7 +435,9 @@ def parse_stderr(line: str):
 
         self.notify(successful_updates, secondary=True)
 
+        # Compare files from rsync stdout to original list to verify transfer
         for f in set(relative_filenames) - transfer_success:
+            logger.warning(f"Transfer of file {f.name!r} considered a failure")
             self._files_transferred += 1
             current_outstanding = self.queue.unfinished_tasks - (
                 self._files_transferred - previously_transferred