Register FIB data (#766)

* Added a new FastAPI router, API end point, workflow, and entry point to use for registering FIB atlas images
* Added logic to the FIB Context to trigger the registration of FIB atlas images
* Added new columns to the ImagingSite table to keep track of stage position information in the FIB
* Enabled and fixed a prior workflow that was set up to create GIFs of the milling sites as images are acquired
* Added tests and fixed old ones

Author: tieneupin

pyproject.toml

@@ -115,6 +115,7 @@ TomographyMetadataContext = "murfey.client.contexts.tomo_metadata:TomographyMeta
 "data_collection" = "murfey.workflows.register_data_collection:run"
 "data_collection_group" = "murfey.workflows.register_data_collection_group:run"
 "experiment_type_update" = "murfey.workflows.register_experiment_type_update:run"
+"fib.register_atlas" = "murfey.workflows.fib.register_atlas:run"
 "pato" = "murfey.workflows.notifications:notification_setup"
 "picked_particles" = "murfey.workflows.spa.picking:particles_picked"
 "picked_tomogram" = "murfey.workflows.tomo.picking:picked_tomogram"

src/murfey/client/analyser.py

@@ -400,7 +400,7 @@ def _analyse(self):
                         mdoc_for_reading = transferred_file
                 if not self._context:
                     if not self._find_extension(transferred_file):
-                        logger.error(f"No extension found for {transferred_file}")
+                        logger.debug(f"No extension found for {transferred_file}")
                         continue
                     if not self._find_context(transferred_file):
                         logger.debug(

src/murfey/client/contexts/fib.py

@@ -6,7 +6,6 @@
 from datetime import datetime
 from pathlib import Path
 from typing import NamedTuple
-from xml.etree import ElementTree as ET
 
 import xmltodict
 
@@ -30,21 +29,6 @@ class MillingProgress(NamedTuple):
     timestamp: float
 
 
-class ElectronSnapshotMetadata(NamedTuple):
-    slot_num: int | None  # Which slot in the FIB-SEM it is from
-    image_num: int
-    image_dir: str  # Partial path from EMproject.emxml parent to the image
-    status: str
-    x_len: float | None
-    y_len: float | None
-    z_len: float | None
-    x_center: float | None
-    y_center: float | None
-    z_center: float | None
-    extent: tuple[float, float, float, float] | None
-    rotation_angle: float | None
-
-
 def _number_from_name(name: str) -> int:
     """
     In the AutoTEM and Maps workflows for the FIB, the sites and images are
@@ -89,72 +73,6 @@ def _file_transferred_to(
     return destination
 
 
-def _parse_electron_snapshot_metadata(xml_file: Path):
-    metadata_dict = {}
-    root = ET.parse(xml_file).getroot()
-    datasets = root.findall(".//Datasets/Dataset")
-    for dataset in datasets:
-        # Extract all string-based values
-        name, image_dir, status = [
-            node.text
-            if ((node := dataset.find(node_path)) is not None and node.text is not None)
-            else ""
-            for node_path in (
-                ".//Name",
-                ".//FinalImages",
-                ".//Status",
-            )
-        ]
-
-        # Extract all float values
-        cx, cy, cz, x_len, y_len, z_len, rotation_angle = [
-            float(node.text)
-            if ((node := dataset.find(node_path)) is not None and node.text is not None)
-            else None
-            for node_path in (
-                ".//BoxCenter/CenterX",
-                ".//BoxCenter/CenterY",
-                ".//BoxCenter/CenterZ",
-                ".//BoxSize/SizeX",
-                ".//BoxSize/SizeY",
-                ".//BoxSize/SizeZ",
-                ".//RotationAngle",
-            )
-        ]
-
-        # Calculate the extent of the image
-        extent = None
-        if (
-            cx is not None
-            and cy is not None
-            and x_len is not None
-            and y_len is not None
-        ):
-            extent = (
-                x_len - (cx / 2),
-                x_len + (cx / 2),
-                y_len - (cy / 2),
-                y_len - (cy / 2),
-            )
-
-        # Append metadata for current site to dict
-        metadata_dict[name] = ElectronSnapshotMetadata(
-            slot_num=None if cx is None else (1 if cx < 0 else 2),
-            image_num=_number_from_name(name),
-            status=status,
-            image_dir=image_dir,
-            x_len=x_len,
-            y_len=y_len,
-            z_len=z_len,
-            x_center=cx,
-            y_center=cy,
-            z_center=cz,
-            extent=extent,
-            rotation_angle=rotation_angle,
-        )
-    return metadata_dict
-
-
 class FIBContext(Context):
     def __init__(
         self,
@@ -168,9 +86,6 @@ def __init__(
         self._machine_config = machine_config
         self._milling: dict[int, list[MillingProgress]] = {}
         self._lamellae: dict[int, Lamella] = {}
-        self._electron_snapshots: dict[str, Path] = {}
-        self._electron_snapshot_metadata: dict[str, ElectronSnapshotMetadata] = {}
-        self._electron_snapshots_submitted: set[str] = set()
 
     def post_transfer(
         self,
@@ -207,18 +122,35 @@ def post_transfer(
                         name=lamella_name,
                         number=lamella_number,
                     )
+                if not (source := _get_source(transferred_file, environment)):
+                    logger.warning(f"No source found for file {transferred_file}")
+                    return
+                if not (
+                    destination_file := _file_transferred_to(
+                        environment=environment,
+                        source=source,
+                        file_path=transferred_file,
+                        rsync_basepath=Path(
+                            self._machine_config.get("rsync_basepath", "")
+                        ),
+                    )
+                ):
+                    logger.warning(
+                        f"File {transferred_file.name!r} not found on storage system"
+                    )
+                    return
                 if not self._milling.get(lamella_number):
                     self._milling[lamella_number] = [
                         MillingProgress(
                             timestamp=timestamp,
-                            file=transferred_file,
+                            file=destination_file,
                         )
                     ]
                 else:
                     self._milling[lamella_number].append(
                         MillingProgress(
                             timestamp=timestamp,
-                            file=transferred_file,
+                            file=destination_file,
                         )
                     )
                 gif_list = [
@@ -227,26 +159,25 @@ def post_transfer(
                         self._milling[lamella_number], key=lambda x: x.timestamp
                     )
                 ]
-                if environment:
-                    raw_directory = Path(
-                        environment.default_destinations[self._basepath]
-                    ).name
-                    # post gif list to gif making API call
-                    capture_post(
-                        base_url=str(environment.url.geturl()),
-                        router_name="workflow.correlative_router",
-                        function_name="make_gif",
-                        token=self._token,
-                        instrument_name=environment.instrument_name,
-                        year=datetime.now().year,
-                        visit_name=environment.visit,
-                        session_id=environment.murfey_session,
-                        data={
-                            "lamella_number": lamella_number,
-                            "images": gif_list,
-                            "raw_directory": raw_directory,
-                        },
-                    )
+                raw_directory = Path(
+                    environment.default_destinations[self._basepath]
+                ).name
+                # Submit job to backend to construct a GIF
+                capture_post(
+                    base_url=str(environment.url.geturl()),
+                    router_name="workflow.correlative_router",
+                    function_name="make_gif",
+                    token=self._token,
+                    instrument_name=environment.instrument_name,
+                    year=datetime.now().year,
+                    visit_name=environment.visit,
+                    session_id=environment.murfey_session,
+                    data={
+                        "lamella_number": lamella_number,
+                        "images": [str(file) for file in gif_list],
+                        "raw_directory": raw_directory,
+                    },
+                )
             elif transferred_file.name == "ProjectData.dat":
                 with open(transferred_file, "r") as dat:
                     try:
@@ -269,34 +200,11 @@ def post_transfer(
         # Maps
         # -----------------------------------------------------------------------------
         elif self._acquisition_software == "maps":
-            # Electron snapshot metadata file
-            if transferred_file.name == "EMproject.emxml":
-                # Extract all "Electron Snapshot" metadata and store it
-                self._electron_snapshot_metadata = _parse_electron_snapshot_metadata(
-                    transferred_file
-                )
-                # If dataset hasn't been transferred, register it
-                for dataset_name in list(self._electron_snapshot_metadata.keys()):
-                    if dataset_name not in self._electron_snapshots_submitted:
-                        if dataset_name in self._electron_snapshots:
-                            logger.info(f"Registering {dataset_name!r}")
-
-                            ## Workflow to trigger goes here
-
-                            # Clear old entry after triggering workflow
-                            self._electron_snapshots_submitted.add(dataset_name)
-                            with lock:
-                                self._electron_snapshots.pop(dataset_name, None)
-                                self._electron_snapshot_metadata.pop(dataset_name, None)
-                        else:
-                            logger.debug(f"Waiting for image for {dataset_name}")
-            # Electron snapshot image
-            elif (
+            if (
+                # Electron snapshot images are grid atlases
                 "Electron Snapshot" in transferred_file.name
                 and transferred_file.suffix in (".tif", ".tiff")
             ):
-                # Store file in Context memory
-                dataset_name = transferred_file.stem
                 if not (source := _get_source(transferred_file, environment)):
                     logger.warning(f"No source found for file {transferred_file}")
                     return
@@ -314,24 +222,35 @@ def post_transfer(
                         f"File {transferred_file.name!r} not found on storage system"
                     )
                     return
-                self._electron_snapshots[dataset_name] = destination_file
-
-                if dataset_name not in self._electron_snapshots_submitted:
-                    # If the metadata and image are both present, register dataset
-                    if dataset_name in list(self._electron_snapshot_metadata.keys()):
-                        logger.info(f"Registering {dataset_name!r}")
 
-                        ## Workflow to trigger goes here
+                # Register image in database
+                self._register_atlas(destination_file, environment)
+                return
 
-                        # Clear old entry after triggering workflow
-                        self._electron_snapshots_submitted.add(dataset_name)
-                        with lock:
-                            self._electron_snapshots.pop(dataset_name, None)
-                            self._electron_snapshot_metadata.pop(dataset_name, None)
-                    else:
-                        logger.debug(f"Waiting for metadata for {dataset_name}")
         # -----------------------------------------------------------------------------
         # Meteor
         # -----------------------------------------------------------------------------
         elif self._acquisition_software == "meteor":
             pass
+
+    def _register_atlas(self, file: Path, environment: MurfeyInstanceEnvironment):
+        """
+        Constructs the URL and dictionary to be posted to the server, which then triggers
+        the processing of the electron snapshot image.
+        """
+
+        try:
+            capture_post(
+                base_url=str(environment.url.geturl()),
+                router_name="workflow_fib.router",
+                function_name="register_fib_atlas",
+                token=self._token,
+                instrument_name=environment.instrument_name,
+                data={"file": str(file)},
+                session_id=environment.murfey_session,
+            )
+            logger.info(f"Registering atlas image {file.name!r}")
+            return True
+        except Exception as e:
+            logger.error(f"Error encountered registering atlas image {file.name}:\n{e}")
+            return False

src/murfey/server/api/workflow.py

@@ -4,6 +4,7 @@
 from pathlib import Path
 from typing import Any, Dict, List, Optional
 
+import numpy as np
 import sqlalchemy
 from fastapi import APIRouter, Depends
 from ispyb.sqlalchemy import (
@@ -1208,29 +1209,43 @@ async def make_gif(
     ]
     output_dir = (
         (machine_config.rsync_basepath or Path("")).resolve()
-        / secure_filename(year)
+        / secure_filename(str(year))
         / secure_filename(visit_name)
         / "processed"
     )
     output_dir.mkdir(exist_ok=True)
     output_dir = output_dir / secure_filename(gif_params.raw_directory)
     output_dir.mkdir(exist_ok=True)
     output_path = output_dir / f"lamella_{gif_params.lamella_number}_milling.gif"
-    image_full_paths = [
-        output_dir.parent / gif_params.raw_directory / i for i in gif_params.images
-    ]
+
     if Image is not None:
-        images = [Image.open(f) for f in image_full_paths]
+        images = [Image.open(f) for f in gif_params.images]
     else:
         images = []
     for im in images:
         im.thumbnail((512, 512))
-    images[0].save(
+
+    # Normalize and convert individual frames to 8-bit
+    arr: list[np.ndarray] = []
+    for im in images:
+        frame = np.array(im).astype(np.float32)
+        vmin, vmax = np.percentile(frame, (0.5, 99.5))
+        scale = 255 / ((vmax - vmin) or 1)
+        np.clip(frame, a_min=vmin, a_max=vmax, out=frame)
+        np.subtract(frame, vmin, out=frame)
+        np.multiply(frame, scale, out=frame)
+        arr.append(frame.astype(np.uint8))
+    arr = np.array(arr).astype(np.uint8)
+
+    # Convert back to Image objects and save as GIF
+    converted = [Image.fromarray(arr[f], mode="L") for f in range(len(images))]
+    converted[0].save(
         output_path,
         format="GIF",
-        append_images=images[1:],
+        append_images=converted[1:],
         save_all=True,
         duration=30,
         loop=0,
     )
+
     return {"output_gif": str(output_path)}