Added i15-1 zebra shutter

src/dodal/beamlines/i03.py

@@ -61,7 +61,7 @@
     ZebraSources,
     ZebraTTLOutputs,
 )
-from dodal.devices.zebra.zebra_controlled_shutter import ZebraShutter
+from dodal.devices.zebra.zebra_controlled_shutter import MXZebraShutter
 from dodal.devices.zocalo import ZocaloResults, ZocaloSource
 from dodal.log import set_beamline as set_log_beamline
 from dodal.utils import BeamlinePrefix, get_beamline_name
@@ -293,8 +293,8 @@ def panda(path_provider: PathProvider) -> HDFPanda:
 
 
 @devices.factory()
-def sample_shutter() -> ZebraShutter:
-    return ZebraShutter(f"{PREFIX.beamline_prefix}-EA-SHTR-01:")
+def sample_shutter() -> MXZebraShutter:
+    return MXZebraShutter(f"{PREFIX.beamline_prefix}-EA-SHTR-01:")
 
 
 @devices.factory()

src/dodal/beamlines/i04.py

@@ -50,7 +50,7 @@
     ZebraSources,
     ZebraTTLOutputs,
 )
-from dodal.devices.zebra.zebra_controlled_shutter import ZebraShutter
+from dodal.devices.zebra.zebra_controlled_shutter import MXZebraShutter
 from dodal.devices.zocalo import ZocaloResults
 from dodal.devices.zocalo.zocalo_results import ZocaloSource
 from dodal.log import set_beamline as set_log_beamline
@@ -120,8 +120,8 @@ def beamstop(config_client: ConfigClient) -> Beamstop:
 
 
 @devices.factory()
-def sample_shutter() -> ZebraShutter:
-    return ZebraShutter(f"{PREFIX.beamline_prefix}-EA-SHTR-01:")
+def sample_shutter() -> MXZebraShutter:
+    return MXZebraShutter(f"{PREFIX.beamline_prefix}-EA-SHTR-01:")
 
 
 @devices.factory()

src/dodal/beamlines/i15_1.py

@@ -18,6 +18,7 @@
 from dodal.devices.motors import XYPhiStage, XYStage, XYZStage, YZStage
 from dodal.devices.slits import Slits
 from dodal.devices.synchrotron import Synchrotron
+from dodal.devices.zebra.zebra_controlled_shutter import ZebraFastShutter
 from dodal.log import set_beamline as set_log_beamline
 from dodal.utils import BeamlinePrefix, get_beamline_name
 
@@ -221,3 +222,11 @@ def gonio_interlock() -> GonioInterlock:
     return GonioInterlock(
         bl_prefix=PREFIX.beamline_prefix, interlock_suffix="-VA-OMRON-01:INT3:ILK"
     )
+
+
+@devices.factory()
+def fast_shutter() -> ZebraFastShutter:
+    return ZebraFastShutter(
+        set_pv=f"{PREFIX.beamline_prefix}-EA-ZEBRA-01:SOFT_IN:B3",
+        get_pv=f"{PREFIX.beamline_prefix}-EA-ZEBRA-01:OUT4_TTL:STA",
+    )

src/dodal/beamlines/i23.py

@@ -21,7 +21,7 @@
     ZebraSources,
     ZebraTTLOutputs,
 )
-from dodal.devices.zebra.zebra_controlled_shutter import ZebraShutter
+from dodal.devices.zebra.zebra_controlled_shutter import MXZebraShutter
 from dodal.log import set_beamline as set_log_beamline
 from dodal.utils import BeamlinePrefix, get_beamline_name, get_hostname
 
@@ -83,8 +83,8 @@ def pin_tip_detection() -> PinTipDetection:
 
 
 @devices.factory()
-def shutter() -> ZebraShutter:
-    return ZebraShutter(f"{PREFIX.beamline_prefix}-EA-SHTR-01:")
+def shutter() -> MXZebraShutter:
+    return MXZebraShutter(f"{PREFIX.beamline_prefix}-EA-SHTR-01:")
 
 
 @devices.factory()

src/dodal/beamlines/i24.py

@@ -33,7 +33,7 @@
     ZebraSources,
     ZebraTTLOutputs,
 )
-from dodal.devices.zebra.zebra_controlled_shutter import ZebraShutter
+from dodal.devices.zebra.zebra_controlled_shutter import MXZebraShutter
 from dodal.log import set_beamline as set_log_beamline
 from dodal.utils import BeamlinePrefix, get_beamline_name
 
@@ -173,7 +173,7 @@ def synchrotron() -> Synchrotron:
 
 
 @devices.factory()
-def sample_shutter() -> ZebraShutter:
-    return ZebraShutter(
+def sample_shutter() -> MXZebraShutter:
+    return MXZebraShutter(
         f"{PREFIX.beamline_prefix}-EA-SHTR-01:",
     )

src/dodal/devices/fast_shutter.py

@@ -9,6 +9,7 @@
     SignalRW,
     StandardReadable,
     StandardReadableFormat,
+    StrictEnum,
     derived_signal_rw,
     soft_signal_r_and_setter,
 )
@@ -19,6 +20,11 @@
 EnumTypesT = TypeVar("EnumTypesT", bound=EnumTypes)
 
 
+class OpenClose(StrictEnum):
+    OPEN = "Open"
+    CLOSE = "Close"
+
+
 class GenericFastShutter(StandardReadable, Movable[EnumTypesT], Generic[EnumTypesT]):
     """Enum device specialised for a fast shutter with configured open_state and
     close_state so it is generic enough to be used with any device or plan without
@@ -111,15 +117,15 @@ def _create_shutter_state(self):
 
 class DualFastShutter(GenericFastShutter[EnumTypesT], Generic[EnumTypesT]):
     """A fast shutter device that handles the positions of two other fast shutters. The
-    "active" shutter is the one that corrosponds to the selected_shutter signal. For
+    "active" shutter is the one that corresponds to the selected_shutter signal. For
     example, active shutter is shutter1 if selected_source is at SelectedSource.SOURCE1
     and vise versa for shutter2 and SelectedSource.SOURCE2. Whenever a move is done on
     this device, the inactive shutter is always set to the close_state.
 
     Args:
-        shutter1 (GenericFastShutter): Active shutter that corrosponds to
+        shutter1 (GenericFastShutter): Active shutter that corresponds to
             SelectedSource.SOURCE1.
-        shutter2 (GenericFastShutter): Active shutter that corrosponds to
+        shutter2 (GenericFastShutter): Active shutter that corresponds to
             SelectedSource.SOURCE2.
         selected_source (SignalRW): Signal that decides the active shutter.
         name (str, optional): Name of this device.

src/dodal/devices/zebra/zebra_controlled_shutter.py

@@ -2,12 +2,22 @@
 from ophyd_async.core import (
     DEFAULT_TIMEOUT,
     AsyncStatus,
+    DeviceMock,
+    SignalRW,
     StandardReadable,
     StrictEnum,
+    YesNo,
+    callback_on_mock_put,
+    default_mock_class,
+    derived_signal_rw,
+    set_and_wait_for_other_value,
+    set_mock_value,
     wait_for_value,
 )
 from ophyd_async.epics.core import epics_signal_r, epics_signal_rw, epics_signal_w
 
+from dodal.devices.fast_shutter import GenericFastShutter, OpenClose
+
 
 class ZebraShutterState(StrictEnum):
     CLOSE = "Close"
@@ -19,7 +29,7 @@ class ZebraShutterControl(StrictEnum):
     AUTO = "Auto"
 
 
-class ZebraShutter(StandardReadable, Movable[ZebraShutterState]):
+class MXZebraShutter(StandardReadable, Movable[ZebraShutterState]):
     """The shutter on most MX beamlines is controlled by the zebra.
 
     Internally in the zebra there are two AND gates, one for manual control and one for
@@ -51,3 +61,49 @@ async def set(self, value: ZebraShutterState):
             match=value,
             timeout=DEFAULT_TIMEOUT,
         )
+
+
+class MockZebraFastShutter(DeviceMock["ZebraFastShutter"]):
+    async def connect(self, device: "ZebraFastShutter") -> None:
+        callback_on_mock_put(
+            device._set_pv,  # noqa: SLF001
+            lambda state, *_, **__: set_mock_value(
+                device._get_pv,  # noqa: SLF001
+                1 if state == YesNo.YES else 0,
+            ),
+        )
+
+
+@default_mock_class(MockZebraFastShutter)
+class ZebraFastShutter(GenericFastShutter[OpenClose]):
+    """A fast shutter controlled by the zebra that doesn't have the automatic/manual
+    protection on top that the MXZebraShutter does. See https://jira.diamond.ac.uk/browse/I15_1-1626
+    to bring them in line.
+    """
+
+    def __init__(
+        self,
+        set_pv: str,
+        get_pv: str,
+        name: str = "",
+    ):
+        self._set_pv = epics_signal_w(YesNo, set_pv)
+        self._get_pv = epics_signal_r(int, get_pv)
+        super().__init__(OpenClose.OPEN, OpenClose.CLOSE, name)
+
+    def _create_shutter_state(self) -> SignalRW[OpenClose]:
+        return derived_signal_rw(
+            self._read_shutter_state,
+            self._set_shutter_state,
+            get_pv=self._get_pv,
+        )
+
+    def _read_shutter_state(self, get_pv: int) -> OpenClose:
+        return OpenClose.CLOSE if get_pv == 0 else OpenClose.OPEN
+
+    async def _set_shutter_state(self, value: OpenClose):
+        set_value = YesNo.YES if value == OpenClose.OPEN else YesNo.NO
+        readback_value = 1 if value == OpenClose.OPEN else 0
+        await set_and_wait_for_other_value(
+            self._set_pv, set_value, self._get_pv, readback_value
+        )

tests/devices/test_zebra_shutter.py

@@ -1,8 +1,17 @@
 import pytest
-from ophyd_async.core import callback_on_mock_put, init_devices, set_mock_value
+from ophyd_async.core import (
+    YesNo,
+    callback_on_mock_put,
+    get_mock_put,
+    init_devices,
+    set_mock_value,
+)
+from ophyd_async.testing import assert_reading, partial_reading
 
+from dodal.devices.fast_shutter import OpenClose
 from dodal.devices.zebra.zebra_controlled_shutter import (
-    ZebraShutter,
+    MXZebraShutter,
+    ZebraFastShutter,
     ZebraShutterControl,
     ZebraShutterState,
 )
@@ -11,7 +20,7 @@
 @pytest.fixture
 async def sim_shutter():
     async with init_devices(mock=True):
-        sim_shutter = ZebraShutter(
+        sim_shutter = MXZebraShutter(
             prefix="sim_shutter",
             name="shutter",
         )
@@ -25,7 +34,7 @@ def propagate_status(value: ZebraShutterState, *args, **kwargs):
 
 @pytest.mark.parametrize("new_state", [ZebraShutterState.OPEN, ZebraShutterState.CLOSE])
 async def test_set_shutter_open(
-    sim_shutter: ZebraShutter, new_state: ZebraShutterState
+    sim_shutter: MXZebraShutter, new_state: ZebraShutterState
 ):
     await sim_shutter.set(new_state)
     reading = await sim_shutter.read()
@@ -36,7 +45,80 @@ async def test_set_shutter_open(
     )
 
 
-async def test_given_shutter_in_auto_then_when_set_raises(sim_shutter: ZebraShutter):
+async def test_given_shutter_in_auto_then_when_set_raises(sim_shutter: MXZebraShutter):
     set_mock_value(sim_shutter.control_mode, ZebraShutterControl.AUTO)
     with pytest.raises(UserWarning):
         await sim_shutter.set(ZebraShutterState.OPEN)
+
+
+@pytest.fixture
+def int_fast_shutter() -> ZebraFastShutter:
+    with init_devices(mock=True):
+        shutter = ZebraFastShutter(set_pv="SET", get_pv="GET")
+    return shutter
+
+
+@pytest.mark.parametrize(
+    "pv_value, expected_reading",
+    [
+        [0, OpenClose.CLOSE],
+        [1, OpenClose.OPEN],
+    ],
+)
+async def test_given_fast_shutter_pv_at_int_then_reads_expected_enum(
+    int_fast_shutter: ZebraFastShutter, pv_value: int, expected_reading: OpenClose
+):
+    set_mock_value(int_fast_shutter._get_pv, pv_value)
+
+    await assert_reading(
+        int_fast_shutter,
+        {
+            f"{int_fast_shutter.name}-shutter_state": partial_reading(expected_reading),
+        },
+    )
+
+
+@pytest.mark.parametrize(
+    "fast_shutter_state, expected_pv_value",
+    [
+        [OpenClose.CLOSE, YesNo.NO],
+        [OpenClose.OPEN, YesNo.YES],
+    ],
+)
+async def test_when_fast_shutter_state_changed_then_pv_set_correctly(
+    int_fast_shutter: ZebraFastShutter,
+    fast_shutter_state: OpenClose,
+    expected_pv_value: int,
+):
+    await int_fast_shutter.set(fast_shutter_state)
+
+    mock = get_mock_put(int_fast_shutter._set_pv)
+    mock.assert_called_once_with(expected_pv_value)
+
+
+@pytest.mark.parametrize(
+    "initial_readback_pv_state, initial_setpoint_pv_state, set_fast_shutter_state",
+    [
+        [1, YesNo.YES, OpenClose.CLOSE],
+        [0, YesNo.NO, OpenClose.OPEN],
+    ],
+)
+async def test_when_fast_shutter_state_changed_then_pv_readback_correct(
+    int_fast_shutter: ZebraFastShutter,
+    initial_readback_pv_state: int,
+    initial_setpoint_pv_state: YesNo,
+    set_fast_shutter_state: OpenClose,
+):
+    set_mock_value(int_fast_shutter._get_pv, initial_readback_pv_state)
+    set_mock_value(int_fast_shutter._set_pv, initial_setpoint_pv_state)
+
+    await int_fast_shutter.set(set_fast_shutter_state)
+
+    await assert_reading(
+        int_fast_shutter,
+        {
+            f"{int_fast_shutter.name}-shutter_state": partial_reading(
+                set_fast_shutter_state
+            ),
+        },
+    )