fix(lamni,flomni,omny): re-issue rotation after an interrupted move instead of skipping

A Ctrl+C during lsamrot/fsamroy/osamroy rotation leaves the cached
user_setpoint at the target even though the axis never finished moving
(stop() halts the motor mid-move). The next scan then wrongly concluded
"no rotation required" from setpoint equality alone, leaving LamNI's
air bearings unclamped (confirmed via galil show all) and the beamline
unable to scan until the setpoint was nudged manually.

lamni_rotation() now also requires
lgalil_is_air_off_and_orchestra_enabled() before skipping, since that
flag only goes true once the Galil #CENROT sequence (air-clamp plus
centering) has fully completed. flomni_rotation()/omny_rotation() use
a readback-vs-tolerance check instead, since those axes have no
air-bearing clamp step.

Along the way, fixed lgalil_is_air_off_and_orchestra_enabled() itself:
it did bool(socket_put_and_receive(...)) directly on the raw Galil
reply string, which is always truthy, so it could never report
anything but True. Confirmed via HW testing in LamNI.

csaxs_bec/devices/omny/galil/lgalil_ophyd.py

@@ -172,8 +172,8 @@ class LamniGalilController(GalilController):
 
     def lgalil_is_air_off_and_orchestra_enabled(self) -> bool:
         # TODO: move this to the LamNI-specific controller
-        rt_not_blocked_by_galil = bool(self.socket_put_and_receive("MG@OUT[9]"))
-        air_off = bool(self.socket_put_and_receive("MG@OUT[13]"))
+        rt_not_blocked_by_galil = bool(float(self.socket_put_and_receive("MG@OUT[9]")))
+        air_off = bool(float(self.socket_put_and_receive("MG@OUT[13]")))
         return rt_not_blocked_by_galil and air_off
 
 

csaxs_bec/scans/flomni_fermat_scan.py

@@ -371,7 +371,11 @@ class FlomniFermatScan(ScanBase):
             angle (float): The target angle for the flomni rotation.
         """
         fsamroy_current_setpoint = self.dev.fsamroy.user_setpoint.get()
-        if angle == fsamroy_current_setpoint:
+        fsamroy_readback = self.dev.fsamroy.readback.get()
+        already_at_angle = angle == fsamroy_current_setpoint and np.isclose(
+            fsamroy_readback, angle, atol=self.dev.fsamroy.tolerance
+        )
+        if already_at_angle:
             logger.info("No rotation required")
             return
 

csaxs_bec/scans/lamni_fermat_scan.py

@@ -481,14 +481,21 @@ class LamniFermatScan(ScanBase):
     def lamni_rotation(self, angle: float):
         """
         Rotate LamNI to the specified angle. The rotation is only performed
-        if the current setpoint of the rotation stage is different from the requested angle.
+        if the current setpoint of the rotation stage is different from the requested angle,
+        or the stage did not actually settle into measurement configuration
+        (e.g. a previous rotation was interrupted before the air bearings clamped
+        and the piezo stage was released back to interferometer feedback).
 
         Args:
             angle (float): Rotation angle in degrees
         """
         # get last setpoint (cannot be based on pos get because they will deviate slightly)
         lsamrot_current_setpoint = self.dev.lsamrot.user_setpoint.get()
-        if angle == lsamrot_current_setpoint:
+        already_at_angle = (
+            angle == lsamrot_current_setpoint
+            and self.dev.lsamrot.controller.lgalil_is_air_off_and_orchestra_enabled()
+        )
+        if already_at_angle:
             logger.info("No rotation required")
             return
 

csaxs_bec/scans/omny_fermat_scan.py

@@ -347,7 +347,11 @@ class OmnyFermatScan(ScanBase):
             angle (float): Rotation angle in degrees.
         """
         osamroy_current_setpoint = self.dev.osamroy.user_setpoint.get()
-        if angle == osamroy_current_setpoint:
+        osamroy_readback = self.dev.osamroy.readback.get()
+        already_at_angle = angle == osamroy_current_setpoint and np.isclose(
+            osamroy_readback, angle, atol=self.dev.osamroy.tolerance
+        )
+        if already_at_angle:
             logger.info("No rotation required.")
             return
         logger.info("Rotating to requested angle")

tests/tests_devices/test_lgalil_air_orchestra.py

@@ -0,0 +1,47 @@
+"""Regression test for LamniGalilController.lgalil_is_air_off_and_orchestra_enabled().
+
+The method used to do `bool(self.socket_put_and_receive(...))` directly on the
+raw Galil "MG" string reply (e.g. "0.0000"), which is always truthy in Python
+regardless of the actual register value -- the method could never report
+anything but True. Every other digital-register read in this module goes
+through float(...) first; this was the one place missing it.
+"""
+
+import pytest
+
+from csaxs_bec.devices.omny.galil.lgalil_ophyd import LamniGalilController
+from csaxs_bec.devices.sim.sim_lamni import SimLamniGalilMotor
+from csaxs_bec.devices.sim.sim_socket import SimStateRegistry
+
+HOST = "mpc2680.psi.ch"
+PORT = 8081
+
+
+@pytest.fixture(autouse=True)
+def reset_sim_state():
+    """Avoid cross-test leakage of shared (host, port)-keyed simulation state."""
+    SimStateRegistry.reset()
+    yield
+    SimStateRegistry.reset()
+
+
+@pytest.fixture
+def lsamrot(dm_with_devices):
+    LamniGalilController._reset_controller()
+    motor = SimLamniGalilMotor("C", name="lsamrot", host=HOST, port=PORT, device_manager=dm_with_devices)
+    motor.controller.on()
+    yield motor
+    motor.controller.off()
+    motor.controller._reset_controller()
+
+
+def test_lgalil_is_air_off_and_orchestra_enabled_reflects_registers(lsamrot):
+    # sim default: settled (air off, RT feedback not blocked)
+    assert lsamrot.controller.lgalil_is_air_off_and_orchestra_enabled() is True
+
+    lsamrot.controller.socket_put_confirmed("CB9")  # RT feedback blocked
+    assert lsamrot.controller.lgalil_is_air_off_and_orchestra_enabled() is False
+
+    lsamrot.controller.socket_put_confirmed("SB9")
+    lsamrot.controller.socket_put_confirmed("CB13")  # air back on
+    assert lsamrot.controller.lgalil_is_air_off_and_orchestra_enabled() is False

tests/tests_scans/test_flomni_fermat_scan.py

@@ -0,0 +1,45 @@
+from unittest import mock
+
+from csaxs_bec.scans.flomni_fermat_scan import FlomniFermatScan
+
+
+def _scan_stub(user_setpoint: float, readback: float, tolerance: float = 0.5):
+    """A minimal stand-in for a FlomniFermatScan instance, isolated to just the
+    attributes flomni_rotation() touches (self.dev.fsamroy, self.actions).
+    """
+    scan = mock.MagicMock()
+    scan.scan_info.metadata = {"RID": "rid-test"}
+    scan.dev.fsamroy.user_setpoint.get.return_value = user_setpoint
+    scan.dev.fsamroy.readback.get.return_value = readback
+    scan.dev.fsamroy.tolerance = tolerance
+    return scan
+
+
+def test_flomni_rotation_moves_when_setpoint_differs():
+    scan = _scan_stub(user_setpoint=5.0, readback=10.0, tolerance=0.5)
+
+    FlomniFermatScan.flomni_rotation(scan, 10.0)
+
+    scan.actions.set.assert_called_once()
+    args, kwargs = scan.actions.set.call_args
+    assert args[0] is scan.dev.fsamroy
+    assert args[1] == 10.0
+    assert kwargs.get("wait") is False
+
+
+def test_flomni_rotation_skipped_when_settled_at_target():
+    scan = _scan_stub(user_setpoint=10.0, readback=10.1, tolerance=0.5)
+
+    FlomniFermatScan.flomni_rotation(scan, 10.0)
+
+    scan.actions.set.assert_not_called()
+
+
+def test_flomni_rotation_moves_when_setpoint_matches_but_readback_far_off():
+    # simulates a previous rotation interrupted before the axis actually
+    # reached the target: the setpoint cache matches, the readback doesn't
+    scan = _scan_stub(user_setpoint=10.0, readback=2.0, tolerance=0.5)
+
+    FlomniFermatScan.flomni_rotation(scan, 10.0)
+
+    scan.actions.set.assert_called_once()

tests/tests_scans/test_lamni_fermat_scan.py

@@ -0,0 +1,46 @@
+from unittest import mock
+
+from csaxs_bec.scans.lamni_fermat_scan import LamniFermatScan
+
+
+def _scan_stub(user_setpoint: float, orchestra_enabled: bool):
+    """A minimal stand-in for a LamniFermatScan instance, isolated to just the
+    attributes lamni_rotation() touches (self.dev.lsamrot, self.actions).
+    """
+    scan = mock.MagicMock()
+    scan.dev.lsamrot.user_setpoint.get.return_value = user_setpoint
+    scan.dev.lsamrot.controller.lgalil_is_air_off_and_orchestra_enabled.return_value = (
+        orchestra_enabled
+    )
+    return scan
+
+
+def test_lamni_rotation_moves_when_setpoint_differs():
+    scan = _scan_stub(user_setpoint=5.0, orchestra_enabled=True)
+
+    LamniFermatScan.lamni_rotation(scan, 10.0)
+
+    # a genuine target change must move regardless of the orchestra-enabled
+    # state, and must not even need to query it (short-circuit)
+    scan.dev.lsamrot.controller.lgalil_is_air_off_and_orchestra_enabled.assert_not_called()
+    scan.dev.lsamrot.set.assert_called_once_with(10.0)
+    scan.dev.lsamrot.set.return_value.wait.assert_called_once_with()
+
+
+def test_lamni_rotation_skipped_when_settled_at_target():
+    scan = _scan_stub(user_setpoint=10.0, orchestra_enabled=True)
+
+    LamniFermatScan.lamni_rotation(scan, 10.0)
+
+    scan.dev.lsamrot.set.assert_not_called()
+
+
+def test_lamni_rotation_moves_when_setpoint_matches_but_not_settled():
+    # simulates a previous rotation interrupted before the air bearings
+    # clamped and the piezo stage was released back to interferometer feedback
+    scan = _scan_stub(user_setpoint=10.0, orchestra_enabled=False)
+
+    LamniFermatScan.lamni_rotation(scan, 10.0)
+
+    scan.dev.lsamrot.set.assert_called_once_with(10.0)
+    scan.dev.lsamrot.set.return_value.wait.assert_called_once_with()

tests/tests_scans/test_omny_fermat_scan.py

@@ -0,0 +1,44 @@
+from unittest import mock
+
+from csaxs_bec.scans.omny_fermat_scan import OmnyFermatScan
+
+
+def _scan_stub(user_setpoint: float, readback: float, tolerance: float = 0.5):
+    """A minimal stand-in for an OmnyFermatScan instance, isolated to just the
+    attributes omny_rotation() touches (self.dev.osamroy, self.actions).
+    """
+    scan = mock.MagicMock()
+    scan.dev.osamroy.user_setpoint.get.return_value = user_setpoint
+    scan.dev.osamroy.readback.get.return_value = readback
+    scan.dev.osamroy.tolerance = tolerance
+    return scan
+
+
+def test_omny_rotation_moves_when_setpoint_differs():
+    scan = _scan_stub(user_setpoint=5.0, readback=10.0, tolerance=0.5)
+
+    OmnyFermatScan.omny_rotation(scan, 10.0)
+
+    scan.actions.set.assert_called_once()
+    args, kwargs = scan.actions.set.call_args
+    assert args[0] is scan.dev.osamroy.user_setpoint
+    assert args[1] == 10.0
+    assert kwargs.get("wait") is False
+
+
+def test_omny_rotation_skipped_when_settled_at_target():
+    scan = _scan_stub(user_setpoint=10.0, readback=10.1, tolerance=0.5)
+
+    OmnyFermatScan.omny_rotation(scan, 10.0)
+
+    scan.actions.set.assert_not_called()
+
+
+def test_omny_rotation_moves_when_setpoint_matches_but_readback_far_off():
+    # simulates a previous rotation interrupted before the axis actually
+    # reached the target: the setpoint cache matches, the readback doesn't
+    scan = _scan_stub(user_setpoint=10.0, readback=2.0, tolerance=0.5)
+
+    OmnyFermatScan.omny_rotation(scan, 10.0)
+
+    scan.actions.set.assert_called_once()