feat(motor): add readback tolerance check to EpicsMotor.move()

src/sophys/common/devices/motor.py

@@ -1,44 +1,146 @@
-from ophyd import Device, EpicsMotor, Component, EpicsSignal, FormattedComponent
+from ophyd import (  # numpydoc ignore=GL08
+    Component,
+    Device,
+    EpicsMotor,
+    EpicsSignal,
+    FormattedComponent,
+)
 from ophyd.device import create_device_from_components
+from ophyd.status import AndStatus, SubscriptionStatus
+from ophyd.status import wait as status_wait
 
 
-class MotorMixinResolution(Device):
+class ReadbackEpicsMotor(EpicsMotor):
+    """
+    An EpicsMotor subclass that includes readback tolerance checking.
+
+    This motor extends the standard `EpicsMotor` to ensure that a move operation
+    does not complete until both the standard motion status (`dmov`) and the
+    actual readback value (`user_readback`) are within a specified `tolerance`
+    from the target position.
+
+    Parameters
+    ----------
+    prefix : str, optional
+        The EPICS PV prefix for the motor records.
+    name : str
+        The name of the device (required by Ophyd).
+    tolerance : float, optional
+        The acceptable absolute difference between the target position and
+        the `user_readback` value to consider the move successful.
+        Defaults to 0.001.
+    kind : Kind, optional
+        Ophyd component kind (e.g., Kind.normal, Kind.hinted).
+    read_attrs : list, optional
+        Attributes to include in regular reads.
+    configuration_attrs : list, optional
+        Attributes to include in configuration reads.
+    parent : Device, optional
+        The parent device instance, if applicable.
+    **kwargs
+        Additional keyword arguments passed to `EpicsMotor`.
+    """
+
+    def __init__(  # numpydoc ignore=GL08
+        self,
+        prefix="",
+        *,
+        name,
+        tolerance=0.001,
+        kind=None,
+        read_attrs=None,
+        configuration_attrs=None,
+        parent=None,
+        **kwargs,
+    ):
+        self.tolerance = tolerance
+        super().__init__(
+            prefix,
+            name=name,
+            kind=kind,
+            read_attrs=read_attrs,
+            configuration_attrs=configuration_attrs,
+            parent=parent,
+            **kwargs,
+        )
+
+    def move(self, position, wait=True, **kwargs):
+        """
+        Move the motor to a target position and wait for readback tolerance.
+
+        Unlike the standard `EpicsMotor.move`, this method blocks or returns
+        a status object that only completes when both the motor motion is done
+        and the actual readback (`user_readback`) is within the defined
+        `tolerance` from the target position.
+
+        Parameters
+        ----------
+        position : float
+            The target position to move the motor to.
+        wait : bool, optional
+            If True, blocks execution until the motion and readback tolerance
+            criteria are met. If False, returns the status object immediately.
+            Defaults to True.
+        **kwargs : dict
+            Additional keyword arguments passed to the parent's move method.
+        """
+        self._started_moving = False
+        dmov_status = super(EpicsMotor, self).move(position, **kwargs)
+        self.user_setpoint.put(position, wait=False)
+
+        def check_readback(*args, value, **kwargs):  # numpydoc ignore=GL08
+            return abs(value - position) <= self.tolerance
+
+        rbv_status = SubscriptionStatus(self.user_readback, check_readback)
+        combined_status = AndStatus(dmov_status, rbv_status)
+
+        try:
+            if wait:
+                status_wait(combined_status)
+        except KeyboardInterrupt:
+            self.stop()
+            raise
+
+        return combined_status
+
+
+class MotorMixinResolution(Device):  # numpydoc ignore=GL08
     motor_step_size = Component(EpicsSignal, ".MRES", kind="config", auto_monitor=True)
 
     steps_per_revolution = Component(EpicsSignal, ".SREV", kind="omitted")
     units_per_revolution = Component(EpicsSignal, ".UREV", kind="omitted")
 
 
-class MotorMixinMiscellaneous(Device):
+class MotorMixinMiscellaneous(Device):  # numpydoc ignore=GL08
     display_precision = Component(
         EpicsSignal, ".PREC", kind="config", auto_monitor=True
     )
     code_version = Component(EpicsSignal, ".VERS", kind="config")
 
 
-class MotorMixinMotion(Device):
+class MotorMixinMotion(Device):  # numpydoc ignore=GL08
     max_velocity = Component(EpicsSignal, ".VMAX", kind="config")
     base_velocity = Component(EpicsSignal, ".VBAS", kind="config")
 
 
-class ExtendedEpicsMotor(
-    EpicsMotor, MotorMixinResolution, MotorMixinMiscellaneous, MotorMixinMotion
+class ExtendedEpicsMotor(  # numpydoc ignore=GL08
+    ReadbackEpicsMotor, MotorMixinResolution, MotorMixinMiscellaneous, MotorMixinMotion
 ):
     pass
 
 
-class ControllableMotor(EpicsMotor):
+class ControllableMotor(ReadbackEpicsMotor):  # numpydoc ignore=PR01
     """Custom EpicsMotor that enables control before a plan and disables it after."""
 
     enable_control = Component(EpicsSignal, ".CNEN", kind="config", auto_monitor=True)
 
-    def __init__(self, *args, **kwargs):
+    def __init__(self, *args, **kwargs):  # numpydoc ignore=GL08
         super().__init__(*args, **kwargs)
 
         self.stage_sigs["enable_control"] = 1
 
 
-class VirtualControllableMotorBaseClass(EpicsMotor):
+class VirtualControllableMotorBaseClass(ReadbackEpicsMotor):  # numpydoc ignore=GL08
     pass
 
 
@@ -105,7 +207,9 @@ def unstage(self):
     return (VirtualControllableMotorClass, {"attr_keys": formattedComponents.keys()})
 
 
-def MotorGroup(prefix, motors_suffixes, **kwargs):
+def MotorGroup(
+    prefix, motors_suffixes, tolerance=0.001, **kwargs
+):  # numpydoc ignore=PR01,PR02
     """
     Function to instantiate several motor devices.
 
@@ -134,8 +238,12 @@ def MotorGroup(prefix, motors_suffixes, **kwargs):
     ----------
     prefix : str
         The prefix of the motor group.
-    motors_suffixes : dict of (string, string)
+    motors_suffixes : dict of (str, str)
         The real motors that constitute this motor group, in the form of .
+    tolerance : float, optional
+        The acceptable absolute difference between the target position and
+        the `user_readback` value to consider the move successful.
+        Defaults to 0.001.
     name : str
         Name of the created motor group.
     """
@@ -154,4 +262,4 @@ def MotorGroup(prefix, motors_suffixes, **kwargs):
 
     devClass = create_device_from_components(name="motor_group", **components)
 
-    return devClass(prefix=prefix, **kwargs)
+    return devClass(prefix=prefix, tolerance=tolerance, **kwargs)