Created a dcm as a baseline device for i11

src/dodal/beamlines/i11.py

@@ -13,6 +13,9 @@
     AutotunedCyberstarBlower,
     CyberstarBlower,
 )
+
+# from dodal.devices.common_dcm import BaseDCM, PitchAndRollCrystal, RollCrystal
+from dodal.devices.i11.dcm import DCM
 from dodal.devices.i11.diff_stages import (
     DiffractometerBase,
     DiffractometerStage,
@@ -124,3 +127,11 @@ def slits_4() -> Slits:
 @device_factory()
 def slits_5() -> Slits:
     return Slits(prefix=f"{PREFIX.beamline_prefix}-AL-SLITS-05:")
+
+
+@device_factory()
+def dcm() -> DCM:
+    return DCM(
+        prefix=f"{PREFIX.beamline_prefix}-MO-DCM-01:",
+        xtal_prefix=f"{PREFIX.beamline_prefix}-DI-DCM-01:",
+    )

src/dodal/devices/i11/dcm.py

@@ -0,0 +1,82 @@
+import numpy as np
+from ophyd_async.core import Array1D, derived_signal_r, soft_signal_r_and_setter
+from ophyd_async.epics.core import epics_signal_r
+from ophyd_async.epics.motor import Motor
+
+from dodal.common.crystal_metadata import (
+    CrystalMetadata,
+    MaterialsEnum,
+    make_crystal_metadata_from_material,
+)
+from dodal.devices.common_dcm import (
+    BaseDCM,
+    PitchAndRollCrystal,
+    StationaryCrystal,
+)
+
+# Conversion constant for energy and wavelength, taken from the X-Ray data booklet
+# Converts between energy in KeV and wavelength in angstrom
+_CONVERSION_CONSTANT = 12.3984
+
+
+class DCM(BaseDCM[PitchAndRollCrystal, StationaryCrystal]):
+    """
+    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,
+        xtal_prefix: str,
+        name: str = "",
+        crystal_metadata: CrystalMetadata | None = None,
+    ) -> None:
+        cm = crystal_metadata or make_crystal_metadata_from_material(
+            MaterialsEnum.Si, (1, 1, 1)
+        )
+        with self.add_children_as_readables():
+            self.perp_in_mm = Motor(prefix + "PERP")
+
+            # temperatures
+            self.xtal1_holder_temp = epics_signal_r(str, xtal_prefix + "PT100-1.SEVR")
+            self.xtal1_temp = epics_signal_r(str, xtal_prefix + "PT100-2.SEVR")
+
+            self.xtal2_heater_temp = epics_signal_r(str, xtal_prefix + "PT100-3.SEVR")
+            self.xtal2_temp = epics_signal_r(str, xtal_prefix + "PT100-4.SEVR")
+
+            self.xtal1_heater_temp = epics_signal_r(
+                float, xtal_prefix + "H1:TTEMP:CALC"
+            )
+            self.xtal2_heater_temp = epics_signal_r(
+                float, xtal_prefix + "H2:TTEMP:CALC"
+            )
+
+            self.crystal_metadata_usage, _ = soft_signal_r_and_setter(
+                str, initial_value=cm.usage
+            )
+            self.crystal_metadata_type, _ = soft_signal_r_and_setter(
+                str, initial_value=cm.type
+            )
+            reflection_array = np.array(cm.reflection)
+            self.crystal_metadata_reflection, _ = soft_signal_r_and_setter(
+                Array1D[np.uint64],
+                initial_value=reflection_array,
+            )
+        super().__init__(prefix, PitchAndRollCrystal, StationaryCrystal, name)
+
+        with self.add_children_as_readables():
+            self.wavelength = derived_signal_r(
+                self._wavelength_from_energy,
+                energy=self.energy_in_kev,
+                unit="angstrom",
+            )
+
+    def _wavelength_from_energy(self, energy: float, unit: str) -> float:
+        if energy > 0:
+            return _CONVERSION_CONSTANT / energy
+        return 0

tests/devices/i11/test_i11_dcm.py

@@ -0,0 +1,56 @@
+import bluesky.plans as bp
+import pytest
+from bluesky.run_engine import RunEngine
+from ophyd_async.core import init_devices
+from ophyd_async.testing import (
+    assert_emitted,
+    set_mock_value,
+)
+
+from dodal.devices.i11.dcm import DCM
+
+
+@pytest.fixture
+async def i11_dcm() -> DCM:
+    async with init_devices(mock=True):
+        i11_dcm = DCM(prefix="x-MO-DCM-01:", xtal_prefix="x-DI-DCM-01:")
+    return i11_dcm
+
+
+def test_count_i11_dcm(
+    RE: RunEngine,
+    run_engine_documents: dict[str, list[dict]],
+    i11_dcm: DCM,
+):
+    RE(bp.count([i11_dcm]))
+    assert_emitted(
+        run_engine_documents,
+        start=1,
+        descriptor=1,
+        event=1,
+        stop=1,
+    )
+
+
+@pytest.mark.parametrize(
+    "wavelength,energy,unit",
+    [
+        (0.0, 0.0, "angstrom"),
+        (1.0, 12.3984, "angstrom"),
+        (2.0, 6.1992, "angstrom"),
+    ],
+)
+async def test_i11_wavelength(
+    wavelength: float,
+    energy: float,
+    unit: str,
+    i11_dcm: DCM,
+):
+    set_mock_value(i11_dcm.energy_in_kev.user_readback, energy)
+    set_mock_value(i11_dcm.wavelength_in_a.user_readback, wavelength)
+
+    reading = await i11_dcm.read()
+
+    assert reading["i11_dcm-energy_in_kev"]["value"] == energy
+    assert reading["i11_dcm-wavelength_in_a"]["value"] == wavelength
+    assert reading["i11_dcm-wavelength"]["value"] == wavelength