Port UndulatorDCM to ophyd-async

src/dodal/beamlines/i03.py

@@ -6,7 +6,7 @@
 from dodal.devices.aperturescatterguard import AperturePositions, ApertureScatterguard
 from dodal.devices.attenuator import Attenuator
 from dodal.devices.backlight import Backlight
-from dodal.devices.DCM import DCM
+from dodal.devices.dcm import DCM
 from dodal.devices.detector import DetectorParams
 from dodal.devices.detector.detector_motion import DetectorMotion
 from dodal.devices.eiger import EigerDetector
@@ -50,10 +50,9 @@ def dcm(wait_for_connection: bool = True, fake_with_ophyd_sim: bool = False) ->
     return device_instantiation(
         DCM,
         "dcm",
-        "",
+        "-MO-DCM-01:",
         wait_for_connection,
         fake_with_ophyd_sim,
-        daq_configuration_path=DAQ_CONFIGURATION_PATH,
     )
 
 
@@ -321,6 +320,8 @@ def undulator_dcm(
         fake=fake_with_ophyd_sim,
         undulator=undulator(wait_for_connection, fake_with_ophyd_sim),
         dcm=dcm(wait_for_connection, fake_with_ophyd_sim),
+        daq_configuration_path=DAQ_CONFIGURATION_PATH,
+        id_gap_lookup_table_path="/dls_sw/i03/software/daq_configuration/lookup/BeamLine_Undulator_toGap.txt",
     )
 
 

src/dodal/beamlines/i04.py

@@ -4,7 +4,7 @@
 from dodal.devices.attenuator import Attenuator
 from dodal.devices.backlight import Backlight
 from dodal.devices.beamstop import BeamStop
-from dodal.devices.DCM import DCM
+from dodal.devices.dcm import DCM
 from dodal.devices.detector import DetectorParams
 from dodal.devices.detector.detector_motion import DetectorMotion
 from dodal.devices.eiger import EigerDetector
@@ -188,10 +188,9 @@ def dcm(wait_for_connection: bool = True, fake_with_ophyd_sim: bool = False) ->
     return device_instantiation(
         DCM,
         "dcm",
-        "",
+        "-MO-DCM-01:",
         wait_for_connection,
         fake_with_ophyd_sim,
-        daq_configuration_path=DAQ_CONFIGURATION_PATH,
     )
 
 

src/dodal/devices/dcm.py

@@ -0,0 +1,39 @@
+from ophyd_async.core import StandardReadable
+from ophyd_async.epics.motion import Motor
+from ophyd_async.epics.signal import epics_signal_r
+
+
+class DCM(StandardReadable):
+    """
+    A double crystal monochromator (DCM), used to select the energy of the beam.
+
+    perp describes the gap between the 2 DCM crystals which has to change as you alter
+    the angle to select the requested energy.
+
+    offset ensures that the beam exits the DCM at the same point, regardless of energy.
+    """
+
+    def __init__(
+        self,
+        prefix: str,
+        name: str = "",
+    ) -> None:
+        with self.add_children_as_readables():
+            self.bragg_in_degrees = Motor(prefix + "BRAGG")
+            self.roll_in_mrad = Motor(prefix + "ROLL")
+            self.offset_in_mm = Motor(prefix + "OFFSET")
+            self.perp_in_mm = Motor(prefix + "PERP")
+            self.energy_in_kev = Motor(prefix + "ENERGY")
+            self.pitch_in_mrad = Motor(prefix + "PITCH")
+            self.wavelength = Motor(prefix + "WAVELENGTH")
+
+            # temperatures
+            self.xtal1_temp = epics_signal_r(float, prefix + "TEMP1")
+            self.xtal2_temp = epics_signal_r(float, prefix + "TEMP2")
+            self.xtal1_heater_temp = epics_signal_r(float, prefix + "TEMP3")
+            self.xtal2_heater_temp = epics_signal_r(float, prefix + "TEMP4")
+            self.backplate_temp = epics_signal_r(float, prefix + "TEMP5")
+            self.perp_temp = epics_signal_r(float, prefix + "TEMP6")
+            self.perp_sub_assembly_temp = epics_signal_r(float, prefix + "TEMP7")
+
+        super().__init__(name)

src/dodal/devices/undulator.py

@@ -1,28 +1,34 @@
 from enum import Enum
 
-from ophyd import Component, Device, EpicsMotor, EpicsSignalRO
+from ophyd_async.core import StandardReadable
+from ophyd_async.epics.motion import Motor
+from ophyd_async.epics.signal import epics_signal_r
 
 # The acceptable difference, in mm, between the undulator gap and the DCM
 # energy, when the latter is converted to mm using lookup tables
 UNDULATOR_DISCREPANCY_THRESHOLD_MM = 2e-3
 
 
-class UndulatorGapAccess(Enum):
+class UndulatorGapAccess(str, Enum):
     ENABLED = "ENABLED"
     DISABLED = "DISABLED"
 
 
-class Undulator(Device):
-    gap_motor = Component(EpicsMotor, "BLGAPMTR")
-    current_gap = Component(EpicsSignalRO, "CURRGAPD")
-    gap_access = Component(EpicsSignalRO, "IDBLENA")
-    gap_discrepancy_tolerance_mm: float = UNDULATOR_DISCREPANCY_THRESHOLD_MM
+class Undulator(StandardReadable):
+    """
+    An Undulator-type insertion device, used to control photon emission at a given
+    beam energy.
+    """
 
     def __init__(
         self,
-        lookup_table_path="/dls_sw/i03/software/daq_configuration/lookup/BeamLine_Undulator_toGap.txt",
-        *args,
-        **kwargs,
-    ):
-        super().__init__(*args, **kwargs)
-        self.lookup_table_path = lookup_table_path
+        prefix: str,
+        name: str = "",
+    ) -> None:
+        with self.add_children_as_readables():
+            self.gap_motor = Motor(prefix + "BLGAPMTR")
+            self.current_gap = epics_signal_r(float, prefix + "CURRGAPD")
+            self.gap_access = epics_signal_r(UndulatorGapAccess, prefix + "IDBLENA")
+        self.gap_discrepancy_tolerance_mm: float = UNDULATOR_DISCREPANCY_THRESHOLD_MM
+
+        super().__init__(name)

src/dodal/devices/undulator_dcm.py

@@ -1,14 +1,19 @@
+import asyncio
+
 import numpy as np
-from numpy import argmin, loadtxt, ndarray
-from ophyd import Component, Device, Signal
-from ophyd.status import Status
+from bluesky.protocols import Movable
+from numpy import argmin, ndarray
+from ophyd_async.core import AsyncStatus, StandardReadable
 
-from dodal.devices.DCM import DCM
-from dodal.devices.undulator import Undulator, UndulatorGapAccess
+from dodal.beamlines.beamline_parameters import get_beamline_parameters
 from dodal.log import LOGGER
 
-ENERGY_TIMEOUT_S = 30
-STATUS_TIMEOUT_S = 10
+from .dcm import DCM
+from .undulator import Undulator, UndulatorGapAccess
+from .util.lookup_tables import energy_distance_table
+
+ENERGY_TIMEOUT_S: float = 30.0
+STATUS_TIMEOUT_S: float = 10.0
 
 # Enable to allow testing when the beamline is down, do not change in production!
 TEST_MODE = False
@@ -18,10 +23,6 @@ class AccessError(Exception):
     pass
 
 
-def _get_energy_distance_table(lookup_table_path: str) -> ndarray:
-    return loadtxt(lookup_table_path, comments=["#", "Units"])
-
-
 def _get_closest_gap_for_energy(
     dcm_energy_ev: float, energy_to_distance_table: ndarray
 ) -> float:
@@ -30,58 +31,105 @@ def _get_closest_gap_for_energy(
     return table[1][idx]
 
 
-class UndulatorDCM(Device):
+class UndulatorDCM(StandardReadable, Movable):
     """
-    Composite device to handle changing beamline energies
+    Composite device to handle changing beamline energies, wraps the Undulator and the
+    DCM. The DCM has a motor which controls the beam energy, when it moves, the
+    Undulator gap may also have to change to enable emission at the new energy.
+    The relationship between the two motor motor positions is provided via a lookup
+    table.
+
+    Calling unulator_dcm.set(energy) will move the DCM motor, perform a table lookup
+    and move the Undulator gap motor if needed. So the set method can be thought of as
+    a comprehensive way to set beam energy.
     """
 
-    class EnergySignal(Signal):
-        parent: "UndulatorDCM"
-
-        def set(self, value, *, timeout=None, settle_time=None, **kwargs) -> Status:
-            energy_kev = value
-            access_level = self.parent.undulator.gap_access.get(as_string=True)
-            if access_level == UndulatorGapAccess.DISABLED.value and not TEST_MODE:
-                raise AccessError(
-                    "Undulator gap access is disabled. Contact Control Room"
-                )
-
-            # Get 2d np.array converting energies to undulator gap distance, from lookup table
-            energy_to_distance_table = _get_energy_distance_table(
-                self.parent.undulator.lookup_table_path
-            )
-            LOGGER.info(f"Setting DCM energy to {energy_kev:.2f} kev")
+    def __init__(
+        self,
+        undulator: Undulator,
+        dcm: DCM,
+        id_gap_lookup_table_path: str,
+        daq_configuration_path: str,
+        prefix: str = "",
+        name: str = "",
+    ):
+        super().__init__(name)
+
+        # Attributes are set after super call so they are not renamed to
+        # <name>-undulator, etc.
+        self.undulator = undulator
+        self.dcm = dcm
 
-            status = self.parent.dcm.energy_in_kev.move(
-                energy_kev, timeout=ENERGY_TIMEOUT_S
+        # These attributes are just used by hyperion for lookup purposes
+        self.id_gap_lookup_table_path = id_gap_lookup_table_path
+        self.dcm_pitch_converter_lookup_table_path = (
+            daq_configuration_path + "/lookup/BeamLineEnergy_DCM_Pitch_converter.txt"
+        )
+        self.dcm_roll_converter_lookup_table_path = (
+            daq_configuration_path + "/lookup/BeamLineEnergy_DCM_Roll_converter.txt"
+        )
+        # I03 configures the DCM Perp as a side effect of applying this fixed value to the DCM Offset after an energy change
+        # Nb this parameter is misleadingly named to confuse you
+        self.dcm_fixed_offset_mm = get_beamline_parameters(
+            daq_configuration_path + "/domain/beamlineParameters"
+        )["DCM_Perp_Offset_FIXED"]
+
+    def set(self, value: float) -> AsyncStatus:
+        async def _set():
+            await asyncio.gather(
+                self._set_dcm_energy(value),
+                self._set_undulator_gap_if_required(value),
             )
 
-            # Use the lookup table to get the undulator gap associated with this dcm energy
-            gap_to_match_dcm_energy = _get_closest_gap_for_energy(
-                energy_kev * 1000, energy_to_distance_table
+        return AsyncStatus(_set())
+
+    async def _set_dcm_energy(self, energy_kev: float) -> None:
+        access_level = await self.undulator.gap_access.get_value()
+        if access_level is UndulatorGapAccess.DISABLED and not TEST_MODE:
+            raise AccessError("Undulator gap access is disabled. Contact Control Room")
+
+        await self.dcm.energy_in_kev.set(
+            energy_kev,
+            timeout=ENERGY_TIMEOUT_S,
+        )
+
+    async def _set_undulator_gap_if_required(self, energy_kev: float) -> None:
+        LOGGER.info(f"Setting DCM energy to {energy_kev:.2f} kev")
+        gap_to_match_dcm_energy = await self._gap_to_match_dcm_energy(energy_kev)
+
+        # Check if undulator gap is close enough to the value from the DCM
+        current_gap = await self.undulator.current_gap.get_value()
+        if (
+            abs(gap_to_match_dcm_energy - current_gap)
+            > self.undulator.gap_discrepancy_tolerance_mm
+        ):
+            LOGGER.info(
+                f"Undulator gap mismatch. {abs(gap_to_match_dcm_energy-current_gap):.3f}mm is outside tolerance.\
+                Moving gap to nominal value, {gap_to_match_dcm_energy:.3f}mm"
             )
-
-            # Check if undulator gap is close enough to the value from the DCM
-            current_gap = self.parent.undulator.current_gap.get()
-
-            if (
-                abs(gap_to_match_dcm_energy - current_gap)
-                > self.parent.undulator.gap_discrepancy_tolerance_mm
-            ):
-                LOGGER.info(
-                    f"Undulator gap mismatch. {abs(gap_to_match_dcm_energy-current_gap):.3f}mm is outside tolerance.\
-                    Moving gap to nominal value, {gap_to_match_dcm_energy:.3f}mm"
+            if not TEST_MODE:
+                # Only move if the gap is sufficiently different to the value from the
+                # DCM lookup table AND we're not in TEST_MODE
+                await self.undulator.gap_motor.set(
+                    gap_to_match_dcm_energy,
+                    timeout=STATUS_TIMEOUT_S,
                 )
-                if not TEST_MODE:
-                    status &= self.parent.undulator.gap_motor.move(
-                        gap_to_match_dcm_energy, timeout=STATUS_TIMEOUT_S
-                    )
-
-            return status
-
-    energy_kev = Component(EnergySignal)
+            else:
+                LOGGER.debug("In test mode, not moving ID gap")
+        else:
+            LOGGER.debug(
+                "Gap is already in the correct place for the new energy value "
+                f"{energy_kev}, no need to ask it to move"
+            )
 
-    def __init__(self, undulator: Undulator, dcm: DCM, *args, **kwargs):
-        super().__init__(*args, **kwargs)
-        self.undulator = undulator
-        self.dcm = dcm
+    async def _gap_to_match_dcm_energy(self, energy_kev: float) -> float:
+        # Get 2d np.array converting energies to undulator gap distance, from lookup table
+        energy_to_distance_table = await energy_distance_table(
+            self.id_gap_lookup_table_path
+        )
+
+        # Use the lookup table to get the undulator gap associated with this dcm energy
+        return _get_closest_gap_for_energy(
+            energy_kev * 1000,
+            energy_to_distance_table,
+        )

src/dodal/devices/util/adjuster_plans.py

@@ -8,12 +8,13 @@
 from bluesky import plan_stubs as bps
 from bluesky.run_engine import Msg
 from ophyd.epics_motor import EpicsMotor
+from ophyd_async.epics.motion import Motor
 
 from dodal.log import LOGGER
 
 
 def lookup_table_adjuster(
-    lookup_table: Callable[[float], float], output_device: EpicsMotor, input
+    lookup_table: Callable[[float], float], output_device: EpicsMotor | Motor, input
 ):
     """Returns a callable that adjusts a value according to a lookup table"""