feat: Hamamatsu recording frame-rate cap (INTERNALFRAMERATE + fast readout)

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_hamamatsu.py

@@ -210,6 +210,61 @@ def get_strobe_time(self) -> float:
 
         return (line_interval_s + trigger_delay_s) * 1000.0
 
+    def set_frame_rate(self, fps: float) -> float:
+        """Cap the internal free-run rate (DCAM INTERNALFRAMERATE) for recording.
+
+        In CONTINUOUS (internal-trigger) mode the sensor otherwise free-runs at its
+        exposure/readout-limited maximum, forcing the read thread to process frames
+        the recording pipeline only downsamples away.  Setting INTERNALFRAMERATE caps
+        delivery near the requested rate.  Exposure is unaffected: the frame period is
+        constrained to be >= exposure + readout, so a too-fast request is clamped by
+        the camera (reflected in the read-back) while a slower request just adds
+        inter-frame dead time.
+
+        Returns the rate the camera accepted, or the exposure-limited max on any
+        failure, so callers can still size/pace against a real number (base contract).
+        """
+        try:
+            fallback = 1000.0 / self.get_total_frame_time()
+        except Exception:
+            # get_strobe_time() raises on a DCAM read error; honor the "always
+            # returns a usable rate" contract by assuming the requested rate (the
+            # caller then paces/sizes against it, same as if we'd never capped).
+            self._log.exception("could not read frame timing; assuming the requested rate.")
+            fallback = float(fps) if fps and fps > 0 else 0.0
+        if fps is None or fps <= 0:
+            return fallback
+        try:
+            with self._pause_streaming():
+                # Fast readout FIRST: high frame rates need it, and the sensor's
+                # default may be the low-noise (slow) mode whose INTERNALFRAMERATE
+                # ceiling is far below `fps`.  It also raises the valuemax we read
+                # next, so the clamp isn't pinned to the slow-mode limit.  Best-
+                # effort — log if it doesn't take, but still try to set the rate.
+                # NOTE: this leaves the camera in fast readout after a recording
+                # (not restored); acceptable for throughput-oriented recording, see
+                # PR notes / follow-up to make readout speed configurable.
+                if not self._set_prop(DCAM_IDPROP.READOUTSPEED, DCAMPROP.READOUTSPEED.FASTEST):
+                    self._log.warning("Could not set fast readout speed; achievable fps may be limited.")
+
+                attr = self._camera.prop_getattr(DCAM_IDPROP.INTERNALFRAMERATE)
+                if isinstance(attr, bool):
+                    self._log.warning(
+                        "INTERNALFRAMERATE not available on this camera; leaving the internal frame rate uncapped."
+                    )
+                    return fallback
+                target = max(attr.valuemin, min(float(fps), attr.valuemax))
+                if not self._set_prop(DCAM_IDPROP.INTERNALFRAMERATE, target):
+                    return fallback
+                achieved = self._camera.prop_getvalue(DCAM_IDPROP.INTERNALFRAMERATE)
+            if isinstance(achieved, bool):
+                return fallback
+            self._log.info(f"set_frame_rate({fps}) set INTERNALFRAMERATE -> {achieved:.3f} fps")
+            return achieved
+        except Exception:
+            self._log.exception("set_frame_rate failed; falling back to exposure-limited max.")
+            return fallback
+
     def set_frame_format(self, frame_format: CameraFrameFormat):
         if frame_format != CameraFrameFormat.RAW:
             raise ValueError("Only the RAW frame format is supported by this camera.")

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,36 @@ 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 set_frame_rate(self, fps: float) -> float:
+        """Set the frame rate via PRECISE_FRAMERATE option.
+
+        _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
+                 current achievable frame rate without changing settings.
+
+        Returns:
+            The achievable frame rate in frames per second, or current rate if not changed.
+        """
+        if fps is None or fps <= 0:
+            return 1000.0 / self.get_total_frame_time()
+        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 1000.0 / self.get_total_frame_time()
+        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 1000.0 / self.get_total_frame_time()
+        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,