add i18 beamline definition

pyproject.toml

@@ -15,7 +15,8 @@ description = "Ophyd devices and other utils that could be used across DLS beaml
 dependencies = [
     "click",
     "ophyd",
-    "ophyd-async >= 0.8.0a5",
+    "ophyd-async>=0.8.0a5",
+    "ophyd-async[sim]",
     "bluesky",
     "pyepics",
     "dataclasses-json",

src/dodal/beamlines/i18.py

@@ -0,0 +1,286 @@
+from pathlib import Path
+
+from ophyd_async.fastcs.panda import HDFPanda
+
+from dodal.common.beamlines.beamline_utils import (
+    device_instantiation,
+    get_path_provider,
+    set_path_provider,
+)
+from dodal.common.beamlines.beamline_utils import set_beamline as set_utils_beamline
+from dodal.common.beamlines.device_helpers import numbered_slits
+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.sim_detector import SimDetector
+from dodal.devices.i18.table import Table
+from dodal.devices.i22.dcm import CrystalMetadata, 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.devices.xspress3.xspress3 import Xspress3
+from dodal.log import set_beamline as set_log_beamline
+from dodal.utils import BeamlinePrefix, get_beamline_name, skip_device
+
+BL = get_beamline_name("i18")
+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(),
+    )
+)
+
+
+def synchrotron(
+    wait_for_connection: bool = True, fake_with_ophyd_sim: bool = False
+) -> Synchrotron:
+    return device_instantiation(
+        Synchrotron,
+        "synchrotron",
+        "",
+        wait_for_connection,
+        fake_with_ophyd_sim,
+    )
+
+
+# not ready yet
+@skip_device()
+def undulator(
+    wait_for_connection: bool = True,
+    fake_with_ophyd_sim: bool = False,
+) -> Undulator:
+    return device_instantiation(
+        Undulator,
+        "undulator",
+        f"{BeamlinePrefix(BL).insertion_prefix}-MO-SERVC-01:",
+        wait_for_connection,
+        fake_with_ophyd_sim,
+        bl_prefix=False,
+        poles=80,
+        length=2.0,
+    )
+
+
+@skip_device()
+def slits_1(
+    wait_for_connection: bool = True,
+    fake_with_ophyd_sim: bool = False,
+) -> Slits:
+    return numbered_slits(
+        1,
+        wait_for_connection,
+        fake_with_ophyd_sim,
+    )
+
+
+# Must document what PandAs are physically connected to
+# See: https://github.com/bluesky/ophyd-async/issues/284
+@skip_device()
+def panda1(
+    wait_for_connection: bool = True,
+    fake_with_ophyd_sim: bool = False,
+) -> HDFPanda:
+    return device_instantiation(
+        HDFPanda,
+        "panda1",
+        "-MO-PANDA-01:",
+        wait_for_connection,
+        fake_with_ophyd_sim,
+        path_provider=get_path_provider(),
+    )
+
+
+def old_xspress3(
+    wait_for_connection: bool = True, fake_with_ophyd_sim: bool = False
+) -> Xspress3:
+    return device_instantiation(
+        Xspress3,
+        prefix="-EA-XSP-02:",
+        name="Xspress3",
+        num_channels=16,
+        wait=wait_for_connection,
+        fake=fake_with_ophyd_sim,
+    )
+
+
+# odin detectors are not yet supported.
+# There is a controls project in the works,
+# not ready anytime soon
+@skip_device()
+def xspress3_odin(
+    wait_for_connection: bool = True, fake_with_ophyd_sim: bool = False
+) -> Xspress3:
+    return device_instantiation(
+        Xspress3,
+        prefix="-EA-XSP-03:",
+        name="Xspress3",
+        num_channels=4,
+        wait=wait_for_connection,
+        fake=fake_with_ophyd_sim,
+    )
+
+
+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"),
+)
+
+
+def dcm(
+    wait_for_connection: bool = True,
+    fake_with_ophyd_sim: bool = False,
+) -> DoubleCrystalMonochromator:
+    return device_instantiation(
+        DoubleCrystalMonochromator,
+        "dcm",
+        f"{BeamlinePrefix(BL).beamline_prefix}-MO-DCM-01:",
+        wait_for_connection,
+        fake_with_ophyd_sim,
+        bl_prefix=False,
+        temperature_prefix=f"{BeamlinePrefix(BL).beamline_prefix}-DI-DCM-01:",
+        crystal_1_metadata=crystal_1_metadata,
+        crystal_2_metadata=crystal_2_metadata,
+    )
+
+
+@skip_device()
+def i0(
+    wait_for_connection: bool = True,
+    fake_with_ophyd_sim: bool = False,
+) -> TetrammDetector:
+    return device_instantiation(
+        TetrammDetector,
+        "i0",
+        "-DI-XBPM-02:",
+        wait_for_connection,
+        fake_with_ophyd_sim,
+        type="Cividec Diamond XBPM",
+        path_provider=get_path_provider(),
+    )
+
+
+@skip_device()
+def it(
+    wait_for_connection: bool = True,
+    fake_with_ophyd_sim: bool = False,
+) -> TetrammDetector:
+    return device_instantiation(
+        TetrammDetector,
+        "it",
+        "-DI-XBPM-01:",
+        wait_for_connection,
+        fake_with_ophyd_sim,
+        type="Tetramm",
+        path_provider=get_path_provider(),
+    )
+
+
+@skip_device
+def vfm(
+    wait_for_connection: bool = True,
+    fake_with_ophyd_sim: bool = False,
+) -> KBMirror:
+    return device_instantiation(
+        KBMirror,
+        "vfm",
+        "-OP-VFM-01:",
+        wait_for_connection,
+        fake_with_ophyd_sim,
+    )
+
+
+def hfm(
+    wait_for_connection: bool = True,
+    fake_with_ophyd_sim: bool = False,
+) -> KBMirror:
+    return device_instantiation(
+        KBMirror,
+        "hfm",
+        "-OP-HFM-01:",
+        wait_for_connection,
+        fake_with_ophyd_sim,
+    )
+
+
+def diode(wait_for_connection: bool = True, fake_with_ophyd_sim: bool = False) -> Diode:
+    return device_instantiation(
+        Diode,
+        "diodad7bdiode",
+        "-DI-PHDGN-07:",
+        wait_for_connection,
+        fake_with_ophyd_sim,
+    )
+
+
+def main_table(
+    wait_for_connection: bool = True, fake_with_ophyd_sim: bool = False
+) -> Table:
+    return device_instantiation(
+        Table,
+        "table",
+        "-MO-TABLE-01:",
+        wait_for_connection,
+        fake_with_ophyd_sim,
+    )
+
+
+def thor_labs_table(
+    wait_for_connection: bool = True, fake_with_ophyd_sim: bool = False
+) -> Table:
+    return device_instantiation(
+        Table,
+        "table",
+        "-MO-TABLE-02:",
+        wait_for_connection,
+        fake_with_ophyd_sim,
+    )
+
+
+# SIMULATED DEVICES
+
+
+# this is a mock table, not sure how does it relate to the real Table, maybe just a fake option in instantiation is needed
+@skip_device()
+def raster_stage(
+    wait_for_connection: bool = True, fake_with_ophyd_sim: bool = False
+) -> Table:
+    return device_instantiation(
+        Table,
+        "raster_stage",
+        "-MO-SIM-01:",
+        wait_for_connection,
+        fake_with_ophyd_sim,
+    )
+
+
+def sim_detector(wait_for_connection: bool = True, fake_with_ophyd_sim: bool = False):
+    return device_instantiation(
+        SimDetector,
+        "sim_detector",
+        "-MO-SIM-01:",
+        wait_for_connection,
+        fake_with_ophyd_sim,
+    )

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/sim_detector.py

@@ -0,0 +1,18 @@
+from ophyd_async.epics.motor import Motor
+from ophyd_async.sim.demo import PatternGenerator
+
+
+class SimDetector:
+    def __init__(self, name: str, motor: Motor, motor_field: str):
+        self.name = name
+        self.motor = motor
+        self.motor_field = motor_field
+        self.pattern_generator = PatternGenerator(
+            saturation_exposure_time=0.1, detector_height=100, detector_width=100
+        )
+
+    def read(self):
+        return {self.name: {"value": self.pattern_generator}}
+
+    def describe(self):
+        return {self.name: {"source": "synthetic", "dtype": "number"}}

src/dodal/devices/i18/table.py

@@ -0,0 +1,28 @@
+from ophyd_async.core import (
+    AsyncStatus,
+    StandardReadable,
+)
+from ophyd_async.epics.motor import Motor
+from pydantic import BaseModel
+
+
+class Four_D_Position(BaseModel):
+    x: float
+    y: float
+    z: float
+    theta: float
+
+
+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: Four_D_Position):
+        self.x.set(value.x)
+        self.y.set(value.y)