fix(LamNI): don't auto-retry lsamrot moves aborted by stop()
CI for csaxs_bec / test (pull_request) Successful in 1m50s
Read the Docs Deploy Trigger / trigger-rtd-webhook (push) Successful in 2s
CI for csaxs_bec / test (push) Successful in 1m57s

move_and_finish()'s spurious-trip retry (added in 1f85106) couldn't tell a
genuine Galil safety-thread trip apart from a user/scan-initiated stop()
(e.g. Ctrl+C): an aborted move looked like an incomplete move either way,
so the abort got silently undone by resetting the error latches and
re-issuing the original move to the same target.

Track stop() requests on a per-motor threading.Event, cleared at the start
of each move() and checked (twice, since reset_axis_errors() itself takes
a few round-trips) before the retry branch fires, so an aborted move is
now reported as a failure instead of being retried.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
This commit was merged in pull request #281.
This commit is contained in:
x01dc
2026-07-28 11:37:23 +02:00
co-authored by Claude Sonnet 5
parent c42595922d
commit 89c5ed1c9e
2 changed files with 53 additions and 7 deletions
+16 -7
View File
@@ -254,6 +254,7 @@ class LamniGalilMotor(Device, PositionerBase):
self.tolerance = kwargs.pop("tolerance", 0.5)
self.device_mapping = kwargs.pop("device_mapping", {})
self.device_manager = device_manager
self._stop_requested = threading.Event()
if len(self.device_mapping) > 0 and self.device_manager is None:
raise BECConfigError(
@@ -356,6 +357,7 @@ class LamniGalilMotor(Device, PositionerBase):
If motion fails other than timing out
"""
self._started_moving = False
self._stop_requested.clear()
timeout = kwargs.pop("timeout", 100)
status = super().move(position, timeout=timeout, **kwargs)
@@ -387,21 +389,27 @@ class LamniGalilMotor(Device, PositionerBase):
def move_and_finish():
success = self._wait_for_move_done(position)
if not success and is_lsamrot:
if not success and is_lsamrot and not self._stop_requested.is_set():
logger.warning(
f"{self.name}: move to {position} did not reach target; resetting "
"Galil error latches and retrying once."
)
self.controller.reset_axis_errors()
try:
self.user_setpoint.put(position, wait=False)
except GalilError as exc:
logger.error(
f"{self.name}: retry after resetting Galil error latches failed: {exc}"
if self._stop_requested.is_set():
logger.info(
f"{self.name}: stop requested during error-latch reset; skipping retry."
)
success = False
else:
success = self._wait_for_move_done(position)
try:
self.user_setpoint.put(position, wait=False)
except GalilError as exc:
logger.error(
f"{self.name}: retry after resetting Galil error latches failed: {exc}"
)
success = False
else:
success = self._wait_for_move_done(position)
if not success:
print(" stop")
@@ -486,5 +494,6 @@ class LamniGalilMotor(Device, PositionerBase):
self._run_subs(sub_type=self.SUB_READBACK, value=val, timestamp=time.time())
def stop(self, *, success=False):
self._stop_requested.set()
self.controller.stop_all_axes()
return super().stop(success=success)
@@ -8,6 +8,9 @@ setting `allaxrer`/`caperr`; the known manual recovery is to send
that scenario against the simulated LAMNI Galil controller.
"""
import threading
import time
import pytest
from csaxs_bec.devices.omny.galil.galil_ophyd import GalilError
@@ -107,6 +110,40 @@ def test_lsamrot_persistent_fault_raises_after_exactly_one_retry(lsamrot, monkey
assert len(reset_calls) == 1
def test_lsamrot_stop_mid_move_is_not_auto_retried(dm_with_devices, monkeypatch):
"""A user/scan-initiated stop() (e.g. Ctrl+C) must not be mistaken for a
spurious safety-thread trip: the aborted move must not be silently
re-issued to its original target.
"""
reset_calls = []
monkeypatch.setattr(
LamniGalilController,
"reset_axis_errors",
lambda self: reset_calls.append(1),
)
LamniGalilController._reset_controller()
motor = SimLamniGalilMotor(
"C", name="lsamrot", host=HOST, port=PORT, device_manager=dm_with_devices, sim_velocity=1.0
)
motor.controller.on()
try:
status = motor.move(10.0, wait=False)
# give the move a moment to actually start moving before aborting it
time.sleep(0.2)
motor.stop()
with pytest.raises(Exception):
status.wait(timeout=10)
assert reset_calls == []
assert motor.readback.get() != pytest.approx(10.0, abs=motor.tolerance)
finally:
motor.controller.off()
motor.controller._reset_controller()
def test_non_lsamrot_axis_is_not_auto_recovered(lsamx, monkeypatch):
"""The shared allaxrer/caperr/allaxref latches must only be auto-reset for
lsamrot; other axes on the same controller should fail immediately, same