Adding Pimte area detector devices

src/dodal/devices/areadetector/epics/drivers/pimte1_driver.py

@@ -0,0 +1,28 @@
+from enum import Enum
+
+from ophyd_async.epics.areadetector.drivers.ad_base import ADBase
+from ophyd_async.epics.areadetector.utils import ad_r, ad_rw
+from ophyd_async.epics.signal import epics_signal_rw
+
+
+class Pimte1Driver(ADBase):
+    def __init__(self, prefix: str) -> None:
+        self.trigger_mode = ad_rw(Pimte1Driver.TriggerMode, prefix + "TriggerMode")
+        self.initialize = ad_rw(int, prefix + "Initialize")
+        self.set_temperture = epics_signal_rw(float, prefix + "SetTemperature")
+        self.read_backtemperture = ad_r(float, prefix + "MeasuredTemperature")
+        self.speed = ad_rw(Pimte1Driver.SpeedMode, prefix + "SpeedSelection")
+        super().__init__(prefix)
+
+    class SpeedMode(str, Enum):
+        adc_50Khz = "0: 50 KHz - 20000 ns"
+        adc_100Khz = "1: 100 kHz - 10000 ns"
+        adc_200Khz = "2: 200 kHz - 5000 ns"
+        adc_500Khz = "3: 500 kHz - 2000 ns"
+        adc_1Mhz = "4: 1 MHz - 1000 ns"
+        adc_2Mhz = "5: 2 MHz - 500 ns"
+
+    class TriggerMode(str, Enum):
+        internal = "Free Run"
+        ext_trigger = "Ext Trigger"
+        bulb_mode = "Bulb Mode"

src/dodal/devices/areadetector/epics/pimte_controller.py

@@ -0,0 +1,65 @@
+import asyncio
+from typing import Optional, Set  # , Set
+
+from ophyd_async.core import AsyncStatus, DetectorControl, DetectorTrigger
+from ophyd_async.epics.areadetector.drivers.ad_base import (
+    DEFAULT_GOOD_STATES,
+    DetectorState,
+    start_acquiring_driver_and_ensure_status,
+)
+from ophyd_async.epics.areadetector.utils import ImageMode, stop_busy_record
+
+from dodal.devices.areadetector.epics.drivers.pimte1_driver import Pimte1Driver
+
+TRIGGER_MODE = {
+    DetectorTrigger.internal: Pimte1Driver.TriggerMode.internal,
+    DetectorTrigger.constant_gate: Pimte1Driver.TriggerMode.ext_trigger,
+    DetectorTrigger.variable_gate: Pimte1Driver.TriggerMode.ext_trigger,
+}
+
+
+class PimteController(DetectorControl):
+    def __init__(
+        self,
+        driver: Pimte1Driver,
+        good_states: Set[DetectorState] = set(DEFAULT_GOOD_STATES),
+    ) -> None:
+        self.driver = driver
+        self.good_states = good_states
+
+    def get_deadtime(self, exposure: float) -> float:
+        return exposure + 0.1
+
+    async def _process_setting(self) -> None:
+        await self.driver.initialize.set(1)
+
+    async def set_temperature(self, temperature: float) -> None:
+        await self.driver.set_temperture.set(temperature)
+        await self._process_setting()
+
+    async def set_speed(self, speed: Pimte1Driver.SpeedMode) -> None:
+        await self.driver.speed.set(speed)
+        await self._process_setting()
+
+    async def arm(
+        self,
+        num: int = 1,
+        trigger: DetectorTrigger = DetectorTrigger.internal,
+        exposure: Optional[float] = None,
+    ) -> AsyncStatus:
+        funcs = [
+            self.driver.num_images.set(999_999 if num == 0 else num),
+            self.driver.image_mode.set(ImageMode.multiple),
+            self.driver.trigger_mode.set(TRIGGER_MODE[trigger]),
+        ]
+        if exposure is not None:
+            funcs.append(self.driver.acquire_time.set(exposure))
+
+        await asyncio.gather(*funcs)
+        await self._process_setting()
+        return await start_acquiring_driver_and_ensure_status(
+            self.driver, good_states=self.good_states
+        )
+
+    async def disarm(self):
+        await stop_busy_record(self.driver.acquire, False, timeout=1)

src/dodal/devices/areadetector/pimteAD.py

@@ -0,0 +1,44 @@
+from typing import Sequence
+from bluesky.protocols import Hints
+from ophyd_async.core import DirectoryProvider, SignalR, StandardDetector
+from ophyd_async.epics.areadetector.drivers import ADBaseShapeProvider
+from ophyd_async.epics.areadetector.writers import HDFWriter, NDFileHDF, NDPluginStats
+
+from dodal.devices.areadetector.epics.drivers.pimte1_driver import Pimte1Driver
+from dodal.devices.areadetector.epics.pimte_controller import PimteController
+
+
+class HDFStatsPimte(StandardDetector):
+    _controller: PimteController
+    _writer: HDFWriter
+
+    def __init__(
+        self,
+        prefix: str,
+        directory_provider: DirectoryProvider,
+        name: str,
+        config_sigs: Sequence[SignalR] = (),
+        **scalar_sigs: str,
+    ):
+        self.drv = Pimte1Driver(prefix + "CAM:")
+        self.hdf = NDFileHDF(prefix + "HDF5:")
+        self.stats = NDPluginStats(prefix + "STAT:")
+        # taken from i22 but this does nothing atm
+
+        super().__init__(
+            PimteController(self.drv),
+            HDFWriter(
+                self.hdf,
+                directory_provider,
+                lambda: self.name,
+                ADBaseShapeProvider(self.drv),
+                sum="StatsTotal",
+                **scalar_sigs,
+            ),
+            config_sigs=config_sigs,
+            name=name,
+        )
+
+    @property
+    def hints(self) -> Hints:
+        return self._writer.hints

tests/devices/unit_tests/test_pimte.py

@@ -0,0 +1,76 @@
+import bluesky.plan_stubs as bps
+import pytest
+from bluesky.run_engine import RunEngine
+from bluesky.utils import new_uid
+from ophyd_async.core import DeviceCollector, StaticDirectoryProvider, set_sim_value
+
+from dodal.devices.areadetector.pimteAD import HDFStatsPimte
+
+CURRENT_DIRECTORY = "."  # str(Path(__file__).parent)
+
+
+async def make_detector(prefix: str = "") -> HDFStatsPimte:
+    dp = StaticDirectoryProvider(CURRENT_DIRECTORY, f"test-{new_uid()}")
+
+    async with DeviceCollector(sim=True):
+        detector = HDFStatsPimte(prefix, dp, "pimte")
+    return detector
+
+
+def count_sim(det: HDFStatsPimte, times: int = 1):
+    """Test plan to do the equivalent of bp.count for a sim detector."""
+
+    yield from bps.stage_all(det)
+    yield from bps.open_run()
+    yield from bps.declare_stream(det, name="primary", collect=False)
+    for _ in range(times):
+        read_value = yield from bps.rd(det._writer.hdf.num_captured)
+        yield from bps.trigger(det, wait=False, group="wait_for_trigger")
+
+        yield from bps.sleep(0.001)
+        set_sim_value(det._writer.hdf.num_captured, read_value + 1)
+
+        yield from bps.wait(group="wait_for_trigger")
+        yield from bps.create()
+        yield from bps.read(det)
+        yield from bps.save()
+
+    yield from bps.close_run()
+    yield from bps.unstage_all(det)
+
+
+@pytest.fixture
+async def single_detector(RE: RunEngine) -> HDFStatsPimte:
+    detector = await make_detector(prefix="TEST")
+
+    set_sim_value(detector._controller.driver.array_size_x, 10)
+    set_sim_value(detector._controller.driver.array_size_y, 20)
+    set_sim_value(detector.hdf.file_path_exists, True)
+    set_sim_value(detector._writer.hdf.num_captured, 0)
+    return detector
+
+
+async def test_pimte(RE: RunEngine, single_detector: HDFStatsPimte):
+    names = []
+    docs = []
+    RE.subscribe(lambda name, _: names.append(name))
+    RE.subscribe(lambda _, doc: docs.append(doc))
+
+    RE(count_sim(single_detector))
+    writer = single_detector._writer
+
+    assert (
+        await writer.hdf.file_path.get_value()
+        == writer._directory_provider().root.as_posix()
+    )
+
+    assert names == [
+        "start",
+        "descriptor",
+        "stream_resource",
+        "stream_resource",
+        "stream_datum",
+        "stream_datum",
+        "event",
+        "stop",
+    ]

tests/devices/unit_tests/test_pimte1Driver.py

@@ -0,0 +1,121 @@
+import pytest
+from ophyd_async.core import DeviceCollector
+
+from dodal.devices.areadetector.epics.drivers.pimte1_driver import Pimte1Driver
+
+# Long enough for multiple asyncio event loop cycles to run so
+# all the tasks have a chance to run
+A_BIT = 0.001
+
+
+@pytest.fixture
+async def sim_pimte_driver():
+    async with DeviceCollector(sim=True):
+        sim_pimte_driver = Pimte1Driver("BLxxI-MO-TABLE-01:X")
+        # Signals connected here
+
+    assert sim_pimte_driver.name == "sim_pimte_driver"
+    yield sim_pimte_driver
+
+
+async def test_sim_pimte_driver(sim_pimte_driver: Pimte1Driver) -> None:
+    pass
+
+
+"""
+async def test_motor_moving_well(sim_motor: motor.Motor) -> None:
+    set_sim_put_proceeds(sim_motor.setpoint, False)
+    s = sim_motor.set(0.55)
+    watcher = Mock()
+    s.watch(watcher)
+    done = Mock()
+    s.add_callback(done)
+    await asyncio.sleep(A_BIT)
+    assert watcher.call_count == 1
+    assert watcher.call_args == call(
+        name="sim_motor",
+        current=0.0,
+        initial=0.0,
+        target=0.55,
+        unit="mm",
+        precision=3,
+        time_elapsed=pytest.approx(0.0, abs=0.05),
+    )
+    watcher.reset_mock()
+    assert 0.55 == await sim_motor.setpoint.get_value()
+    assert not s.done
+    await asyncio.sleep(0.1)
+    set_sim_value(sim_motor.readback, 0.1)
+    assert watcher.call_count == 1
+    assert watcher.call_args == call(
+        name="sim_motor",
+        current=0.1,
+        initial=0.0,
+        target=0.55,
+        unit="mm",
+        precision=3,
+        time_elapsed=pytest.approx(0.1, abs=0.05),
+    )
+    set_sim_put_proceeds(sim_motor.setpoint, True)
+    await asyncio.sleep(A_BIT)
+    assert s.done
+    done.assert_called_once_with(s)
+
+
+async def test_motor_moving_stopped(sim_motor: motor.Motor):
+    set_sim_put_proceeds(sim_motor.setpoint, False)
+    s = sim_motor.set(1.5)
+    s.add_callback(Mock())
+    await asyncio.sleep(0.2)
+    assert not s.done
+    await sim_motor.stop()
+    set_sim_put_proceeds(sim_motor.setpoint, True)
+    await asyncio.sleep(A_BIT)
+    assert s.done
+    assert s.success is False
+
+
+async def test_read_motor(sim_motor: motor.Motor):
+    sim_motor.stage()
+    assert (await sim_motor.read())["sim_motor"]["value"] == 0.0
+    assert (await sim_motor.describe())["sim_motor"][
+        "source"
+    ] == "sim://BLxxI-MO-TABLE-01:X.RBV"
+    assert (await sim_motor.read_configuration())["sim_motor-velocity"]["value"] == 1
+    assert (await sim_motor.describe_configuration())["sim_motor-units"]["shape"] == []
+    set_sim_value(sim_motor.readback, 0.5)
+    assert (await sim_motor.read())["sim_motor"]["value"] == 0.5
+    sim_motor.unstage()
+    # Check we can still read and describe when not staged
+    set_sim_value(sim_motor.readback, 0.1)
+    assert (await sim_motor.read())["sim_motor"]["value"] == 0.1
+    assert await sim_motor.describe()
+
+
+async def test_set_velocity(sim_motor: motor.Motor) -> None:
+    v = sim_motor.velocity
+    assert (await v.describe())["sim_motor-velocity"][
+        "source"
+    ] == "sim://BLxxI-MO-TABLE-01:X.VELO"
+    q: asyncio.Queue[Dict[str, Reading]] = asyncio.Queue()
+    v.subscribe(q.put_nowait)
+    assert (await q.get())["sim_motor-velocity"]["value"] == 1.0
+    await v.set(2.0)
+    assert (await q.get())["sim_motor-velocity"]["value"] == 2.0
+    v.clear_sub(q.put_nowait)
+    await v.set(3.0)
+    assert (await v.read())["sim_motor-velocity"]["value"] == 3.0
+    assert q.empty()
+
+
+def test_motor_in_re(sim_motor: motor.Motor, RE: RunEngine) -> None:
+    sim_motor.move(0)
+
+    def my_plan():
+        sim_motor.move(0)
+        return
+        yield
+
+    with pytest.raises(RuntimeError, match="Will deadlock run engine if run in a plan"):
+        RE(my_plan())
+"""

tests/devices/unit_tests/test_pimteController.py

@@ -0,0 +1,55 @@
+from unittest.mock import patch
+
+import pytest
+from ophyd_async.core import DetectorTrigger, DeviceCollector
+from ophyd_async.epics.areadetector.controllers import (
+    ADSimController,
+)
+from ophyd_async.epics.areadetector.drivers import ADBase
+from ophyd_async.epics.areadetector.utils import ImageMode
+
+from dodal.devices.areadetector.epics.drivers.pimte1_driver import Pimte1Driver
+from dodal.devices.areadetector.epics.pimte_controller import PimteController
+
+
+@pytest.fixture
+async def pimte(RE) -> PimteController:
+    async with DeviceCollector(sim=True):
+        drv = Pimte1Driver("DRIVER:")
+        controller = PimteController(drv)
+
+    return controller
+
+
+@pytest.fixture
+async def ad(RE) -> ADSimController:
+    async with DeviceCollector(sim=True):
+        drv = ADBase("DRIVER:")
+        controller = ADSimController(drv)
+
+    return controller
+
+
+async def test_pimte_controller(RE, pimte: PimteController):
+    with patch("ophyd_async.core.signal.wait_for_value", return_value=None):
+        await pimte.arm(num=1, exposure=0.002, trigger=DetectorTrigger.internal)
+
+    driver = pimte.driver
+
+    assert await driver.num_images.get_value() == 1
+    assert await driver.image_mode.get_value() == ImageMode.multiple
+    assert await driver.trigger_mode.get_value() == Pimte1Driver.TriggerMode.internal
+    assert await driver.acquire.get_value() is True
+    assert await driver.acquire_time.get_value() == 0.002
+    assert pimte.get_deadtime(2) == 2 + 0.1
+
+    with patch(
+        "ophyd_async.epics.areadetector.utils.wait_for_value", return_value=None
+    ):
+        await pimte.disarm()
+        await pimte.set_temperature(20)
+        await pimte.set_speed(driver.SpeedMode.adc_200Khz)
+    assert await driver.set_temperture.get_value() == 20
+    assert await driver.speed.get_value() == driver.SpeedMode.adc_200Khz
+
+    assert await driver.acquire.get_value() is False