feat(ome-tiff): opt-in per-timepoint split for offloadable OME-TIFF acquisitions

software/control/_def.py

@@ -871,6 +871,8 @@ class SOFTWARE_POS_LIMIT:
 DISK_SPACE_RESERVE_GB = 10.0  # Minimum free space to keep on the save disk before pausing
 DISK_SPACE_POLL_INTERVAL_S = 5.0  # How often to re-check free space while paused
 SIMULATED_DISK_CAPACITY_GB = 0.0  # Dev only: pretend the save disk has this capacity (0 = use the real disk)
+# Write one OME-TIFF per timepoint (<t>/ome_tiff/...) so timepoints can be offloaded during the run
+OME_TIFF_SPLIT_TIMEPOINTS = False
 
 CAMERA_SN = {"ch 1": "SN1", "ch 2": "SN2"}  # for multiple cameras, to be overwritten in the configuration file
 

software/control/acquisition_yaml_loader.py

@@ -62,6 +62,9 @@ class AcquisitionYAMLData:
     # Per-run opt-in for large acquisition mode. None = the file did not say, so the controller's
     # current flag is left alone (the global setting still applies when the run's parameters are built).
     large_acquisition_mode: Optional[bool] = None
+    # Per-run opt-in for the OME-TIFF per-timepoint split. None = the file did not say, so the
+    # controller's current flag is left alone (the global setting still applies at build_params time).
+    ome_tiff_split_timepoints: Optional[bool] = None
 
 
 def parse_acquisition_yaml(file_path: str) -> AcquisitionYAMLData:
@@ -186,6 +189,9 @@ def parse_acquisition_dict(data: dict, source: str = "<dict>") -> AcquisitionYAM
         large_acquisition_mode=(
             None if acq.get("large_acquisition_mode") is None else bool(acq.get("large_acquisition_mode"))
         ),
+        ome_tiff_split_timepoints=(
+            None if acq.get("ome_tiff_split_timepoints") is None else bool(acq.get("ome_tiff_split_timepoints"))
+        ),
     )
 
 

software/control/core/acquisition_settings.py

@@ -189,6 +189,9 @@ def apply_acquisition_settings(
         # Only an explicit key opts this run in; absent leaves the controller's flag alone (a pre-flight
         # dialog may have set it). The global setting is folded in when the run's parameters are built.
         controller.set_large_acquisition_mode(data.large_acquisition_mode)
+    if data.ome_tiff_split_timepoints is not None:
+        # Same rule as large_acquisition_mode: only an explicit key opts this run in.
+        controller.set_split_ome_timepoints(data.ome_tiff_split_timepoints)
     controller.set_widget_type(data.widget_type)
     controller.set_scan_size(float(data.scan_size_mm or 0.0))
     controller.set_overlap_percent(float(data.overlap_percent))

software/control/core/job_processing.py

@@ -51,6 +51,13 @@ class AcquisitionInfo:
         physical_size_z_um: Z step size in micrometers (for OME-XML).
         physical_size_x_um: Pixel size in X in micrometers (for OME-XML).
         physical_size_y_um: Pixel size in Y in micrometers (for OME-XML).
+        split_timepoints: When True, each timepoint is written to its own OME-TIFF file
+            (SizeT=1) under that timepoint's folder, so a finished timepoint can be handed to
+            the transfer manifest while the run continues. When False (the default), all
+            timepoints share one pre-allocated file per (region, FOV).
+        acquisition_start_time: Wall-clock start (time.time()) of the whole run, so that
+            per-timepoint files can express DeltaT relative to the acquisition start rather
+            than to their own first plane. Only used when split_timepoints is True.
     """
 
     total_time_points: int
@@ -62,6 +69,8 @@ class AcquisitionInfo:
     physical_size_z_um: Optional[float] = None
     physical_size_x_um: Optional[float] = None
     physical_size_y_um: Optional[float] = None
+    split_timepoints: bool = False
+    acquisition_start_time: Optional[float] = None
 
 
 # NOTE(imo): We want this to be fast.  But pydantic does not support numpy serialization natively, which means
@@ -297,10 +306,15 @@ def run(self) -> SaveResult:
             ome_folder = ome_tiff_writer.ome_output_folder(self.acquisition_info, self.capture_info)
             base_name = ome_tiff_writer.ome_base_name(self.capture_info)
             stack_key = os.path.join(ome_folder, base_name)
+            split = self.acquisition_info.split_timepoints
+            time_point = self.capture_info.time_point or 0
+            if split:
+                # One simulated stack per timepoint, mirroring the per-timepoint files on disk.
+                stack_key = f"{stack_key}:t{time_point}"
 
-            # Determine 5D shape (T, Z, C, Y, X)
+            # Determine 5D shape (T, Z, C, Y, X); a split stack holds a single timepoint.
             shape = (
-                self.acquisition_info.total_time_points,
+                1 if split else self.acquisition_info.total_time_points,
                 self.acquisition_info.total_z_levels,
                 self.acquisition_info.total_channels,
                 image.shape[0],
@@ -311,7 +325,7 @@ def run(self) -> SaveResult:
                 image=image,
                 stack_key=stack_key,
                 shape=shape,
-                time_point=self.capture_info.time_point or 0,
+                time_point=0 if split else time_point,
                 z_index=self.capture_info.z_index,
                 channel_index=self.capture_info.configuration_idx,
             )
@@ -374,7 +388,8 @@ def _save_ome_tiff(self, image: np.ndarray, info: CaptureInfo) -> Tuple[str, boo
             target_dtype = np.dtype(metadata[ome_tiff_writer.DTYPE_KEY])
             image_to_store = image if image.dtype == target_dtype else image.astype(target_dtype)
 
-            time_point = int(info.time_point)
+            # A split file holds a single timepoint, so its plane index along T is always 0.
+            time_point = 0 if self.acquisition_info.split_timepoints else int(info.time_point)
             z_index = int(info.z_index)
             channel_index = int(info.configuration_idx)
             shape = tuple(metadata[ome_tiff_writer.SHAPE_KEY])

software/control/core/multi_point_controller.py

@@ -84,6 +84,7 @@ def _save_acquisition_yaml(
             "xy_mode": params.xy_mode,
             "skip_saving": params.skip_saving,
             "large_acquisition_mode": params.large_acquisition_mode,
+            "ome_tiff_split_timepoints": params.split_ome_timepoints,
         },
         "objective": objective_info or {},
         "sample": {
@@ -247,6 +248,8 @@ def __init__(
         self.skip_saving = False
         # Per-run opt-in for large acquisition mode (pre-flight dialog); reset when the run finishes.
         self.large_acquisition_mode = False
+        # Per-run opt-in for the OME-TIFF per-timepoint split (pre-flight dialog); reset when the run finishes.
+        self.split_ome_timepoints = False
         # Single pause control point of the running acquisition; exists only for runs in large
         # acquisition mode (operator pause + disk-space guard hold it), None otherwise.
         self._pause_gate: Optional[PauseGate] = None
@@ -467,6 +470,11 @@ def set_large_acquisition_mode(self, enabled: bool) -> None:
         self.large_acquisition_mode = bool(enabled)
         self._log.info(f"Large acquisition mode for the next run: {self.large_acquisition_mode}")
 
+    def set_split_ome_timepoints(self, enabled: bool) -> None:
+        """Opt this run into one OME-TIFF per timepoint (folded with the global setting in build_params)."""
+        self.split_ome_timepoints = bool(enabled)
+        self._log.info(f"OME-TIFF per-timepoint split for the next run: {self.split_ome_timepoints}")
+
     def set_xy_mode(self, xy_mode):
         self.xy_mode = xy_mode
 
@@ -1109,6 +1117,7 @@ def build_params(
             # Effective mode for this run: per-run opt-in (pre-flight dialog / YAML) or the global
             # setting. Recorded in acquisition.yaml so the run's behaviour is reproducible.
             large_acquisition_mode=self.large_acquisition_mode or control._def.LARGE_ACQUISITION_MODE,
+            split_ome_timepoints=self.split_ome_timepoints or control._def.OME_TIFF_SPLIT_TIMEPOINTS,
             plate_num_rows=plate_num_rows,
             plate_num_cols=plate_num_cols,
             xy_mode=self.xy_mode,
@@ -1122,6 +1131,7 @@ def _on_acquisition_completed(self):
         # A per-run opt-in (the pre-flight dialog) must not leak into the next run. Reset it first so a
         # failure further down this method cannot leave the flag set. The global setting is unaffected.
         self.large_acquisition_mode = False
+        self.split_ome_timepoints = False
         if self._pause_gate is not None:
             self._pause_gate.clear()
             self._pause_gate = None

software/control/core/multi_point_utils.py

@@ -66,6 +66,9 @@ class AcquisitionParameters:
     # Effective large acquisition mode for this run (per-run opt-in OR the global setting), resolved
     # once by MultiPointController.build_params.
     large_acquisition_mode: bool = False
+    # Effective OME-TIFF per-timepoint split for this run (per-run opt-in OR the global setting),
+    # resolved once by MultiPointController.build_params. Ignored by the other file formats.
+    split_ome_timepoints: bool = False
 
     # Plate dimensions (only used when xy_mode is plate-based, e.g. "Select Wells").
     plate_num_rows: int = 8  # For 96-well plate

software/control/core/multi_point_worker.py

@@ -165,6 +165,9 @@ def __init__(
             physical_size_z_um=self._physical_size_z_um,
             physical_size_x_um=self._pixel_size_um,
             physical_size_y_um=self._pixel_size_um,
+            # Opt-in per-timepoint OME-TIFF split (one <t>/ome_tiff/... file per timepoint); off = today's layout.
+            split_timepoints=acquisition_parameters.split_ome_timepoints,
+            acquisition_start_time=self.timestamp_acquisition_started,
         )
 
         self.time_point = 0

software/control/core/utils_ome_tiff_writer.py

@@ -37,16 +37,25 @@
 PHYSICAL_SIZE_X_UNIT_KEY = "physical_size_x_unit"
 PHYSICAL_SIZE_Y_KEY = "physical_size_y"
 PHYSICAL_SIZE_Y_UNIT_KEY = "physical_size_y_unit"
+SPLIT_TIMEPOINTS_KEY = "split_timepoints"
+GLOBAL_TIME_POINT_KEY = "global_time_point"
 
 
 def ome_output_folder(acq_info: "AcquisitionInfo", info: "CaptureInfo") -> str:
+    if acq_info.split_timepoints:
+        # info.save_directory IS the timepoint folder, so each timepoint gets its own stack.
+        return os.path.join(info.save_directory, "ome_tiff")
     base_dir = acq_info.experiment_path or os.path.dirname(info.save_directory)
     return os.path.join(base_dir, "ome_tiff")
 
 
 def metadata_temp_path(acq_info: "AcquisitionInfo", info: "CaptureInfo", base_name: str) -> str:
     base_identifier = acq_info.experiment_path or info.save_directory
     key = f"{base_identifier}:{base_name}"
+    if acq_info.split_timepoints:
+        # Per-timepoint progress files, so an aborted timepoint's leftover progress is never
+        # adopted by the next timepoint (which writes a different file).
+        key = f"{key}:t{info.time_point}"
     digest = hashlib.sha1(key.encode("utf-8")).hexdigest()
     # Use squid_ome_ prefix to avoid conflicts with other OME-TIFF applications
     return os.path.join(tempfile.gettempdir(), f"squid_ome_{digest}_metadata.json")
@@ -94,11 +103,19 @@ def initialize_metadata(acq_info: "AcquisitionInfo", info: "CaptureInfo", image:
     physical_size_x_unit = "µm" if physical_size_x is not None else None
     physical_size_y = float(acq_info.physical_size_y_um) if acq_info.physical_size_y_um is not None else None
     physical_size_y_unit = "µm" if physical_size_y is not None else None
+    split_timepoints = bool(acq_info.split_timepoints)
+    # A split file holds a single timepoint; DeltaT stays relative to the acquisition start so
+    # that the per-timepoint files remain comparable to each other.
+    size_t = 1 if split_timepoints else int(acq_info.total_time_points)
+    if split_timepoints and acq_info.acquisition_start_time is not None:
+        start_time = acq_info.acquisition_start_time
+    else:
+        start_time = info.capture_time
     return {
         DTYPE_KEY: np.dtype(image.dtype).str,
         AXES_KEY: "TZCYX",
         SHAPE_KEY: [
-            int(acq_info.total_time_points),
+            size_t,
             int(acq_info.total_z_levels),
             int(acq_info.total_channels),
             int(image.shape[-2]),
@@ -107,12 +124,12 @@ def initialize_metadata(acq_info: "AcquisitionInfo", info: "CaptureInfo", image:
         CHANNEL_NAMES_KEY: channel_names,
         WRITTEN_INDICES_KEY: [],
         SAVED_COUNT_KEY: 0,
-        EXPECTED_COUNT_KEY: int(acq_info.total_time_points)
-        * int(acq_info.total_z_levels)
-        * int(acq_info.total_channels),
+        EXPECTED_COUNT_KEY: size_t * int(acq_info.total_z_levels) * int(acq_info.total_channels),
         PLANES_KEY: {},
-        START_TIME_KEY: info.capture_time,
+        START_TIME_KEY: start_time,
         COMPLETED_KEY: False,
+        SPLIT_TIMEPOINTS_KEY: split_timepoints,
+        GLOBAL_TIME_POINT_KEY: int(info.time_point) if split_timepoints and info.time_point is not None else None,
         TIME_INCREMENT_KEY: time_increment,
         TIME_INCREMENT_UNIT_KEY: time_increment_unit,
         PHYSICAL_SIZE_Z_KEY: physical_size_z,
@@ -125,9 +142,12 @@ def initialize_metadata(acq_info: "AcquisitionInfo", info: "CaptureInfo", image:
 
 
 def update_plane_metadata(metadata: Dict[str, Any], info: "CaptureInfo") -> Dict[str, Any]:
-    plane_key = f"{info.time_point}-{info.configuration_idx}-{info.z_index}"
+    # In a split file the plane index is local to that file, which holds a single timepoint.
+    # DeltaT below still measures from START_TIME_KEY (the acquisition start when split).
+    the_t = 0 if metadata.get(SPLIT_TIMEPOINTS_KEY) else int(info.time_point)
+    plane_key = f"{the_t}-{info.configuration_idx}-{info.z_index}"
     plane_data: Dict[str, Any] = {
-        "TheT": int(info.time_point),
+        "TheT": the_t,
         "TheZ": int(info.z_index),
         "TheC": int(info.configuration_idx),
     }
@@ -176,6 +196,14 @@ def metadata_for_imwrite(metadata: Dict[str, Any]) -> Dict[str, Any]:
     return meta
 
 
+def ome_description(metadata: Dict[str, Any]) -> Optional[str]:
+    """Description text for a split file, which alone cannot say which timepoint it holds."""
+    global_time_point = metadata.get(GLOBAL_TIME_POINT_KEY)
+    if global_time_point is None:
+        return None
+    return f"Timepoint {global_time_point} of the acquisition (one timepoint per file)"
+
+
 def build_base_ome_xml(metadata: Dict[str, Any]) -> str:
     # Lazy import: xml.etree.ElementTree only needed for XML generation functions
     import xml.etree.ElementTree as ET
@@ -206,6 +234,10 @@ def build_base_ome_xml(metadata: Dict[str, Any]) -> str:
             image.set("AcquisitionDate", acq_time)
         except Exception:
             pass
+    description = ome_description(metadata)
+    if description is not None:
+        # OME requires Description before Pixels, so add it while Image is still empty.
+        ET.SubElement(image, "{ns}Description".format(ns="{" + ns + "}")).text = description
     pixels = ET.SubElement(
         image,
         "{ns}Pixels".format(ns="{" + ns + "}"),
@@ -264,6 +296,15 @@ def augment_ome_xml(existing_xml: Optional[str], metadata: Dict[str, Any]) -> st
         except Exception:
             pass
 
+    description = ome_description(metadata)
+    if description is not None:
+        description_elem = image.find("ome:Description", ns)
+        if description_elem is None:
+            # OME requires Description to precede Pixels.
+            description_elem = ET.Element("{http://www.openmicroscopy.org/Schemas/OME/2016-06}Description")
+            image.insert(0, description_elem)
+        description_elem.text = description
+
     pixels = image.find("ome:Pixels", ns)
     if pixels is None:
         return existing_xml or ""