fix(omny): let OmnyFermatScan.corridor_size accept None, auto-estimate

omny_fermat_scan.py's corridor_size was Annotated[float] = 3, never
updated to accept None the way flomni_fermat_scan.py's already does.
But omny.py's tomo_scan_projection() calling code passes
corridor_size=None whenever omny.corridor_size (a property defaulting
to -1) hasn't been set to a positive value -- producing
ScanInputValidationError: Invalid type for scan argument 'corridor_size':
None is neither float or int.

Ported flomni's fix directly: corridor_size is now
Annotated[float | None] = None, capped at the new MAX_CORRIDOR_SIZE
(3 um) class constant if a caller passes something larger, and
prepare_scan() now calls the same _estimate_corridor_size() static
method flomni uses (median nearest-neighbor distance * 1.5, capped at
MAX_CORRIDOR_SIZE) to pick a sensible corridor size from the actual
scan positions whenever corridor_size is None, instead of crashing.

2 new tests for _estimate_corridor_size() in test_omny_fermat_scan.py.

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()
 

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)