diff --git a/src/sophys/common/devices/motor.py b/src/sophys/common/devices/motor.py index bebde43..49185fa 100644 --- a/src/sophys/common/devices/motor.py +++ b/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)