feat(psi motor): add optional tolerance signal

ophyd_devices/devices/psi_motor.py

@@ -91,6 +91,12 @@ class EpicsMotor(OphydEpicsMotor):
         Signal, value=0, kind=Kind.config, doc="Settle time after move completion in seconds"
     )
     timeout = Cpt(Signal, kind=Kind.config, doc="Timeout for move completion in seconds.")
+    tolerance = Cpt(
+        Signal,
+        value=-1,
+        kind=Kind.config,
+        doc="Maximum final position error in engineering units; negative values disable the check.",
+    )
 
     def __init__(
         self,
@@ -103,6 +109,7 @@ class EpicsMotor(OphydEpicsMotor):
         parent=None,
         **kwargs,
     ):
+        self._move_target = None
         super().__init__(
             prefix=prefix,
             name=name,
@@ -164,6 +171,23 @@ class EpicsMotor(OphydEpicsMotor):
                     severity,
                 )
 
+        tolerance = self.tolerance.get()
+        if success and tolerance >= 0 and self._move_target is not None:
+            try:
+                target = self._move_target
+                actual = self.user_readback.get(use_monitor=False)
+                if not np.isclose(actual, target, atol=tolerance, rtol=0):
+                    success = False
+                    exc = RuntimeError(
+                        f"Motor {self.name} stopped at {actual}, expected {target} "
+                        f"(tolerance: {tolerance})"
+                    )
+            except Exception as e:
+                success = False
+                exc = RuntimeError(
+                    f"Motor {self.name} failed to read back position after motion: {e}"
+                )
+
         return success, exc
 
     def _move_changed(self, timestamp=None, value=None, sub_type=None, **kwargs):
@@ -188,6 +212,7 @@ class EpicsMotor(OphydEpicsMotor):
 
     def _done_moving(self, success=True, timestamp=None, value=None, **kwargs):
         """Overload PositionerBase._done_moving to pass kwargs to _SUB_REQ_DONE callbacks."""
+        self._move_target = None
         if success:
             self._run_subs(sub_type=self.SUB_DONE, timestamp=timestamp, value=value)
 
@@ -243,6 +268,7 @@ class EpicsMotor(OphydEpicsMotor):
 
         self.subscribe(set_status, event_type=self._SUB_REQ_DONE, run=False)
 
+        self._move_target = position
         self.user_setpoint.put(position, wait=False)
         try:
             if wait:

tests/test_psi_motors.py

@@ -92,6 +92,98 @@ def test_epics_motor_move(mock_epics_motor):
         assert isinstance(status.exception(), RuntimeError)
 
 
+@pytest.mark.parametrize("motor_class", [EpicsMotor, EpicsMotorEC])
+@pytest.mark.parametrize(
+    "tolerance, target, actual, success",
+    [
+        (None, 5, 2, True),  # Default tolerance is disabled.
+        (-0.5, 5, 2, True),
+        (0.25, 5, 4.875, True),
+        (0.25, 5, 4.75, True),  # The boundary is inclusive.
+        (0.25, 5, 4.5, False),
+        (0.25, 5, 5.5, False),
+        (0, 5, 5, True),
+        (0, 5, 4.999, False),
+        (0.25, 100000, 99999.5, False),  # No relative tolerance at large positions.
+    ],
+)
+def test_epics_motor_move_tolerance(motor_class, tolerance, target, actual, success):
+    """DMOV must fail a move outside an enabled absolute position tolerance."""
+    with patched_device(motor_class, name="test_motor", prefix="SIM:MOTOR") as motor:
+        assert motor.tolerance.get() == -1
+        assert motor.tolerance.kind == ophyd.Kind.config
+        assert "test_motor_tolerance" in motor.read_configuration()
+        assert "test_motor_tolerance" not in motor.read()
+        if tolerance is not None:
+            motor.tolerance.put(tolerance)
+        motor.user_setpoint._metadata["lower_ctrl_limit"] = -200000
+        motor.user_setpoint._metadata["upper_ctrl_limit"] = 200000
+        motor.motor_mode.put(SpmgStates.GO)
+        motor.user_readback._read_pv.set_severity(ophyd.utils.epics_pvs.AlarmSeverity.NO_ALARM)
+        motor.user_readback._metadata["severity"] = ophyd.utils.epics_pvs.AlarmSeverity.NO_ALARM
+
+        status = motor.move(target)
+        motor.motor_done_move._read_pv.mock_data = 0
+        assert not status.done
+        motor.user_readback._read_pv.mock_data = actual
+        assert not status.done
+        motor.motor_done_move._read_pv.mock_data = 1
+
+        if success:
+            status.wait(timeout=5)
+        else:
+            with pytest.raises(RuntimeError, match="stopped at .*expected .*tolerance:"):
+                status.wait(timeout=5)
+        assert status.done
+        assert status.success is success
+
+
+@pytest.mark.parametrize("motor_class", [EpicsMotor, EpicsMotorEC])
+def test_epics_motor_tolerance_uses_requested_target(motor_class):
+    """An IOC setpoint reset must not hide a move that stopped short of its target."""
+    with patched_device(motor_class, name="test_motor", prefix="SIM:MOTOR") as motor:
+        motor.tolerance.put(0.01)
+        motor.motor_mode.put(SpmgStates.GO)
+        motor.user_setpoint._metadata["lower_ctrl_limit"] = -10
+        motor.user_setpoint._metadata["upper_ctrl_limit"] = 10
+        motor.user_readback._read_pv.set_severity(ophyd.utils.epics_pvs.AlarmSeverity.NO_ALARM)
+        motor.user_readback._metadata["severity"] = ophyd.utils.epics_pvs.AlarmSeverity.NO_ALARM
+
+        status = motor.move(5)
+        motor.motor_done_move._read_pv.mock_data = 0
+        motor.user_readback._read_pv.mock_data = 2
+        # The motor record synchronizes VAL to RBV after an external stop.
+        motor.user_setpoint._read_pv.mock_data = 2
+        motor.motor_done_move._read_pv.mock_data = 1
+
+        with pytest.raises(RuntimeError, match="stopped at 2, expected 5"):
+            status.wait(timeout=5)
+        assert not status.success
+
+        # A subsequent move must use its own target, not the previous one.
+        status = motor.move(6)
+        motor.motor_done_move._read_pv.mock_data = 0
+        motor.user_readback._read_pv.mock_data = 6
+        motor.motor_done_move._read_pv.mock_data = 1
+        status.wait(timeout=5)
+        assert status.success
+
+
+def test_epics_motor_tolerance_preserves_limit_failure(mock_epics_motor):
+    """An existing motion error takes precedence over the position tolerance failure."""
+    motor = mock_epics_motor
+    motor.tolerance.put(0)
+    motor.user_readback._read_pv.set_severity(ophyd.utils.epics_pvs.AlarmSeverity.NO_ALARM)
+    motor.user_readback._metadata["severity"] = ophyd.utils.epics_pvs.AlarmSeverity.NO_ALARM
+    motor.user_setpoint._read_pv.mock_data = 5
+    motor.low_limit_switch._read_pv.mock_data = 1
+
+    success, exception = motor._check_motion_status()
+
+    assert not success
+    assert "low limit switch" in str(exception)
+
+
 def test_move_epics_motor_ec(mock_epics_motor_ec):
     """Test the move method of EpicsMotorEC."""
     motor_ec = mock_epics_motor_ec