Merge pull request #94 from DiamondLightSource/93_add_arming_to_xspre…

…ss3mini

Add xspress3mini arming

src/dodal/devices/status.py

@@ -12,3 +12,16 @@ def value_is(value, **_):
         return value == expected_value
 
     return SubscriptionStatus(subscribable, value_is, timeout=timeout)
+
+
+# Returns a status which is completed when the subscriptable contains a value within the expected_value list
+def await_value_in_list(
+    subscribable: Any, expected_value: list, timeout: Union[None, int] = None
+) -> SubscriptionStatus:
+    def value_is(value, **_):
+        return value in expected_value
+
+    if type(expected_value) != list:
+        raise TypeError(f"expected value {expected_value} is not a list")
+    else:
+        return SubscriptionStatus(subscribable, value_is, timeout=timeout)

src/dodal/devices/xspress3_mini/xspress3_mini.py

@@ -1,23 +1,101 @@
-from ophyd import Component, Device, EpicsSignal, EpicsSignalRO, EpicsSignalWithRBV
+from enum import Enum
 
+from ophyd import (
+    Component,
+    Device,
+    EpicsSignal,
+    EpicsSignalRO,
+    EpicsSignalWithRBV,
+    Signal,
+)
+from ophyd.status import Status
+
+from dodal.devices.status import await_value_in_list
 from dodal.devices.xspress3_mini.xspress3_mini_channel import Xspress3MiniChannel
+from dodal.log import LOGGER
+
+
+class TriggerMode(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(Enum):
+    DISABLED = "Disabled"
+    ENABLED = "Enabled"
+
+
+class EraseState(Enum):
+    DONE = "Done"
+    ERASE = "Erase"
+
+
+class AcquireState(Enum):
+    DONE = "Done"
+    ACQUIRE = "Acquire"
+
+
+class DetectorState(Enum):
+    ACQUIRE = "Acquire"
+    CORRECT = "Correct"
+    READOUT = "Readout"
+    ABORTING = "Aborting"
+
+    IDLE = "Idle"
+    SAVING = "Saving"
+    ERROR = "Error"
+    INTILTIALIZING = "Initializing"
+    DISCONNECTED = "Disconnected"
+    ABORTED = "Aborted"
 
 
 class Xspress3Mini(Device):
+    class ArmingSignal(Signal):
+        def set(self, value, *, timeout=None, settle_time=None, **kwargs):
+            return self.parent.arm()
+
+    do_arm: ArmingSignal = Component(ArmingSignal)
+
     # Assume only one channel for now
     channel_1 = Component(Xspress3MiniChannel, "C1_")
 
     erase: EpicsSignal = Component(EpicsSignal, "ERASE")
     get_max_num_channels = Component(EpicsSignalRO, "MAX_NUM_CHANNELS_RBV")
-
     acquire: EpicsSignal = Component(EpicsSignal, "Acquire")
-
     get_roi_calc_mini: EpicsSignal = Component(EpicsSignal, "MCA1:Enable_RBV")
-
-    NUMBER_ROIS_DEFAULT = 6
-
     trigger_mode_mini: EpicsSignalWithRBV = Component(EpicsSignalWithRBV, "TriggerMode")
-
     roi_start_x: EpicsSignal = Component(EpicsSignal, "ROISUM1:MinX")
     roi_size_x: EpicsSignal = Component(EpicsSignal, "ROISUM1:SizeX")
     acquire_time: EpicsSignal = Component(EpicsSignal, "AcquireTime")
+    detector_state: EpicsSignalRO = Component(EpicsSignalRO, ":DetectorState_RBV")
+    NUMBER_ROIS_DEFAULT = 6
+    acquire_status: Status = None
+
+    detector_busy_states = [
+        DetectorState.ACQUIRE.value,
+        DetectorState.CORRECT.value,
+        DetectorState.ABORTING.value,
+    ]
+
+    def stage(self):
+        self.arm().wait(timeout=10)
+
+    def do_start(self) -> Status:
+        self.erase.put(EraseState.ERASE.value)
+        status = self.channel_1.update_arrays.set(AcquireState.DONE.value)
+        self.acquire_status = self.acquire.set(AcquireState.ACQUIRE.value)
+        return status
+
+    def arm(self) -> Status:
+        LOGGER.info("Arming Xspress3Mini detector...")
+        self.trigger_mode_mini.put(TriggerMode.BURST.value)
+        self.do_start().wait(timeout=10)
+        arm_status = await_value_in_list(self.detector_state, self.detector_busy_states)
+        return arm_status

src/dodal/devices/xspress3_mini/xspress3_mini_channel.py

@@ -2,7 +2,7 @@
 
 
 class Xspress3MiniChannel(Device):
-    sca5_update_arrays_mini = Component(EpicsSignalRO, "SCAS:TS:TSAcquire")
+    update_arrays = Component(EpicsSignal, "SCAS:TS:TSAcquire")
 
     roi_high_limit = Component(EpicsSignal, "SCA5_HLM")
     roi_llm = Component(EpicsSignal, "SCA5_LLM")

tests/devices/unit_tests/test_status.py

@@ -0,0 +1,29 @@
+import pytest
+from ophyd import Component, Device, EpicsSignalRO
+from ophyd.sim import make_fake_device
+
+from dodal.devices.status import await_value_in_list
+
+
+class FakeDevice(Device):
+    pv: EpicsSignalRO = Component(EpicsSignalRO, "test")
+
+
+@pytest.fixture
+def fake_device():
+    MyFakeDevice = make_fake_device(FakeDevice)
+    fake_device = MyFakeDevice(name="test")
+    return fake_device
+
+
+@pytest.mark.parametrize("awaited_value", [(1), (5.3), (False)])
+def test_await_value_in_list_with_no_list_parameter_fails(awaited_value, fake_device):
+    with pytest.raises(TypeError):
+        await_value_in_list(fake_device.pv, awaited_value)
+
+
+def test_await_value_in_list_success(fake_device):
+    status = await_value_in_list(fake_device.pv, [1, 2, 3, 4, 5])
+    assert status.done is False
+    fake_device.pv.sim_put(5)
+    status.wait(timeout=1)

tests/devices/unit_tests/test_xspress3mini.py

@@ -0,0 +1,54 @@
+from unittest.mock import MagicMock
+
+import pytest
+from bluesky import RunEngine
+from bluesky import plan_stubs as bps
+from ophyd.sim import make_fake_device
+from ophyd.status import Status
+
+from dodal.devices.xspress3_mini.xspress3_mini import DetectorState, Xspress3Mini
+
+
+def get_good_status() -> Status:
+    status = Status()
+    status.set_finished()
+    return status
+
+
+def get_bad_status() -> Status:
+    status = Status()
+    status.set_exception(Exception)
+    return status
+
+
+@pytest.fixture
+def fake_xspress3mini():
+    FakeXspress3Mini: Xspress3Mini = make_fake_device(Xspress3Mini)
+    fake_xspress3mini: Xspress3Mini = FakeXspress3Mini(name="xspress3mini")
+    return fake_xspress3mini
+
+
+def test_arm_success_on_busy_state(fake_xspress3mini):
+    fake_xspress3mini.detector_state.sim_put(DetectorState.IDLE.value)
+    status = fake_xspress3mini.arm()
+    assert status.done is False
+    fake_xspress3mini.detector_state.sim_put(DetectorState.ACQUIRE.value)
+    status.wait(timeout=1)
+    fake_xspress3mini.acquire_status.wait(timeout=1)
+
+
+def test_stage_in_busy_state(fake_xspress3mini):
+    fake_xspress3mini.detector_state.sim_put(DetectorState.ACQUIRE.value)
+    RE = RunEngine()
+    RE(bps.stage(fake_xspress3mini))
+    fake_xspress3mini.acquire_status.wait(timeout=1)
+
+
+def test_stage_fails_in_failed_acquire_state(fake_xspress3mini):
+    bad_status = Status()
+    bad_status.set_exception(Exception)
+    RE = RunEngine()
+    fake_xspress3mini.do_start = MagicMock(return_value=get_good_status())
+    fake_xspress3mini.acquire_status = get_bad_status()
+    with pytest.raises(Exception):
+        RE(bps.stage(fake_xspress3mini))