diff --git a/ophyd_devices/interfaces/base_classes/psi_positioner_base.py b/ophyd_devices/interfaces/base_classes/psi_positioner_base.py index bbf6ac4..f9571e0 100644 --- a/ophyd_devices/interfaces/base_classes/psi_positioner_base.py +++ b/ophyd_devices/interfaces/base_classes/psi_positioner_base.py @@ -8,7 +8,9 @@ from ophyd import Kind from ophyd.device import required_for_connection from ophyd.positioner import PositionerBase from ophyd.signal import EpicsSignalBase, Signal +from ophyd.status import StatusTimeoutError, WaitTimeoutError from ophyd.status import wait as status_wait +from ophyd.utils import UnknownStatusFailure from ophyd.utils.epics_pvs import AlarmSeverity, fmt_time from ophyd_devices.interfaces.base_classes.psi_device_base import PSIDeviceBase @@ -23,28 +25,65 @@ class MoveStatusWithTolerance(MoveStatus): """A MoveStatus that checks the final position against a tolerance if provided.""" def __init__(self, positioner, *args, **kwargs): - self._tolerance_exception = None + self._completion_exception = None super().__init__(positioner, *args, **kwargs) self.positioner = positioner def _finished(self, success: bool = True, **kwargs): - - if not success or self.positioner.user_readback is _OPTIONAL_SIGNAL: + if not success: + exc = UnknownStatusFailure( + f"The status {self!r} has failed. To obtain more specific, " + "helpful errors in the future, update the Device to use " + "set_exception(...) instead of _finished(success=False)." + ) + return self._fail_move(exc) + if self.positioner.user_readback is _OPTIONAL_SIGNAL: return super()._finished(success=success, **kwargs) - tol = self.positioner.tolerance.get() - final_position = self.positioner.user_readback.get() - if abs(self.target - final_position) > tol: + try: + tol = self.positioner.tolerance.get() + skip_check = bool(tol == np.inf) + if not skip_check: + final_position = self.positioner.user_readback.get() + outside_tolerance = bool(abs(self.target - final_position) > tol) + except Exception as exc: + if isinstance(exc, (StatusTimeoutError, WaitTimeoutError)): + wrapped = RuntimeError(f"Final position check failed: {exc}") + wrapped.__cause__ = exc + exc = wrapped + return self._fail_move(exc, motion_completed=True) + + if skip_check: + return super()._finished(success=success, **kwargs) + if outside_tolerance: exc = RuntimeError( f"Move to {self.target} failed, " f"final position {final_position} outside of tolerance {tol}" ) - self._tolerance_exception = exc - return self.set_exception(exc) + return self._fail_move(exc, motion_completed=True) return super()._finished(success=success, **kwargs) + def _fail_move(self, exc: Exception, *, motion_completed: bool = False): + # The custom MoveStatus timeout is not a subclass of ophyd's + # StatusTimeoutError, so ophyd's set_exception() can overwrite it. + # Use the same lock as the timeout thread to keep the first failure. + with self._externally_initiated_completion_lock: + if self._externally_initiated_completion or self._exception is not None: + return + self._externally_initiated_completion = True + self._exception = exc + if motion_completed: + self._completion_exception = exc + self._trace_attributes["exception"] = exc + self._settled_event.set() + + self._close_trace() + if self._callback_thread is None: + self._run_callbacks() + def _handle_failure(self): - if self._tolerance_exception is not None and self._exception is self._tolerance_exception: + # Motion has already completed, so a stop here could issue another move. + if self._completion_exception is not None and self._exception is self._completion_exception: return super()._handle_failure() diff --git a/tests/test_psi_positioner.py b/tests/test_psi_positioner.py index e1a2eb9..656e611 100644 --- a/tests/test_psi_positioner.py +++ b/tests/test_psi_positioner.py @@ -6,6 +6,7 @@ import pytest from ophyd.device import Component as Cpt from ophyd.signal import EpicsSignal, Kind, Signal from ophyd.sim import FakeEpicsSignal, FakeEpicsSignalRO +from ophyd.utils import StatusTimeoutError from ophyd_devices.devices.simple_positioner import PSISimplePositioner from ophyd_devices.interfaces.base_classes.psi_positioner_base import ( @@ -14,6 +15,7 @@ from ophyd_devices.interfaces.base_classes.psi_positioner_base import ( RequiredSignalNotSpecified, ) from ophyd_devices.tests.utils import MockPV, patch_dual_pvs +from ophyd_devices.utils.psi_device_base_utils import StatusTimeoutErrorWithErrorInfo def test_cannot_isntantiate_without_required_signals(): @@ -289,6 +291,129 @@ def test_tolerance_checks_requested_target_even_if_setpoint_changes( st.wait(timeout=1) +@pytest.mark.parametrize("signal_name", ["tolerance", "user_readback"]) +def test_completion_read_error_fails_without_stopping_or_rewriting_setpoint( + mock_done_signal_positioner_with_tolerance, signal_name +): + dev = mock_done_signal_positioner_with_tolerance + st = dev.move(5, wait=False) + dev.motor_done_move.sim_put(0) + callbacks_finished = threading.Event() + st.add_callback(lambda _: callbacks_finished.set()) + read_error = RuntimeError(f"{signal_name} read failed") + + with ( + patch.object(getattr(dev, signal_name), "get", side_effect=read_error), + patch.object(dev, "stop", wraps=dev.stop) as stop, + patch.object(dev.user_setpoint, "put", wraps=dev.user_setpoint.put) as put, + ): + dev.motor_done_move.sim_put(1) + assert callbacks_finished.wait(timeout=1) + with pytest.raises(RuntimeError) as raised: + st.wait(timeout=1) + assert raised.value is read_error + stop.assert_not_called() + put.assert_not_called() + + +def test_completion_read_status_timeout_is_wrapped_and_finishes( + mock_done_signal_positioner_with_tolerance, +): + dev = mock_done_signal_positioner_with_tolerance + st = dev.move(5, wait=False) + dev.motor_done_move.sim_put(0) + callbacks_finished = threading.Event() + st.add_callback(lambda _: callbacks_finished.set()) + read_error = StatusTimeoutError("final read timed out") + + with patch.object(dev.tolerance, "get", side_effect=read_error): + dev.motor_done_move.sim_put(1) + assert callbacks_finished.wait(timeout=1) + + with pytest.raises(RuntimeError) as raised: + st.wait(timeout=1) + assert raised.value.__cause__ is read_error + + +def test_late_completion_read_error_preserves_move_timeout_and_stop( + mock_done_signal_positioner_with_tolerance, +): + dev = mock_done_signal_positioner_with_tolerance + st = dev.move(5, wait=False, timeout=0.5) + dev.motor_done_move.sim_put(0) + entered_read = threading.Event() + release_read = threading.Event() + callbacks_finished = threading.Event() + st.add_callback(lambda _: callbacks_finished.set()) + read_error = RuntimeError("late tolerance read failure") + + def blocked_read(): + entered_read.set() + assert release_read.wait(timeout=2) + raise read_error + + with ( + patch.object(dev.tolerance, "get", side_effect=blocked_read), + patch.object(dev, "stop", wraps=dev.stop) as stop, + ): + completion_thread = threading.Thread(target=dev.motor_done_move.sim_put, args=(1,)) + completion_thread.start() + try: + assert entered_read.wait(timeout=1) + assert not st.done + assert callbacks_finished.wait(timeout=2) + with pytest.raises(StatusTimeoutErrorWithErrorInfo) as timed_out: + st.wait(timeout=1) + stop.assert_called_once() + finally: + release_read.set() + completion_thread.join(timeout=1) + + assert not completion_thread.is_alive() + with pytest.raises(StatusTimeoutErrorWithErrorInfo) as after_read: + st.wait(timeout=1) + assert after_read.value is timed_out.value + stop.assert_called_once() + + +def test_move_timeout_remains_structured_after_stop_cancels_done_status( + mock_done_signal_positioner_with_tolerance, +): + dev = mock_done_signal_positioner_with_tolerance + st = dev.move(5, wait=False, timeout=0.2) + dev.motor_done_move.sim_put(0) + callbacks_finished = threading.Event() + st.add_callback(lambda _: callbacks_finished.set()) + + with ( + patch.object( + dev.tolerance, "get", side_effect=AssertionError("unexpected final read") + ) as get, + patch.object(dev, "stop", wraps=dev.stop) as stop, + patch.object(dev, "_done_moving", wraps=dev._done_moving) as done_moving, + ): + assert callbacks_finished.wait(timeout=2) + with pytest.raises(StatusTimeoutErrorWithErrorInfo): + st.wait(timeout=1) + stop.assert_called_once() + done_moving.assert_any_call(success=False) + get.assert_not_called() + + +def test_infinite_tolerance_completion_does_not_read_final_position(mock_psi_positioner): + dev = mock_psi_positioner + assert dev.tolerance.get() == float("inf") + st = dev.move(5, wait=False) + dev.motor_done_move._read_pv.mock_data = 0 + + with patch.object(dev.user_readback, "get", side_effect=RuntimeError("unexpected read")) as get: + dev.motor_done_move._read_pv.mock_data = 1 + st.wait(timeout=1) + get.assert_not_called() + + assert st.success + + def test_move_rejects_tolerance_tighter_than_deadband_before_setpoint_write( mock_done_signal_positioner_with_tolerance, ):