601 convert smargon to ophyd async

src/dodal/beamlines/i23.py

@@ -1,6 +1,5 @@
 from dodal.common.beamlines.beamline_utils import device_instantiation
 from dodal.common.beamlines.beamline_utils import set_beamline as set_utils_beamline
-from dodal.devices.i23.gonio import Gonio
 from dodal.devices.oav.pin_image_recognition import PinTipDetection
 from dodal.log import set_beamline as set_log_beamline
 from dodal.utils import get_beamline_name, get_hostname, skip_device
@@ -19,17 +18,6 @@ def _is_i23_machine():
     return hostname.startswith("i23-ws") or hostname.startswith("i23-control")
 
 
-def gonio(wait_for_connection: bool = True, fake_with_ophyd_sim: bool = False) -> Gonio:
-    """Get the i23 goniometer device"""
-    return device_instantiation(
-        Gonio,
-        "Gonio",
-        "-MO-GONIO-01:",
-        wait_for_connection,
-        fake_with_ophyd_sim,
-    )
-
-
 @skip_device(lambda: not _is_i23_machine())
 def oav_pin_tip_detection(
     wait_for_connection: bool = True, fake_with_ophyd_sim: bool = False

src/dodal/devices/fast_grid_scan.py

@@ -23,7 +23,6 @@
 from pydantic import validator
 from pydantic.dataclasses import dataclass
 
-from dodal.devices.motors import XYZLimitBundle
 from dodal.log import LOGGER
 from dodal.parameters.experiment_parameter_base import AbstractExperimentWithBeamParams
 
@@ -112,35 +111,6 @@ def _get_y_axis(cls, y_axis: GridAxis, values: dict[str, Any]) -> GridAxis:
     def _get_z_axis(cls, z_axis: GridAxis, values: dict[str, Any]) -> GridAxis:
         return GridAxis(values["z2_start"], values["z_step_size"], values["z_steps"])
 
-    def is_valid(self, limits: XYZLimitBundle) -> bool:
-        """
-        Validates scan parameters
-
-        :param limits: The motor limits against which to validate
-                       the parameters
-        :return: True if the scan is valid
-        """
-        x_in_limits = limits.x.is_within(self.x_axis.start) and limits.x.is_within(
-            self.x_axis.end
-        )
-        y_in_limits = limits.y.is_within(self.y_axis.start) and limits.y.is_within(
-            self.y_axis.end
-        )
-
-        first_grid_in_limits = (
-            x_in_limits and y_in_limits and limits.z.is_within(self.z1_start)
-        )
-
-        z_in_limits = limits.z.is_within(self.z_axis.start) and limits.z.is_within(
-            self.z_axis.end
-        )
-
-        second_grid_in_limits = (
-            x_in_limits and z_in_limits and limits.y.is_within(self.y2_start)
-        )
-
-        return first_grid_in_limits and second_grid_in_limits
-
     def get_num_images(self):
         return self.x_steps * self.y_steps + self.x_steps * self.z_steps
 

src/dodal/devices/motors.py

@@ -1,8 +1,3 @@
-from dataclasses import dataclass
-from typing import List, Tuple
-
-import numpy as np
-from ophyd import EpicsMotor
 from ophyd_async.core import Device
 from ophyd_async.epics.motion import Motor
 
@@ -13,46 +8,3 @@ def __init__(self, prefix: str, name: str):
         self.y = Motor(prefix + "Y")
         self.z = Motor(prefix + "Z")
         super().__init__(name)
-
-
-@dataclass
-class MotorLimitHelper:
-    """
-    Represents motor limit(s)
-    """
-
-    motor: EpicsMotor
-
-    def is_within(self, position: float) -> bool:
-        """Checks position against limits
-
-        :param position: The position to check
-        :return: True if position is within the limits
-        """
-        low = float(self.motor.low_limit_travel.get())
-        high = float(self.motor.high_limit_travel.get())
-        return low <= position <= high
-
-
-@dataclass
-class XYZLimitBundle:
-    """
-    Holder for limits reflecting an x, y, z bundle
-    """
-
-    x: MotorLimitHelper
-    y: MotorLimitHelper
-    z: MotorLimitHelper
-
-    def position_valid(
-        self, position: np.ndarray | List[float] | Tuple[float, float, float]
-    ):
-        if len(position) != 3:
-            raise ValueError(
-                f"Position valid expects a 3-vector, got {position} instead"
-            )
-        return (
-            self.x.is_within(position[0])
-            & self.y.is_within(position[1])
-            & self.z.is_within(position[2])
-        )

src/dodal/devices/smargon.py

@@ -1,21 +1,30 @@
+from collections.abc import Generator
+from dataclasses import dataclass
 from enum import Enum
+from math import isclose
+from typing import Collection, cast
 
-from ophyd import Component as Cpt
-from ophyd import Device, EpicsMotor, EpicsSignal
-from ophyd.epics_motor import MotorBundle
-from ophyd.status import StatusBase
+from bluesky import plan_stubs as bps
+from bluesky.utils import Msg
+from ophyd_async.core import AsyncStatus, Device, StandardReadable, wait_for_value
+from ophyd_async.epics.motion import Motor
+from ophyd_async.epics.signal import epics_signal_r
 
-from dodal.devices.motors import MotorLimitHelper, XYZLimitBundle
-from dodal.devices.status import await_approx_value
 from dodal.devices.util.epics_util import SetWhenEnabled
-from dodal.devices.util.motor_utils import ExtendedEpicsMotor
 
 
 class StubPosition(Enum):
     CURRENT_AS_CENTER = 0
     RESET_TO_ROBOT_LOAD = 1
 
 
+def approx_equal_to(target, deadband: float = 1e-9):
+    def approx_equal_to_target(value):
+        return isclose(target, value, rel_tol=0, abs_tol=deadband)
+
+    return approx_equal_to_target
+
+
 class StubOffsets(Device):
     """Stub offsets are used to change the internal co-ordinate system of the smargon by
     adding an offset to x, y, z.
@@ -25,59 +34,104 @@ class StubOffsets(Device):
     set them so that the current position is zero or to pre-defined positions.
     """
 
-    parent: "Smargon"
+    def __init__(self, name: str = "", prefix: str = ""):
+        self.center_at_current_position = SetWhenEnabled(prefix=prefix + "CENTER_CS")
+        self.to_robot_load = SetWhenEnabled(prefix=prefix + "SET_STUBS_TO_RL")
+        super().__init__(name)
 
-    center_at_current_position = Cpt(SetWhenEnabled, "CENTER_CS")
-    to_robot_load = Cpt(SetWhenEnabled, "SET_STUBS_TO_RL")
-
-    def set(self, pos: StubPosition) -> StatusBase:
+    @AsyncStatus.wrap
+    async def set(self, pos: StubPosition):
         if pos == StubPosition.CURRENT_AS_CENTER:
-            status = self.center_at_current_position.set(1)
-            status &= await_approx_value(self.parent.x, 0.0, deadband=0.1)
-            status &= await_approx_value(self.parent.y, 0.0, deadband=0.1)
-            status &= await_approx_value(self.parent.z, 0.0, deadband=0.1)
-            return status
+            await self.center_at_current_position.set(1)
+            smargon = cast(Smargon, self.parent)
+            await wait_for_value(
+                smargon.x.user_readback, approx_equal_to(0.0, 0.1), None
+            )
+            await wait_for_value(
+                smargon.y.user_readback, approx_equal_to(0.0, 0.1), None
+            )
+            await wait_for_value(
+                smargon.z.user_readback, approx_equal_to(0.0, 0.1), None
+            )
         else:
-            return self.to_robot_load.set(1)
+            await self.to_robot_load.set(1)
+
+
+@dataclass
+class AxisLimit:
+    """Represents the minimum and maximum allowable values on an axis"""
+
+    min_value: float
+    max_value: float
+
+    def contains(self, pos: float):
+        """Determine if the specified value is within limits.
+
+        Args:
+            pos: the value to check
+
+        Returns:
+            True if the value does not exceed the limits
+        """
+        return self.min_value <= pos <= self.max_value
 
 
-class Smargon(MotorBundle):
+@dataclass
+class XYZLimits:
+    """The limits of the smargon x, y, z axes."""
+
+    x: AxisLimit
+    y: AxisLimit
+    z: AxisLimit
+
+    def position_valid(self, pos: Collection[float]) -> bool:
+        return all(
+            axis_limits.contains(value)
+            for axis_limits, value in zip([self.x, self.y, self.z], pos)
+        )
+
+
+class Smargon(StandardReadable):
     """
     Real motors added to allow stops following pin load (e.g. real_x1.stop() )
     X1 and X2 real motors provide compound chi motion as well as the compound X travel,
     increasing the gap between x1 and x2 changes chi, moving together changes virtual x.
     Robot loading can nudge these and lead to errors.
     """
 
-    x = Cpt(ExtendedEpicsMotor, "X")
-    y = Cpt(EpicsMotor, "Y")
-    z = Cpt(EpicsMotor, "Z")
-    chi = Cpt(EpicsMotor, "CHI")
-    phi = Cpt(EpicsMotor, "PHI")
-    omega = Cpt(ExtendedEpicsMotor, "OMEGA")
-
-    real_x1 = Cpt(EpicsMotor, "MOTOR_3")
-    real_x2 = Cpt(EpicsMotor, "MOTOR_4")
-    real_y = Cpt(EpicsMotor, "MOTOR_1")
-    real_z = Cpt(EpicsMotor, "MOTOR_2")
-    real_phi = Cpt(EpicsMotor, "MOTOR_5")
-    real_chi = Cpt(EpicsMotor, "MOTOR_6")
-
-    stub_offsets = Cpt(StubOffsets, "")
-
-    disabled = Cpt(EpicsSignal, "DISABLED")
-
-    def get_xyz_limits(self) -> XYZLimitBundle:
-        """Get the limits for the x, y and z axes.
-
-        Note that these limits may not yet be valid until wait_for_connection is called
-        on this MotorBundle.
+    def __init__(self, prefix: str = "", name: str = ""):
+        with self.add_children_as_readables():
+            self.x = Motor(prefix + "X")
+            self.y = Motor(prefix + "Y")
+            self.z = Motor(prefix + "Z")
+            self.chi = Motor(prefix + "CHI")
+            self.phi = Motor(prefix + "PHI")
+            self.omega = Motor(prefix + "OMEGA")
+            self.real_x1 = Motor(prefix + "MOTOR_3")
+            self.real_x2 = Motor(prefix + "MOTOR_4")
+            self.real_y = Motor(prefix + "MOTOR_1")
+            self.real_z = Motor(prefix + "MOTOR_2")
+            self.real_phi = Motor(prefix + "MOTOR_5")
+            self.real_chi = Motor(prefix + "MOTOR_6")
+            self.stub_offsets = StubOffsets(prefix=prefix)
+            self.disabled = epics_signal_r(int, prefix + "DISABLED")
+
+        super().__init__(name)
+
+    def get_xyz_limits(self) -> Generator[Msg, None, XYZLimits]:
+        """Obtain a plan stub that returns the smargon XYZ axis limits
+
+        Yields:
+            Bluesky messages
 
         Returns:
-            XYZLimitBundle: The limits for the underlying motors.
+            the axis limits
         """
-        return XYZLimitBundle(
-            MotorLimitHelper(self.x),
-            MotorLimitHelper(self.y),
-            MotorLimitHelper(self.z),
-        )
+        limits = {}
+        for name, pv in [
+            (attr_name, getattr(self, attr_name)) for attr_name in ["x", "y", "z"]
+        ]:
+            min_value = yield from bps.rd(pv.low_limit_travel)
+            max_value = yield from bps.rd(pv.high_limit_travel)
+            limits[name] = AxisLimit(min_value, max_value)
+        return XYZLimits(**limits)

src/dodal/devices/util/epics_util.py

@@ -1,10 +1,14 @@
 from functools import partial
 from typing import Callable
 
-from ophyd import Component, Device, EpicsSignal
+from bluesky.protocols import Movable
+from ophyd import Component, EpicsSignal
+from ophyd import Device as OphydDevice
 from ophyd.status import Status, StatusBase
+from ophyd_async.core import AsyncStatus, wait_for_value
+from ophyd_async.core import Device as OphydAsyncDevice
+from ophyd_async.epics.signal import epics_signal_r, epics_signal_rw
 
-from dodal.devices.status import await_value
 from dodal.log import LOGGER
 
 
@@ -24,7 +28,7 @@ def epics_signal_put_wait(pv_name: str, wait: float = 3.0) -> Component[EpicsSig
 def run_functions_without_blocking(
     functions_to_chain: list[Callable[[], StatusBase]],
     timeout: float = 60.0,
-    associated_obj: Device | None = None,
+    associated_obj: OphydDevice | None = None,
 ) -> Status:
     """Creates and initiates an asynchronous chaining of functions which return a status
 
@@ -112,16 +116,15 @@ def call_func(func: Callable[[], StatusBase]) -> StatusBase:
     return func()
 
 
-class SetWhenEnabled(Device):
+class SetWhenEnabled(OphydAsyncDevice, Movable):
     """A device that sets the proc field of a PV when it becomes enabled."""
 
-    proc = Component(EpicsSignal, ".PROC")
-    disp = Component(EpicsSignal, ".DISP")
+    def __init__(self, name: str = "", prefix: str = ""):
+        self.proc = epics_signal_rw(int, prefix + ".PROC")
+        self.disp = epics_signal_r(int, prefix + ".DISP")
+        super().__init__(name)
 
-    def set(self, proc: int) -> Status:
-        return run_functions_without_blocking(
-            [
-                lambda: await_value(self.disp, 0),
-                lambda: self.proc.set(proc),
-            ]
-        )
+    @AsyncStatus.wrap
+    async def set(self, value: int):
+        await wait_for_value(self.disp, 0, None)
+        await self.proc.set(value)