473 create a device for xpress3 areadetector

src/dodal/beamlines/i03.py

@@ -29,6 +29,7 @@
 from dodal.devices.undulator_dcm import UndulatorDCM
 from dodal.devices.webcam import Webcam
 from dodal.devices.xbpm_feedback import XBPMFeedback
+from dodal.devices.xspress3.xspress3 import Xspress3
 from dodal.devices.xspress3_mini.xspress3_mini import Xspress3Mini
 from dodal.devices.zebra import Zebra
 from dodal.devices.zebra_controlled_shutter import ZebraShutter
@@ -375,6 +376,36 @@ def xspress3mini(
     )
 
 
+def xspress3(
+    wait_for_connection: bool = True, fake_with_ophyd_sim: bool = False
+) -> Xspress3:
+    """Get the i03 Xspress3 device, instantiate it if it hasn't already been.
+    If this is called when already instantiated in i03, it will return the existing object.
+    """
+    return device_instantiation(
+        Xspress3,
+        "xspress3",
+        "-EA-XSP3-01:",
+        wait_for_connection,
+        fake_with_ophyd_sim,
+    )
+
+
+def attenuator(
+    wait_for_connection: bool = True, fake_with_ophyd_sim: bool = False
+) -> Attenuator:
+    """Get the i03 attenuator device, instantiate it if it hasn't already been.
+    If this is called when already instantiated in i03, it will return the existing object.
+    """
+    return device_instantiation(
+        Attenuator,
+        "attenuator",
+        "-EA-ATTN-01:",
+        wait_for_connection,
+        fake_with_ophyd_sim,
+    )
+
+
 def panda(
     wait_for_connection: bool = True, fake_with_ophyd_sim: bool = False
 ) -> HDFPanda:

src/dodal/beamlines/i20_1.py

@@ -1,6 +1,7 @@
 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.turbo_slit import TurboSlit
+from dodal.devices.xspress3.xspress3 import Xspress3
 from dodal.log import set_beamline as set_log_beamline
 from dodal.utils import get_beamline_name
 
@@ -23,3 +24,20 @@ def turbo_slit(
         wait=wait_for_connection,
         fake=fake_with_ophyd_sim,
     )
+
+
+def express3(
+    wait_for_connection: bool = True, fake_with_ophyd_sim: bool = False
+) -> Xspress3:
+    """
+    16 channels Xspress3 detector
+    """
+
+    return device_instantiation(
+        Xspress3,
+        prefix="-EA-DET-03:",
+        name="Xspress3",
+        num_channels=16,
+        wait=wait_for_connection,
+        fake=fake_with_ophyd_sim,
+    )

src/dodal/devices/xspress3/xspress3.py

@@ -0,0 +1,147 @@
+from enum import Enum
+
+from numpy import float64
+from numpy.typing import NDArray
+from ophyd_async.core import (
+    AsyncStatus,
+    Device,
+    DeviceVector,
+    set_and_wait_for_value,
+    wait_for_value,
+)
+from ophyd_async.epics.signal.signal import (
+    epics_signal_r,
+    epics_signal_rw,
+    epics_signal_rw_rbv,
+)
+
+from dodal.devices.xspress3.xspress3_channel import (
+    AcquireState,
+    Xspress3Channel,
+    Xspress3ROIChannel,
+)
+from dodal.log import LOGGER
+
+
+class TriggerMode(str, Enum):
+    SOFTWARE = "Software"
+    HARDWARE = "Hardware"
+    BURST = "Burst"
+    TTL_Veto_Only = "TTL Veto Only"
+    IDC = "IDC"
+    SOTWARE_START_STOP = "Software Start/Stop"
+    TTL_BOTH = "TTL Both"
+    LVDS_VETO_ONLY = "LVDS Veto Only"
+    LVDS_both = "LVDS Both"
+
+
+class UpdateRBV(str, Enum):
+    DISABLED = "Disabled"
+    ENABLED = "Enabled"
+
+
+'''
+class EraseState(str, Enum):
+    """
+    Epic does not actually have a str value so the only way to set it is
+    to pass int(position).
+    """
+
+    DONE = ""
+    ERASE = ""
+
+'''
+
+
+class AcquireRBVState(str, Enum):
+    DONE = "Done"
+    ACQUIRE = "Acquiring"
+
+
+class DetectorState(str, Enum):
+    IDLE = "Idle"
+    ACQUIRE = "Acquire"
+    READOUT = "Readout"
+    CORRECT = "Correct"
+    Saving = "Saving"
+    ABORTING = "Aborting"
+    ERROR = "Error"
+    WAITING = "Waiting"
+    INTILTIALIZING = "Initializing"
+    DISCONNECTED = "Disconnected"
+    ABORTED = "Aborted"
+
+
+class Xspress3(Device):
+    """Xpress/XpressMini is a region of interest (ROI) picker that sums the detector
+    output into a scaler with user-defined regions. It is often used as a signal
+     discriminator to provide better energy resolution in X-ray detection experiments.
+    """
+
+    def __init__(
+        self, prefix: str, name: str = "", num_channels: int = 1, timeout=1
+    ) -> None:
+        self.channels = DeviceVector(
+            {i: Xspress3Channel(f"{prefix}C{i}_") for i in range(1, num_channels + 1)}
+        )
+        """MCA on/off switch readback"""
+        self.get_roi_calc_status = DeviceVector(
+            {
+                i: epics_signal_rw(float, f"{prefix}MCA{i}:Enable_RBV")
+                for i in range(1, num_channels + 1)
+            }
+        )
+        """start and size of the multi-channel analyzer (MCA) array"""
+        self.roi_mca = DeviceVector(
+            {
+                i: Xspress3ROIChannel(f"{prefix}ROISUM{i}:")
+                for i in range(1, num_channels + 1)
+            }
+        )
+
+        """signal for the correct MCA spectrum (1d array)"""
+        self.dt_corrected_latest_mca = DeviceVector(
+            {
+                i: epics_signal_r(NDArray[float64], f"{prefix}ARR{i}:ArrayData")
+                for i in range(1, num_channels + 1)
+            }
+        )
+
+        """Shared controls for triggering detection"""
+        self.timeout = timeout
+        self.acquire_time = epics_signal_rw(float, prefix + "AcquireTime")
+        # Epic does not actually have a str value for erase so the only way to set it is to pass int(position).
+        self.erase = epics_signal_rw(int, prefix + "ERASE")
+        self.max_num_channels = epics_signal_r(int, prefix + "MAX_NUM_CHANNELS_RBV")
+        # acquire and acquire readback has a different enum
+        self.acquire = epics_signal_rw(AcquireState, prefix + "Acquire")
+        self.acquire_rbv = epics_signal_r(AcquireRBVState, prefix + "Acquire_RBV")
+        self.trigger_mode = epics_signal_rw_rbv(TriggerMode, prefix + "TriggerMode")
+
+        self.detector_state = epics_signal_r(
+            DetectorState, prefix + "DetectorState_RBV"
+        )
+
+        self.set_num_images = epics_signal_rw(int, prefix + "NumImages")
+        self.detector_busy_states = [
+            DetectorState.ACQUIRE,
+            DetectorState.CORRECT,
+            DetectorState.ABORTING,
+        ]
+        super().__init__(name=name)
+
+    async def stage(self) -> AsyncStatus:
+        LOGGER.info("Arming Xspress3 detector...")
+        await self.trigger_mode.set(TriggerMode.BURST)
+        await wait_for_value(
+            self.detector_state,
+            lambda v: v in self.detector_busy_states,
+            timeout=self.timeout,
+        )
+        await self.erase.set(1)
+        await set_and_wait_for_value(
+            self.acquire, AcquireState.ACQUIRE, timeout=self.timeout
+        )
+        return AsyncStatus(
+            wait_for_value(self.acquire_rbv, AcquireRBVState.DONE, timeout=self.timeout)
+        )

src/dodal/devices/xspress3/xspress3_channel.py

@@ -0,0 +1,42 @@
+from enum import Enum
+
+from ophyd_async.core import Device
+from ophyd_async.epics.signal import epics_signal_r, epics_signal_rw
+
+
+class AcquireState(str, Enum):
+    DONE = "Done"
+    ACQUIRE = "Acquire"
+
+
+class Xspress3Channel(Device):
+    """
+    Xspress3 Channel contains the truncated detector data and its collection conditions
+     including the definition of ROI(region of interest).
+    """
+
+    def __init__(self, prefix: str, name: str = "") -> None:
+        self.update_arrays = epics_signal_rw(AcquireState, prefix + "SCAS:TS:TSAcquire")
+
+        self.roi_high_limit = epics_signal_rw(int, prefix + "SCA5_HLM")
+        self.roi_low_limit = epics_signal_rw(int, prefix + "SCA5_LLM")
+        self.time = epics_signal_r(int, prefix + "SCA0:Value_RBV")
+        self.reset_ticks = epics_signal_r(int, prefix + "SCA1:Value_RBV")
+        self.reset_count = epics_signal_r(int, prefix + "SCA2:Value_RBV")
+        self.all_event = epics_signal_r(int, prefix + "SCA3:Value_RBV")
+        self.all_good = epics_signal_r(int, prefix + "SCA4:Value_RBV")
+        self.pileup = epics_signal_r(int, prefix + "SCA7:Value_RBV")
+        self.total_time = epics_signal_r(int, prefix + "SCA8:Value_RBV")
+        self.mca_roi1_LLM = epics_signal_r(int, prefix + "SCA8:Value_RBV")
+        super().__init__(name=name)
+
+
+class Xspress3ROIChannel(Device):
+    """
+    This is the Xspress3 multi-channel analyzer range
+    """
+
+    def __init__(self, prefix: str, name: str = "") -> None:
+        self.roi_start_x = epics_signal_rw(int, prefix + "MinX")
+        self.roi_size_x = epics_signal_rw(int, prefix + "SizeX")
+        super().__init__(name=name)

src/dodal/devices/xspress3_mini/xspress3_mini.py

@@ -75,7 +75,9 @@ def set(self, value, *, timeout=None, settle_time=None, **kwargs):
     erase = Component(EpicsSignal, "ERASE", string=True)
     get_max_num_channels = Component(EpicsSignalRO, "MAX_NUM_CHANNELS_RBV")
     acquire = Component(EpicsSignalWithRBV, "Acquire")
-    get_roi_calc_mini = Component(EpicsSignal, "MCA1:Enable_RBV")
+    get_roi_calc_mini = Component(
+        EpicsSignal, "MCA1:Enable_RBV"
+    )  # THIS CHANGE WITH CHANNEL
     trigger_mode_mini = Component(EpicsSignalWithRBV, "TriggerMode")
     roi_start_x = Component(EpicsSignal, "ROISUM1:MinX")
     roi_size_x = Component(EpicsSignal, "ROISUM1:SizeX")

tests/devices/unit_tests/test_xspress3.py

@@ -0,0 +1,45 @@
+import asyncio
+
+import pytest
+from ophyd_async.core import DeviceCollector, set_mock_value
+
+from dodal.devices.xspress3.xspress3 import (
+    AcquireState,
+    DetectorState,
+    Xspress3,
+)
+
+
+@pytest.fixture
+async def mock_xspress3mini(prefix: str = "BLXX-EA-DET-007:") -> Xspress3:
+    async with DeviceCollector(mock=True):
+        mock_xspress3mini = Xspress3(prefix, "Xspress3Mini", 2)
+    assert mock_xspress3mini.channels[1].name == "Xspress3Mini-channels-1"
+    assert mock_xspress3mini.channels[2].name == "Xspress3Mini-channels-2"
+    assert (
+        mock_xspress3mini.get_roi_calc_status[1].name
+        == "Xspress3Mini-get_roi_calc_status-1"
+    )
+    assert (
+        mock_xspress3mini.get_roi_calc_status[2].name
+        == "Xspress3Mini-get_roi_calc_status-2"
+    )
+    assert mock_xspress3mini.roi_mca[1].name == "Xspress3Mini-roi_mca-1"
+    assert mock_xspress3mini.roi_mca[2].name == "Xspress3Mini-roi_mca-2"
+
+    return mock_xspress3mini
+
+
+async def test_stage_success_on_busy_state(mock_xspress3mini: Xspress3):
+    set_mock_value(mock_xspress3mini.detector_state, DetectorState.ACQUIRE)
+    status = await mock_xspress3mini.stage()
+    assert status.done is False
+    set_mock_value(mock_xspress3mini.acquire, AcquireState.DONE)
+    await asyncio.sleep(0.1)
+    assert status.done is True
+
+
+async def test_stage_fail_on_not_busy_state(mock_xspress3mini: Xspress3):
+    set_mock_value(mock_xspress3mini.detector_state, DetectorState.IDLE)
+    with pytest.raises(TimeoutError):
+        await mock_xspress3mini.stage()