add i18 beamline definition - WITH device_factory

src/dodal/beamlines/i18.py

@@ -0,0 +1,140 @@
+from pathlib import Path
+
+from ophyd_async.fastcs.panda import HDFPanda
+
+from dodal.common.beamlines.beamline_utils import (
+    device_factory,
+    get_path_provider,
+    set_path_provider,
+)
+from dodal.common.beamlines.beamline_utils import set_beamline as set_utils_beamline
+from dodal.common.crystal_metadata import CrystalMetadata
+from dodal.common.visit import (
+    LocalDirectoryServiceClient,
+    StaticVisitPathProvider,
+)
+from dodal.devices.i18.diode import Diode
+from dodal.devices.i18.KBMirror import KBMirror
+from dodal.devices.i18.table import Table
+from dodal.devices.i18.thor_labs_stage import ThorLabsStage
+from dodal.devices.i22.dcm import DoubleCrystalMonochromator
+from dodal.devices.slits import Slits
+from dodal.devices.synchrotron import Synchrotron
+from dodal.devices.tetramm import TetrammDetector
+from dodal.devices.undulator import Undulator
+from dodal.log import set_beamline as set_log_beamline
+from dodal.utils import BeamlinePrefix, get_beamline_name
+
+BL = get_beamline_name("i18")
+PREFIX = BeamlinePrefix(BL)
+set_log_beamline(BL)
+set_utils_beamline(BL)
+
+
+# Currently we must hard-code the visit, determining the visit at runtime requires
+# infrastructure that is still WIP.
+# Communication with GDA is also WIP so for now we determine an arbitrary scan number
+# locally and write the commissioning directory. The scan number is not guaranteed to
+# be unique and the data is at risk - this configuration is for testing only.
+set_path_provider(
+    StaticVisitPathProvider(
+        BL,
+        Path("/dls/i18/data/2024/cm37264-2/bluesky"),
+        client=LocalDirectoryServiceClient(),
+    )
+)
+
+
+@device_factory()
+def synchrotron() -> Synchrotron:
+    return Synchrotron()
+
+
+@device_factory()
+def undulator() -> Undulator:
+    return Undulator(f"{PREFIX.insertion_prefix}-MO-SERVC-01:")
+
+
+@device_factory()
+def dcm() -> DoubleCrystalMonochromator:
+    crystal_1_metadata = CrystalMetadata(
+        usage="Bragg",
+        type="silicon",
+        reflection=(1, 1, 1),
+        d_spacing=(3.13475, "nm"),
+    )
+
+    crystal_2_metadata = CrystalMetadata(
+        usage="Bragg",
+        type="silicon",
+        reflection=(1, 1, 1),
+        d_spacing=(3.13475, "nm"),
+    )
+
+    return DoubleCrystalMonochromator(
+        prefix=f"{PREFIX.beamline_prefix}-MO-DCM-01:",
+        temperature_prefix=f"{PREFIX.beamline_prefix}-DI-DCM-01:",
+        crystal_1_metadata=crystal_1_metadata,
+        crystal_2_metadata=crystal_2_metadata,
+    )
+
+
+@device_factory()
+def slits_1() -> Slits:
+    return Slits(
+        f"{PREFIX.beamline_prefix}-AL-SLITS-01:",
+        x_centre="X:CENTER",
+        y_centre="Y:CENTER",
+    )
+
+
+# PandA IOC needs to be updated to support PVI
+@device_factory(skip=True)
+def panda1() -> HDFPanda:
+    return HDFPanda(
+        f"{PREFIX.beamline_prefix}-MO-PANDA-01:",
+        path_provider=get_path_provider(),
+    )
+
+
+@device_factory()
+def i0() -> TetrammDetector:
+    return TetrammDetector(
+        f"{PREFIX.beamline_prefix}-DI-XBPM-02:",
+        path_provider=get_path_provider(),
+        type="Cividec Diamond XBPM",
+    )
+
+
+@device_factory()
+def it() -> TetrammDetector:
+    return TetrammDetector(
+        f"{PREFIX.beamline_prefix}-DI-XBPM-01:",
+        path_provider=get_path_provider(),
+    )
+
+
+@device_factory(skip=True)
+# VFM uses different IOC than HFM https://github.com/DiamondLightSource/dodal/issues/1009
+def vfm() -> KBMirror:
+    return KBMirror(f"{PREFIX.beamline_prefix}-OP-VFM-01:")
+
+
+@device_factory()
+def hfm() -> KBMirror:
+    return KBMirror(f"{PREFIX.beamline_prefix}-OP-HFM-01:")
+
+
+@device_factory()
+def d7diode() -> Diode:
+    return Diode(f"{PREFIX.beamline_prefix}-DI-PHDGN-07:")
+
+
+@device_factory()
+def main_table() -> Table:
+    return Table(f"{PREFIX.beamline_prefix}-MO-TABLE-01:")
+
+
+@device_factory()
+def thor_labs_stage() -> ThorLabsStage:
+    return ThorLabsStage(f"{PREFIX.beamline_prefix}-MO-TABLE-02:")

src/dodal/devices/i18/KBMirror.py

@@ -0,0 +1,31 @@
+from bluesky.protocols import Movable
+from ophyd_async.core import AsyncStatus, StandardReadable
+from ophyd_async.epics.core import epics_signal_rw
+from pydantic import BaseModel
+
+
+class XYPosition(BaseModel):
+    x: float
+    y: float
+
+
+class KBMirror(StandardReadable, Movable):
+    def __init__(
+        self,
+        prefix: str,
+        name: str = "",
+    ):
+        self._prefix = prefix
+        with self.add_children_as_readables():
+            self.x = epics_signal_rw(float, prefix + "X")
+            self.y = epics_signal_rw(float, prefix + "Y")
+            self.bend1 = epics_signal_rw(float, prefix + "BEND1")
+            self.bend2 = epics_signal_rw(float, prefix + "BEND2")
+            self.curve = epics_signal_rw(float, prefix + "CURVE")
+            self.ellip = epics_signal_rw(float, prefix + "ELLIP")
+        super().__init__(name=name)
+
+    @AsyncStatus.wrap
+    async def set(self, value: XYPosition):
+        self.x.set(value.x)
+        self.y.set(value.y)

src/dodal/devices/i18/diode.py

@@ -0,0 +1,17 @@
+from ophyd_async.core import (
+    StandardReadable,
+)
+from ophyd_async.epics.core import epics_signal_r
+
+
+class Diode(StandardReadable):
+    def __init__(
+        self,
+        prefix: str,
+        name: str = "",
+    ):
+        self._prefix = prefix
+        with self.add_children_as_readables():
+            self.signal = epics_signal_r(float, prefix + "B:DIODE:I")
+
+        super().__init__(name=name)

src/dodal/devices/i18/table.py

@@ -0,0 +1,32 @@
+from ophyd_async.core import (
+    AsyncStatus,
+    StandardReadable,
+)
+from ophyd_async.epics.motor import Motor
+from pydantic import BaseModel
+
+
+class TablePosition(BaseModel):
+    x: float
+    y: float
+    z: float | None = None
+    theta: float | None = None
+
+
+class Table(StandardReadable):
+    def __init__(self, prefix: str, name: str = "") -> None:
+        with self.add_children_as_readables():
+            self.x = Motor(prefix + "X")
+            self.y = Motor(prefix + "Y")
+            self.z = Motor(prefix + "Z")
+            self.theta = Motor(prefix + "THETA")
+        super().__init__(name=name)
+
+    @AsyncStatus.wrap
+    async def set(self, value: TablePosition):
+        self.x.set(value.x)
+        self.y.set(value.y)
+        if value.z:
+            self.z.set(value.z)
+        if value.theta:
+            self.theta.set(value.theta)

src/dodal/devices/i18/thor_labs_stage.py

@@ -0,0 +1,24 @@
+from ophyd_async.core import (
+    AsyncStatus,
+    StandardReadable,
+)
+from ophyd_async.epics.motor import Motor
+from pydantic import BaseModel
+
+
+class XYPosition(BaseModel):
+    x: float
+    y: float
+
+
+class ThorLabsStage(StandardReadable):
+    def __init__(self, prefix: str, name: str = "") -> None:
+        with self.add_children_as_readables():
+            self.x = Motor(prefix + "X")
+            self.y = Motor(prefix + "Y")
+        super().__init__(name=name)
+
+    @AsyncStatus.wrap
+    async def set(self, value: XYPosition):
+        self.x.set(value.x)
+        self.y.set(value.y)

tests/devices/i18/test_kb_mirror.py

@@ -0,0 +1,66 @@
+from unittest.mock import ANY
+
+import pytest
+from ophyd_async.core import DeviceCollector
+from ophyd_async.testing import set_mock_value
+
+from dodal.devices.i18.KBMirror import KBMirror, XYPosition
+
+
+@pytest.fixture
+async def kbmirror() -> KBMirror:
+    """Fixture to set up a mock KBMirror device using DeviceCollector."""
+    async with DeviceCollector(mock=True):
+        kbmirror = KBMirror(prefix="MIRROR:")
+    return kbmirror
+
+
+async def test_setting_xy_position_kbmirror(kbmirror: KBMirror):
+    """
+    Test setting x and y positions on the KBMirror using the ophyd_async mock tools.
+    """
+    # Mock the initial values of the x and y signals
+    set_mock_value(kbmirror.x, 0.0)
+    set_mock_value(kbmirror.y, 0.0)
+
+    # Create a position object
+    position = XYPosition(x=1.23, y=4.56)
+
+    # Call set to update the position
+    await kbmirror.set(position)
+
+    reading = await kbmirror.read()
+    expected_reading = {
+        "kbmirror-y": {
+            "value": 4.56,
+            "timestamp": ANY,
+            "alarm_severity": 0,
+        },
+        "kbmirror-bend1": {
+            "value": 0.0,
+            "timestamp": ANY,
+            "alarm_severity": 0,
+        },
+        "kbmirror-ellip": {
+            "value": 0.0,
+            "timestamp": ANY,
+            "alarm_severity": 0,
+        },
+        "kbmirror-x": {
+            "value": 1.23,
+            "timestamp": ANY,
+            "alarm_severity": 0,
+        },
+        "kbmirror-bend2": {
+            "value": 0.0,
+            "timestamp": ANY,
+            "alarm_severity": 0,
+        },
+        "kbmirror-curve": {
+            "value": 0.0,
+            "timestamp": ANY,
+            "alarm_severity": 0,
+        },
+    }
+
+    assert reading == expected_reading

tests/devices/i18/test_table.py

@@ -0,0 +1,51 @@
+from unittest.mock import ANY
+
+import pytest
+from ophyd_async.core import DeviceCollector
+from ophyd_async.testing import set_mock_value
+
+from dodal.devices.i18.table import Table, TablePosition
+
+
+@pytest.fixture
+async def table() -> Table:
+    """Fixture to set up a mock Table device using DeviceCollector."""
+    async with DeviceCollector(mock=True):
+        table = Table(prefix="MIRROR:")
+    return table
+
+
+async def test_setting_xy_position_table(table: Table):
+    """
+    Test setting x and y positions on the Table using the ophyd_async mock tools.
+    """
+    set_mock_value(table.x.user_readback, 1.23)
+    set_mock_value(table.y.user_readback, 4.56)
+
+    # Create a position object
+    position = TablePosition(x=1.23, y=4.56, z=0.0, theta=0.0)
+
+    # Call set to update the position
+    await table.set(position)
+
+    reading = await table.read()
+    expected_reading = {
+        "table-y": {
+            "value": 4.56,
+            "timestamp": ANY,
+            "alarm_severity": 0,
+        },
+        "table-x": {
+            "value": 1.23,
+            "timestamp": ANY,
+            "alarm_severity": 0,
+        },
+        "table-theta": {
+            "alarm_severity": 0,
+            "timestamp": ANY,
+            "value": 0.0,
+        },
+        "table-z": {"alarm_severity": 0, "timestamp": ANY, "value": 0.0},
+    }
+
+    assert reading == expected_reading

tests/devices/unit_tests/i18/test_thor_labs_stage.py

@@ -0,0 +1,17 @@
+import pytest
+from ophyd_async.core import DeviceCollector
+
+from dodal.devices.i18.thor_labs_stage import ThorLabsStage, XYPosition
+
+
+@pytest.fixture
+async def fake_thor_labs_stage():
+    async with DeviceCollector(mock=True):
+        fake_thor_labs_stage = ThorLabsStage("", "thor_labs_stage")
+
+    return fake_thor_labs_stage
+
+
+async def test_setting(fake_thor_labs_stage: ThorLabsStage):
+    pos = XYPosition(x=5, y=5)
+    await fake_thor_labs_stage.set(pos)