diff --git a/csaxs_bec/scans/omny_fermat_scan.py b/csaxs_bec/scans/omny_fermat_scan.py index 55a7a391..0a40d64b 100644 --- a/csaxs_bec/scans/omny_fermat_scan.py +++ b/csaxs_bec/scans/omny_fermat_scan.py @@ -20,6 +20,8 @@ import time from typing import Annotated import numpy as np +from scipy.spatial import cKDTree + from bec_lib import messages from bec_lib.logger import bec_logger from bec_lib.scan_args import DefaultArgType, ScanArgument, Units @@ -44,6 +46,8 @@ class OmnyFermatScan(ScanBase): # It must be a valid Python identifier, that is, it can only contain letters, numbers, and underscores, and must not start with a number. scan_name = "omny_fermat_scan" + MAX_CORRIDOR_SIZE = 3 # Corridor size is capped at this value (in um) for stability + gui_config = { "Scan Parameters": [ "fovx", @@ -68,7 +72,7 @@ class OmnyFermatScan(ScanBase): step: Annotated[float, ScanArgument(display_name="Step", description="Step size.", units=Units.µm)], zshift: Annotated[float, ScanArgument(display_name="Zshift", description="Shift in z.", units=Units.µm)], angle: Annotated[float | None, ScanArgument(display_name="Angle", description="Rotation angle (will rotate first)", units=Units.deg)] = None, - corridor_size: Annotated[float, ScanArgument(display_name="Corridor Size", description="Corridor size for the corridor optimization. ", units=Units.µm)] = 3, + corridor_size: Annotated[float | None, ScanArgument(display_name="Corridor Size", description="Corridor size for the corridor optimization. ", units=Units.µm)] = None, exp_time: DefaultArgType.ExposureTime = 0, frames_per_trigger: DefaultArgType.FramesPerTrigger = 1, readout_time: DefaultArgType.ReadoutTime = 0, @@ -86,7 +90,7 @@ class OmnyFermatScan(ScanBase): step (float): Step size. zshift (float): Shift in z. angle (float | None): Rotation angle (will rotate first) - corridor_size (float): Corridor size for the corridor optimization. + corridor_size (float | None): Corridor size for the corridor optimization. exp_time (float): Exposure time in seconds frames_per_trigger (int): Number of frames per trigger for devices that support configurable frame counts per trigger. readout_time (float): Configuration for devices that support configurable readout times. @@ -115,6 +119,13 @@ class OmnyFermatScan(ScanBase): logger.warning("The zshift is smaller than -100 um. It will be limited to -100 um.") self.zshift = -100 + if self.corridor_size is not None and self.corridor_size > self.MAX_CORRIDOR_SIZE: + logger.warning( + f"The corridor_size is larger than {self.MAX_CORRIDOR_SIZE} um. It will be" + f" limited to {self.MAX_CORRIDOR_SIZE} um." + ) + self.corridor_size = self.MAX_CORRIDOR_SIZE + self.update_scan_info( exp_time=exp_time, frames_per_trigger=frames_per_trigger, readout_time=readout_time ) @@ -144,8 +155,12 @@ class OmnyFermatScan(ScanBase): f"The number of positions must exceed 20. Currently: {len(positions)}." ) + corridor_size = self.corridor_size + if corridor_size is None: + corridor_size = min(self._estimate_corridor_size(positions), self.MAX_CORRIDOR_SIZE) + self.positions = self.components.optimize_trajectory( - positions=positions, corridor_size=self.corridor_size, optimization_type="corridor" + positions=positions, corridor_size=corridor_size, optimization_type="corridor" ) flip_axes = self.reverse_trajectory() if flip_axes: @@ -330,6 +345,24 @@ class OmnyFermatScan(ScanBase): positions.append(right_upper_corner) return np.array(positions) + @staticmethod + def _estimate_corridor_size(positions: np.ndarray, factor: float = 1.5) -> float: + """ + Estimate the corridor size based on the median nearest-neighbor distance + of the positions, matching the estimation used by the corridor path + optimizer if no corridor_size is provided. + + Args: + positions (np.ndarray): Array of positions + factor (float): Scaling factor for the median distance + + Returns: + float: Estimated corridor size + """ + tree = cKDTree(positions[:, :2]) + dists, _ = tree.query(positions[:, :2], k=2) # k=1 is itself + return factor * np.median(dists[:, 1]) + def prepare_setup(self): self.dev.rtx.controller.clear_trajectory_generator() diff --git a/tests/tests_scans/test_omny_fermat_scan.py b/tests/tests_scans/test_omny_fermat_scan.py index 089afd71..a7187b55 100644 --- a/tests/tests_scans/test_omny_fermat_scan.py +++ b/tests/tests_scans/test_omny_fermat_scan.py @@ -1,5 +1,7 @@ from unittest import mock +import numpy as np + from csaxs_bec.scans.omny_fermat_scan import OmnyFermatScan @@ -42,3 +44,24 @@ def test_omny_rotation_moves_when_setpoint_matches_but_readback_far_off(): OmnyFermatScan.omny_rotation(scan, 10.0) scan.actions.set.assert_called_once() + + +def test_estimate_corridor_size_returns_positive_value_for_regular_grid(): + # 10x10 grid, 1 um spacing -> nearest-neighbor distance is 1 um everywhere + xs, ys = np.meshgrid(np.arange(10), np.arange(10)) + positions = np.column_stack([xs.ravel(), ys.ravel(), np.zeros(xs.size)]) + + result = OmnyFermatScan._estimate_corridor_size(positions) + + assert result > 0 + assert np.isclose(result, 1.5, atol=1e-6) # factor=1.5 * median nn-distance (1.0) + + +def test_estimate_corridor_size_scales_with_factor(): + xs, ys = np.meshgrid(np.arange(5), np.arange(5)) + positions = np.column_stack([xs.ravel(), ys.ravel(), np.zeros(xs.size)]) + + default_factor = OmnyFermatScan._estimate_corridor_size(positions) + doubled_factor = OmnyFermatScan._estimate_corridor_size(positions, factor=3.0) + + assert np.isclose(doubled_factor, 2 * default_factor)