473 create a device for xpress3 areadetector

.vscode/settings.json

@@ -1,5 +1,7 @@
 {
-    "python.testing.pytestArgs": [],
+    "python.testing.pytestArgs": [
+        "tests"
+    ],
     "python.testing.unittestEnabled": false,
     "python.testing.pytestEnabled": true,
     "editor.formatOnSave": true,

src/dodal/beamlines/i03.py

@@ -29,7 +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_mini.xspress3_mini import Xspress3Mini
+from dodal.devices.xspress3.xspress3 import Xspress3
 from dodal.devices.zebra import Zebra
 from dodal.devices.zebra_controlled_shutter import ZebraShutter
 from dodal.devices.zocalo import ZocaloResults
@@ -362,12 +362,12 @@ def zebra(wait_for_connection: bool = True, fake_with_ophyd_sim: bool = False) -
 
 def xspress3mini(
     wait_for_connection: bool = True, fake_with_ophyd_sim: bool = False
-) -> Xspress3Mini:
+) -> Xspress3:
     """Get the i03 Xspress3Mini 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(
-        Xspress3Mini,
+        Xspress3,
         "xspress3mini",
         "-EA-XSP3-01:",
         wait_for_connection,

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,150 @@
+from enum import Enum
+
+from bluesky.protocols import Stageable
+from numpy import float64
+from numpy.typing import NDArray
+from ophyd_async.core import (
+    AsyncStatus,
+    Device,
+    DeviceVector,
+    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 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, Stageable):
+    """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 and signal to noise in X-ray detection experiments.
+     This currently only provide staging functionality.
+
+    Parameters
+    ----------
+    prefix:
+        Beamline part of PV
+    name:
+        Name of the device
+    num_channels:
+        Number of channel xspress3 has, default is 1 for mini.
+    timeout:
+        How long to wait for before timing out for staging/arming of detector default is 1 sec
+    """
+
+    def __init__(
+        self, prefix: str, name: str = "", num_channels: int = 1, timeout: float = 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 corrected 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")
+        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)
+
+    @AsyncStatus.wrap
+    async def stage(self) -> None:
+        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.acquire.set(AcquireState.ACQUIRE)
+        await wait_for_value(
+            self.acquire_rbv, AcquireRBVState.ACQUIRE, timeout=self.timeout
+        )
+
+        await wait_for_value(
+            self.acquire_rbv, AcquireRBVState.DONE, timeout=self.timeout
+        )
+
+    @AsyncStatus.wrap
+    async def unstage(self) -> None:
+        LOGGER.info("unstaging Xspress3 detector...")

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)