Add fswitch device for i22 (#469)

This is an initial implementation for providing the required data for
i22 Nexus files.

In the future this device should be combined with the i04 version.

See #399 for further details.

src/dodal/beamlines/i22.py

@@ -5,6 +5,7 @@
 )
 from dodal.beamlines.beamline_utils import set_beamline as set_utils_beamline
 from dodal.common.visit import StaticVisitDirectoryProvider
+from dodal.devices.i22.fswitch import FSwitch
 from dodal.devices.slits import Slits
 from dodal.devices.tetramm import TetrammDetector
 from dodal.log import set_beamline as set_log_beamline
@@ -120,3 +121,16 @@ def slits_6(
         wait_for_connection,
         fake_with_ophyd_sim,
     )
+
+
+def fswitch(
+    wait_for_connection: bool = True,
+    fake_with_ophyd_sim: bool = False,
+) -> FSwitch:
+    return device_instantiation(
+        FSwitch,
+        "fswitch",
+        "-MO-FSWT-01:",
+        wait_for_connection,
+        fake_with_ophyd_sim,
+    )

src/dodal/devices/i22/fswitch.py

@@ -0,0 +1,56 @@
+import asyncio
+import time
+from enum import Enum
+from typing import Dict
+
+from bluesky.protocols import Reading
+from ophyd_async.core import StandardReadable
+from ophyd_async.core.device import DeviceVector
+from ophyd_async.epics.signal import epics_signal_r
+
+
+class FilterState(str, Enum):
+    """
+    Note that the in/out here refers to the internal rocker
+    position so a PV value of IN implies a filter OUT of beam
+    """
+
+    IN_BEAM = "OUT"
+    OUT_BEAM = "IN"
+
+
+class FSwitch(StandardReadable):
+    """
+    Device for i22's fswitch. A filter switch for manipulating
+    compound refractive lenses. Also referred to as a transfocator.
+
+    This currently only implements the minimum
+    functionality for retrieving the number of lenses inserted.
+
+    Eventually this should be combined with the transfocator device in the i04
+    module but is currently incompatible as the Epics interfaces are different.
+    See https://github.com/DiamondLightSource/dodal/issues/399
+
+    """
+
+    NUM_FILTERS = 128
+
+    def __init__(self, prefix: str, name: str = "") -> None:
+        self.filters = DeviceVector(
+            {
+                i: epics_signal_r(FilterState, f"{prefix}FILTER-{i:03}:STATUS_RBV")
+                for i in range(FSwitch.NUM_FILTERS)
+            }
+        )
+        super().__init__(name)
+
+    async def read(self) -> Dict[str, Reading]:
+        result = await asyncio.gather(
+            *(filter.get_value() for filter in self.filters.values())
+        )
+        num_in = sum(r.value == FilterState.IN_BEAM for r in result)
+        default_reading = await super().read()
+        return {
+            "number_of_lenses": Reading(value=num_in, timestamp=time.time()),
+            **default_reading,
+        }

tests/devices/i22/test_fswitch.py

@@ -0,0 +1,28 @@
+from unittest import mock
+
+import pytest
+from ophyd_async.core import DeviceCollector, set_sim_value
+
+from dodal.devices.i22.fswitch import FilterState, FSwitch
+
+
+@pytest.fixture
+async def fswitch() -> FSwitch:
+    async with DeviceCollector(sim=True):
+        fswitch = FSwitch("DEMO-FSWT-01:")
+
+    return fswitch
+
+
+async def test_reading_fswitch(fswitch: FSwitch):
+    set_sim_value(fswitch.filters.get(0), FilterState.OUT_BEAM)
+    set_sim_value(fswitch.filters.get(1), FilterState.OUT_BEAM)
+    set_sim_value(fswitch.filters.get(2), FilterState.OUT_BEAM)
+
+    reading = await fswitch.read()
+    assert reading == {
+        "number_of_lenses": {
+            "timestamp": mock.ANY,
+            "value": 125,  # three filters out
+        }
+    }