feat: Record + Z-Stack acquisition mode

software/control/acquisition_yaml_loader.py

@@ -51,6 +51,77 @@ class AcquisitionYAMLData:
     flexible_positions: Optional[List[Dict]] = None  # [{name, center_mm}, ...]
 
 
+@dataclass
+class RecordZStackYAMLData:
+    """Parsed record/z-stack acquisition YAML data structure."""
+
+    widget_type: str  # "record_zstack"
+    xy_mode: str = "Select Wells"
+
+    objective_name: Optional[str] = None
+    camera_binning: Optional[Tuple[int, int]] = None
+
+    nt: int = 1
+    delta_t_s: float = 0.0
+
+    laser_af: bool = False
+
+    recording_enabled: bool = False
+    recording_channel: Optional[Dict] = None
+    fps: float = 10.0
+    duration_s: float = 1.0
+    recording_z_offset_um: float = 0.0
+
+    zstack_enabled: bool = False
+    zstack_channels: List[Dict] = field(default_factory=list)
+    z_min_um: float = -3.0
+    z_max_um: float = 3.0
+    z_step_um: float = 1.0
+
+    scan_size_mm: Optional[float] = None
+    overlap_percent: float = 10.0
+    wellplate_regions: Optional[List[Dict]] = None
+
+
+def _parse_camera_binning(obj: dict) -> Optional[Tuple[int, int]]:
+    binning = obj.get("camera_binning")
+    if binning and isinstance(binning, list) and len(binning) == 2:
+        return tuple(binning)
+    return None
+
+
+def _parse_record_zstack_yaml_data(data: dict, acq: dict) -> RecordZStackYAMLData:
+    obj = data.get("objective", {})
+    time_series = data.get("time_series", {})
+    autofocus = data.get("autofocus", {})
+    recording = data.get("recording", {})
+    z_stack = data.get("z_stack", {})
+    wellplate_scan = data.get("wellplate_scan", {})
+
+    return RecordZStackYAMLData(
+        widget_type="record_zstack",
+        xy_mode=acq.get("xy_mode", "Select Wells"),
+        objective_name=obj.get("name"),
+        camera_binning=_parse_camera_binning(obj),
+        nt=time_series.get("nt", 1),
+        delta_t_s=time_series.get("delta_t_s", 0.0),
+        laser_af=autofocus.get("laser_af", False),
+        recording_enabled=recording.get("enabled", False),
+        recording_channel=recording.get("channel"),
+        fps=recording.get("fps", 10.0),
+        duration_s=recording.get("duration_s", 1.0),
+        recording_z_offset_um=recording.get("z_offset_um", 0.0),
+        zstack_enabled=z_stack.get("enabled", False),
+        zstack_channels=z_stack.get("channels", []),
+        z_min_um=z_stack.get("z_min_um", -3.0),
+        z_max_um=z_stack.get("z_max_um", 3.0),
+        z_step_um=z_stack.get("z_step_um", 1.0),
+        scan_size_mm=wellplate_scan.get("scan_size_mm"),
+        overlap_percent=wellplate_scan.get("overlap_percent", 10.0),
+        wellplate_regions=wellplate_scan.get("regions"),
+    )
+
+
 def parse_acquisition_yaml(file_path: str) -> AcquisitionYAMLData:
     """Parse acquisition YAML file and return structured data.
 
@@ -82,11 +153,14 @@ def parse_acquisition_yaml(file_path: str) -> AcquisitionYAMLData:
     flexible_scan = data.get("flexible_scan", {})
 
     # Validate widget_type
-    VALID_WIDGET_TYPES = ("wellplate", "flexible")
+    VALID_WIDGET_TYPES = ("wellplate", "flexible", "record_zstack")
     widget_type = acq.get("widget_type", "wellplate")
     if widget_type not in VALID_WIDGET_TYPES:
         raise ValueError(f"Invalid widget_type '{widget_type}'. Must be one of: {VALID_WIDGET_TYPES}")
 
+    if widget_type == "record_zstack":
+        return _parse_record_zstack_yaml_data(data, acq)
+
     # Parse camera binning
     binning = obj.get("camera_binning")
     if binning and isinstance(binning, list) and len(binning) == 2:

software/control/camera_toupcam.py

@@ -20,6 +20,7 @@
 
 import threading
 import control.toupcam as toupcam
+import control.toupcam_exceptions
 from control.toupcam_exceptions import hresult_checker
 
 log = squid.logging.get_logger(__name__)
@@ -50,6 +51,21 @@ def get_sn_by_model(camera_model: ToupcamCameraModel):
     return None  # return None if no device with the specified model_name is connected
 
 
+def clamp_precise_framerate_tenths(fps: float, min_tenths: int, max_tenths: int) -> int:
+    """Clamp fps (in frames per second) to the allowed range in tenths.
+
+    Args:
+        fps: Desired frame rate in frames per second
+        min_tenths: Minimum allowed value in tenths (0.1 fps units)
+        max_tenths: Maximum allowed value in tenths (0.1 fps units)
+
+    Returns:
+        Clamped value in tenths of fps
+    """
+    tenths = int(round(fps * 10.0))
+    return max(min_tenths, min(max_tenths, tenths))
+
+
 class ToupcamCamera(AbstractCamera):
     TOUPCAM_OPTION_RAW_RAW_VAL = 1
     TOUPCAM_OPTION_RAW_RGB_VAL = 0
@@ -556,6 +572,53 @@ def get_exposure_limits(self) -> Tuple[float, float]:
         (min_exposure, max_exposure, default_exposure) = self._camera.get_ExpTimeRange()
         return min_exposure / 1000.0, max_exposure / 1000.0  # us -> ms
 
+    def _continuous_max_framerate(self) -> float:
+        """Highest frame rate the sensor can sustain in CONTINUOUS (free-run) mode, in fps.
+
+        In continuous mode the exposure pipelines with sensor readout, so the frame period
+        is max(readout, exposure).  This deliberately does NOT use get_total_frame_time()
+        (== readout + trigger_delay + exposure), which is the *sequential* software/hardware-
+        trigger period and under-reports the free-run rate by ~2x.  strobe_time_us is the
+        pure, exposure-independent readout period.  The triggered-mode timing getters
+        (get_strobe_time/get_total_frame_time/_calculate_strobe_info) are left untouched.
+        """
+        readout_ms = self._strobe_info.strobe_time_us / 1000.0
+        frame_ms = max(readout_ms, self.get_exposure_time())
+        return 1000.0 / frame_ms
+
+    def set_frame_rate(self, fps: float) -> float:
+        """Set the frame rate via the PRECISE_FRAMERATE option (CONTINUOUS mode only).
+
+        _calculate_strobe_info (~:128-140) drives PRECISE_FRAMERATE to MAX on mode
+        switch; set_frame_rate must be called **after** entering CONTINUOUS to take
+        effect, and recording restores nothing (next acquisition resets exposure → MAX again).
+
+        Args:
+            fps: Desired frame rate in frames per second. If None or <= 0, returns
+                 the camera's achievable continuous maximum without changing settings.
+
+        Returns:
+            The achievable frame rate in fps.  When the PRECISE_FRAMERATE option can be
+            read/set, that is the clamped requested rate; otherwise (the option is
+            unavailable on this model) it is the sensor's readout/exposure-limited
+            continuous maximum (see _continuous_max_framerate).
+        """
+        if fps is None or fps <= 0:
+            return self._continuous_max_framerate()
+        try:
+            max_tenths = self._camera.get_Option(toupcam.TOUPCAM_OPTION_MAX_PRECISE_FRAMERATE)
+            min_tenths = self._camera.get_Option(toupcam.TOUPCAM_OPTION_MIN_PRECISE_FRAMERATE)
+        except toupcam.HRESULTException as ex:
+            self._log.warning(f"precise-framerate range read failed: {control.toupcam_exceptions.explain(ex)}")
+            return self._continuous_max_framerate()
+        tenths = clamp_precise_framerate_tenths(fps, min_tenths, max_tenths)
+        try:
+            self._camera.put_Option(toupcam.TOUPCAM_OPTION_PRECISE_FRAMERATE, tenths)
+        except toupcam.HRESULTException as ex:
+            self._log.warning(f"set precise-framerate failed: {control.toupcam_exceptions.explain(ex)}")
+            return self._continuous_max_framerate()
+        return tenths / 10.0
+
     @staticmethod
     def _user_gain_to_toupcam(user_gain):
         """

software/control/core/acquisition_setup.py

@@ -0,0 +1,59 @@
+"""Shared acquisition setup helpers.
+
+Free functions used by MultiPointController (and future controllers such as
+RecordZStackController) to set up experiment directories without duplicating
+logic across controller classes.
+"""
+
+import os
+from datetime import datetime
+from typing import Optional, Tuple
+
+from control import utils
+
+
+def compute_pixel_size_um(objective_store, camera) -> Optional[float]:
+    """Compute the physical pixel size in µm from objective and camera metadata.
+
+    Returns the product of the objective's pixel-size factor and the camera's
+    binned pixel size in µm, or None if either value is unavailable or an
+    exception is raised.
+
+    Args:
+        objective_store: ObjectiveStore (or compatible object) with
+            ``get_pixel_size_factor() -> Optional[float]``.
+        camera: AbstractCamera (or compatible) with
+            ``get_pixel_size_binned_um() -> Optional[float]``.
+
+    Returns:
+        Pixel size in µm, or None.
+    """
+    try:
+        pixel_factor = objective_store.get_pixel_size_factor()
+        sensor_pixel_um = camera.get_pixel_size_binned_um()
+        if pixel_factor is not None and sensor_pixel_um is not None:
+            return float(pixel_factor) * float(sensor_pixel_um)
+        return None
+    except Exception:
+        return None
+
+
+def create_experiment_dir(base_path: str, experiment_id: str) -> Tuple[str, str]:
+    """Resolve a unique experiment ID and create its output directory.
+
+    Appends a timestamp to *experiment_id* (spaces replaced with underscores)
+    to guarantee uniqueness, then creates the directory tree under *base_path*.
+
+    Args:
+        base_path: Root directory for all experiments.
+        experiment_id: Human-readable experiment name supplied by the user.
+
+    Returns:
+        A ``(resolved_id, dir_path)`` tuple where *resolved_id* is the
+        timestamped identifier and *dir_path* is the absolute path of the
+        newly created directory.
+    """
+    resolved_id = experiment_id.replace(" ", "_") + "_" + datetime.now().strftime("%Y-%m-%d_%H-%M-%S.%f")
+    dir_path = os.path.join(base_path, resolved_id)
+    utils.ensure_directory_exists(dir_path)
+    return resolved_id, dir_path

software/control/core/multi_point_controller.py

@@ -6,7 +6,6 @@
 import tempfile
 import time
 import yaml
-from datetime import datetime
 from enum import Enum
 from threading import Thread
 from typing import Optional, Tuple, Any
@@ -17,6 +16,7 @@
 from control import utils, utils_acquisition
 import control._def
 from control.core.auto_focus_controller import AutoFocusController
+from control.core.acquisition_setup import create_experiment_dir
 from control.core.multi_point_utils import MultiPointControllerFunctions, ScanPositionInformation, AcquisitionParameters
 from control.core.scan_coordinates import ScanCoordinates
 from control.core.laser_auto_focus_controller import LaserAutofocusController
@@ -438,12 +438,9 @@ def set_overlap_percent(self, overlap_percent: float):
         self.overlap_percent = overlap_percent
 
     def start_new_experiment(self, experiment_ID):  # @@@ to do: change name to prepare_folder_for_new_experiment
-        # generate unique experiment ID
-        self.experiment_ID = experiment_ID.replace(" ", "_") + "_" + datetime.now().strftime("%Y-%m-%d_%H-%M-%S.%f")
+        # generate unique experiment ID and create its output directory
+        self.experiment_ID, experiment_dir = create_experiment_dir(self.base_path, experiment_ID)
         self.recording_start_time = time.time()
-        # create a new folder
-        experiment_dir = os.path.join(self.base_path, self.experiment_ID)
-        utils.ensure_directory_exists(experiment_dir)
         # Save acquisition configuration via ConfigRepository
         self.liveController.microscope.config_repo.save_acquisition_output(
             output_dir=experiment_dir,