Port andor new ad format

src/ophyd_async/epics/adandor/__init__.py

@@ -0,0 +1,9 @@
+from ._andor import Andor2Detector
+from ._andor_controller import Andor2Controller
+from ._andor_io import Andor2DriverIO
+
+__all__ = [
+    "Andor2Detector",
+    "Andor2Controller",
+    "Andor2DriverIO",
+]

src/ophyd_async/epics/adandor/_andor.py

@@ -0,0 +1,45 @@
+from collections.abc import Sequence
+
+from ophyd_async.core import PathProvider
+from ophyd_async.core._signal import SignalR
+from ophyd_async.epics import adcore
+
+from ._andor_controller import Andor2Controller
+from ._andor_io import Andor2DriverIO
+
+
+class Andor2Detector(adcore.AreaDetector[Andor2Controller]):
+    """
+    Andor 2 area detector device (CCD detector 56fps with full chip readout).
+    Andor model:DU897_BV.
+    """
+
+    def __init__(
+        self,
+        prefix: str,
+        path_provider: PathProvider,
+        drv_suffix="cam1:",
+        writer_cls: type[adcore.ADWriter] = adcore.ADHDFWriter,
+        fileio_suffix: str | None = None,
+        name: str = "",
+        config_sigs: Sequence[SignalR] = (),
+        plugins: dict[str, adcore.NDPluginBaseIO] | None = None,
+    ):
+        driver = Andor2DriverIO(prefix + drv_suffix)
+        controller = Andor2Controller(driver)
+
+        writer = writer_cls.with_io(
+            prefix,
+            path_provider,
+            dataset_source=driver,
+            fileio_suffix=fileio_suffix,
+            plugins=plugins,
+        )
+
+        super().__init__(
+            controller=controller,
+            writer=writer,
+            plugins=plugins,
+            name=name,
+            config_sigs=config_sigs,
+        )

src/ophyd_async/epics/adandor/_andor_controller.py

@@ -0,0 +1,49 @@
+import asyncio
+
+from ophyd_async.core import (
+    DetectorTrigger,
+    TriggerInfo,
+)
+from ophyd_async.epics import adcore
+
+from ._andor_io import Andor2DriverIO, Andor2ImageMode, Andor2TriggerMode
+
+_MIN_DEAD_TIME = 0.1
+_MAX_NUM_IMAGE = 999_999
+
+
+class Andor2Controller(adcore.ADBaseController[Andor2DriverIO]):
+    def __init__(
+        self,
+        driver: Andor2DriverIO,
+        good_states: frozenset[adcore.DetectorState] = adcore.DEFAULT_GOOD_STATES,
+    ) -> None:
+        super().__init__(driver, good_states=good_states)
+
+    def get_deadtime(self, exposure: float | None) -> float:
+        return _MIN_DEAD_TIME + (exposure or 0)
+
+    async def prepare(self, trigger_info: TriggerInfo):
+        await self.set_exposure_time_and_acquire_period_if_supplied(
+            trigger_info.livetime
+        )
+        await asyncio.gather(
+            self.driver.trigger_mode.set(self._get_trigger_mode(trigger_info.trigger)),
+            self.driver.num_images.set(
+                trigger_info.total_number_of_triggers or _MAX_NUM_IMAGE
+            ),
+            self.driver.image_mode.set(Andor2ImageMode.MULTIPLE),
+        )
+
+    def _get_trigger_mode(self, trigger: DetectorTrigger) -> Andor2TriggerMode:
+        supported_trigger_types = {
+            DetectorTrigger.INTERNAL: Andor2TriggerMode.INTERNAL,
+            DetectorTrigger.EDGE_TRIGGER: Andor2TriggerMode.EXT_TRIGGER,
+        }
+        if trigger not in supported_trigger_types:
+            raise ValueError(
+                f"{self.__class__.__name__} only supports the following trigger "
+                f"types: {supported_trigger_types} but was asked to "
+                f"use {trigger}"
+            )
+        return supported_trigger_types[trigger]

src/ophyd_async/epics/adandor/_andor_io.py

@@ -0,0 +1,45 @@
+from ophyd_async.core import StrictEnum, SubsetEnum
+from ophyd_async.epics.adcore import ADBaseIO
+from ophyd_async.epics.core import (
+    epics_signal_r,
+    epics_signal_rw,
+    epics_signal_rw_rbv,
+)
+
+
+class Andor2TriggerMode(StrictEnum):
+    INTERNAL = "Internal"
+    EXT_TRIGGER = "External"
+    EXT_START = "External Start"
+    EXT_EXPOSURE = "External Exposure"
+    EXT_FVP = "External FVP"
+    SOFTWARE = "Software"
+
+
+class Andor2ImageMode(StrictEnum):
+    SINGLE = "Single"
+    MULTIPLE = "Multiple"
+    CONTINUOUS = "Continuous"
+    FAST_KINETICS = "Fast Kinetics"
+
+
+class Andor2DataType(SubsetEnum):
+    UINT16 = "UInt16"
+    UINT32 = "UInt32"
+    FLOAT32 = "Float32"
+    FLOAT64 = "Float64"
+
+
+class Andor2DriverIO(ADBaseIO):
+    """
+    Epics pv for andor model:DU897_BV as deployed on p99
+    """
+
+    def __init__(self, prefix: str, name: str = "") -> None:
+        super().__init__(prefix, name=name)
+        self.trigger_mode = epics_signal_rw(Andor2TriggerMode, prefix + "TriggerMode")
+        self.data_type = epics_signal_r(Andor2DataType, prefix + "DataType_RBV")
+        self.andor_accumulate_period = epics_signal_r(
+            float, prefix + "AndorAccumulatePeriod_RBV"
+        )
+        self.image_mode = epics_signal_rw_rbv(Andor2ImageMode, prefix + "ImageMode")

tests/epics/adandor/test_andor.py

@@ -0,0 +1,123 @@
+from typing import cast
+from unittest.mock import AsyncMock, patch
+
+import pytest
+from event_model import StreamDatum, StreamResource
+
+from ophyd_async.core import (
+    DetectorTrigger,
+    PathProvider,
+    TriggerInfo,
+)
+from ophyd_async.epics import adandor
+
+
+@pytest.fixture
+def test_adandor(ad_standard_det_factory) -> adandor.Andor2Detector:
+    return ad_standard_det_factory(adandor.Andor2Detector)
+
+
+@pytest.mark.parametrize("exposure_time", [0.0, 0.1, 1.0, 10.0, 100.0])
+async def test_deadtime_from_exposure_time(
+    exposure_time: float,
+    test_adandor: adandor.Andor2Detector,
+):
+    assert test_adandor._controller.get_deadtime(exposure_time) == exposure_time + 0.1
+
+
+async def test_hints_from_hdf_writer(test_adandor: adandor.Andor2Detector):
+    assert test_adandor.hints == {"fields": ["test_adandor21"]}
+
+
+async def test_can_read(test_adandor: adandor.Andor2Detector):
+    # Standard detector can be used as Readable
+    assert (await test_adandor.read()) == {}
+
+
+async def test_decribe_describes_writer_dataset(
+    test_adandor: adandor.Andor2Detector, one_shot_trigger_info: TriggerInfo
+):
+    assert await test_adandor.describe() == {}
+    await test_adandor.stage()
+    await test_adandor.prepare(one_shot_trigger_info)
+    assert await test_adandor.describe() == {
+        "test_adandor21": {
+            "source": "mock+ca://ANDOR21:HDF1:FullFileName_RBV",
+            "shape": [10, 10],
+            "dtype": "array",
+            "dtype_numpy": "<u2",
+            "external": "STREAM:",
+        }
+    }
+
+
+async def test_can_collect(
+    test_adandor: adandor.Andor2Detector,
+    static_path_provider: PathProvider,
+    one_shot_trigger_info: TriggerInfo,
+):
+    path_info = static_path_provider()
+    full_file_name = path_info.directory_path / f"{path_info.filename}.h5"
+    await test_adandor.stage()
+    await test_adandor.prepare(one_shot_trigger_info)
+    docs = [(name, doc) async for name, doc in test_adandor.collect_asset_docs(1)]
+    assert len(docs) == 2
+    assert docs[0][0] == "stream_resource"
+    stream_resource = cast(StreamResource, docs[0][1])
+    sr_uid = stream_resource["uid"]
+    assert stream_resource["data_key"] == "test_adandor21"
+    assert stream_resource["uri"] == "file://localhost/" + str(full_file_name).lstrip(
+        "/"
+    )
+    assert stream_resource["parameters"] == {
+        "dataset": "/entry/data/data",
+        "swmr": False,
+        "multiplier": 1,
+        "chunk_shape": (1, 10, 10),
+    }
+    assert docs[1][0] == "stream_datum"
+    stream_datum = cast(StreamDatum, docs[1][1])
+    assert stream_datum["stream_resource"] == sr_uid
+    assert stream_datum["seq_nums"] == {"start": 0, "stop": 0}
+    assert stream_datum["indices"] == {"start": 0, "stop": 1}
+
+
+async def test_can_decribe_collect(
+    test_adandor: adandor.Andor2Detector, one_shot_trigger_info: TriggerInfo
+):
+    assert (await test_adandor.describe_collect()) == {}
+    await test_adandor.stage()
+    await test_adandor.prepare(one_shot_trigger_info)
+    assert (await test_adandor.describe_collect()) == {
+        "test_adandor21": {
+            "source": "mock+ca://ANDOR21:HDF1:FullFileName_RBV",
+            "shape": [10, 10],
+            "dtype": "array",
+            "dtype_numpy": "<u2",
+            "external": "STREAM:",
+        }
+    }
+
+
+async def test_unsupported_trigger_excepts(test_adandor: adandor.Andor2Detector):
+    with patch(
+        "ophyd_async.epics.adcore._hdf_writer.ADHDFWriter.open", new_callable=AsyncMock
+    ) as mock_open:
+        with pytest.raises(
+            ValueError,
+            # str(EnumClass.value) handling changed in Python 3.11
+            match=(
+                "Andor2Controller only supports the following trigger types: .* but"
+            ),
+        ):
+            await test_adandor.prepare(
+                TriggerInfo(
+                    number_of_triggers=0,
+                    trigger=DetectorTrigger.VARIABLE_GATE,
+                    deadtime=1.1,
+                    livetime=1,
+                    frame_timeout=3,
+                )
+            )
+
+    mock_open.assert_called_once()