Add tests for kinetix and vimba, add detector classes for both.

src/ophyd_async/epics/areadetector/__init__.py

@@ -1,4 +1,5 @@
 from .aravis import AravisDetector
+from .kinetix import KinetixDetector
 from .pilatus import PilatusDetector
 from .single_trigger_det import SingleTriggerDet
 from .utils import (
@@ -9,9 +10,12 @@
     ad_r,
     ad_rw,
 )
+from .vimba import VimbaDetector
 
 __all__ = [
     "AravisDetector",
+    "KinetixDetector",
+    "VimbaDetector",
     "SingleTriggerDet",
     "FileWriteMode",
     "ImageMode",

src/ophyd_async/epics/areadetector/controllers/vimba_controller.py

@@ -51,7 +51,7 @@ async def arm(
     ) -> AsyncStatus:
         await asyncio.gather(
             self._drv.trigger_mode.set(TRIGGER_MODE[trigger]),
-            self._drv.expose_out_mode.set(EXPOSE_OUT_MODE[trigger]),
+            self._drv.expose_mode.set(EXPOSE_OUT_MODE[trigger]),
             self._drv.num_images.set(num),
             self._drv.image_mode.set(ImageMode.multiple),
         )
@@ -61,7 +61,9 @@ async def arm(
         ]:
             await self._drv.acquire_time.set(exposure)
         if trigger != DetectorTrigger.internal:
-            self._drv.trigger_source.set(VimbaTriggerSource.line1)
+            self._drv.trig_source.set(VimbaTriggerSource.line1)
+        else:
+            self._drv.trig_source.set(VimbaTriggerSource.freerun)
         return await start_acquiring_driver_and_ensure_status(
             self._drv, good_states=self.good_states
         )

src/ophyd_async/epics/areadetector/drivers/__init__.py

@@ -5,13 +5,17 @@
     start_acquiring_driver_and_ensure_status,
 )
 from .aravis_driver import AravisDriver
+from .kinetix_driver import KinetixDriver
 from .pilatus_driver import PilatusDriver
+from .vimba_driver import VimbaDriver
 
 __all__ = [
     "ADBase",
     "ADBaseShapeProvider",
     "PilatusDriver",
     "AravisDriver",
+    "KinetixDriver",
+    "VimbaDriver",
     "start_acquiring_driver_and_ensure_status",
     "DetectorState",
 ]

src/ophyd_async/epics/areadetector/kinetix.py

@@ -0,0 +1,48 @@
+from bluesky.protocols import HasHints, Hints
+
+from ophyd_async.core import DirectoryProvider, StandardDetector
+from ophyd_async.epics.areadetector.controllers.kinetix_controller import (
+    KinetixController,
+)
+from ophyd_async.epics.areadetector.drivers import ADBaseShapeProvider
+from ophyd_async.epics.areadetector.drivers.kinetix_driver import KinetixDriver
+from ophyd_async.epics.areadetector.writers import HDFWriter, NDFileHDF
+
+
+class KinetixDetector(StandardDetector, HasHints):
+    """
+    Ophyd-async implementation of an ADKinetix Detector.
+    https://github.com/NSLS-II/ADKinetix
+    """
+
+    _controller: KinetixController
+    _writer: HDFWriter
+
+    def __init__(
+        self,
+        name: str,
+        directory_provider: DirectoryProvider,
+        driver: KinetixDriver,
+        hdf: NDFileHDF,
+        **scalar_sigs: str,
+    ):
+        # Must be child of Detector to pick up connect()
+        self.drv = driver
+        self.hdf = hdf
+
+        super().__init__(
+            KinetixController(self.drv),
+            HDFWriter(
+                self.hdf,
+                directory_provider,
+                lambda: self.name,
+                ADBaseShapeProvider(self.drv),
+                **scalar_sigs,
+            ),
+            config_sigs=(self.drv.acquire_time, self.drv.acquire),
+            name=name,
+        )
+
+    @property
+    def hints(self) -> Hints:
+        return self._writer.hints

src/ophyd_async/epics/areadetector/vimba.py

@@ -0,0 +1,45 @@
+from bluesky.protocols import HasHints, Hints
+
+from ophyd_async.core import DirectoryProvider, StandardDetector
+from ophyd_async.epics.areadetector.controllers.vimba_controller import VimbaController
+from ophyd_async.epics.areadetector.drivers import ADBaseShapeProvider
+from ophyd_async.epics.areadetector.drivers.vimba_driver import VimbaDriver
+from ophyd_async.epics.areadetector.writers import HDFWriter, NDFileHDF
+
+
+class VimbaDetector(StandardDetector, HasHints):
+    """
+    Ophyd-async implementation of an ADVimba Detector.
+    """
+
+    _controller: VimbaController
+    _writer: HDFWriter
+
+    def __init__(
+        self,
+        name: str,
+        directory_provider: DirectoryProvider,
+        driver: VimbaDriver,
+        hdf: NDFileHDF,
+        **scalar_sigs: str,
+    ):
+        # Must be child of Detector to pick up connect()
+        self.drv = driver
+        self.hdf = hdf
+
+        super().__init__(
+            VimbaController(self.drv),
+            HDFWriter(
+                self.hdf,
+                directory_provider,
+                lambda: self.name,
+                ADBaseShapeProvider(self.drv),
+                **scalar_sigs,
+            ),
+            config_sigs=(self.drv.acquire_time, self.drv.acquire),
+            name=name,
+        )
+
+    @property
+    def hints(self) -> Hints:
+        return self._writer.hints

tests/epics/areadetector/test_kinetix.py

@@ -0,0 +1,150 @@
+import pytest
+from bluesky.run_engine import RunEngine
+
+from ophyd_async.core import (
+    DetectorTrigger,
+    DeviceCollector,
+    DirectoryProvider,
+    set_sim_value,
+)
+from ophyd_async.epics.areadetector.drivers.kinetix_driver import KinetixDriver
+from ophyd_async.epics.areadetector.kinetix import KinetixDetector
+from ophyd_async.epics.areadetector.writers.nd_file_hdf import NDFileHDF
+
+
+@pytest.fixture
+async def adkinetix_driver(RE: RunEngine) -> KinetixDriver:
+    async with DeviceCollector(sim=True):
+        driver = KinetixDriver("DRV:")
+
+    return driver
+
+
+@pytest.fixture
+async def hdf(RE: RunEngine) -> NDFileHDF:
+    async with DeviceCollector(sim=True):
+        hdf = NDFileHDF("HDF:")
+
+    return hdf
+
+
+@pytest.fixture
+async def adkinetix(
+    RE: RunEngine,
+    static_directory_provider: DirectoryProvider,
+    adkinetix_driver: KinetixDriver,
+    hdf: NDFileHDF,
+) -> KinetixDetector:
+    async with DeviceCollector(sim=True):
+        adkinetix = KinetixDetector(
+            "adkinetix",
+            static_directory_provider,
+            driver=adkinetix_driver,
+            hdf=hdf,
+        )
+
+    return adkinetix
+
+
+async def test_get_deadtime(
+    adkinetix: KinetixDetector,
+):
+    # Currently Kinetix driver doesn't support getting deadtime.
+    assert adkinetix._controller.get_deadtime(0) == 0.001
+
+
+async def test_trigger_modes(adkinetix: KinetixDetector):
+    set_sim_value(adkinetix.drv.trigger_mode, "Internal")
+
+    async def setup_trigger_mode(trig_mode: DetectorTrigger):
+        await adkinetix.controller.arm(num=1, trigger=trig_mode)
+        # Prevent timeouts
+        set_sim_value(adkinetix.drv.acquire, True)
+
+    # Default TriggerSource
+    assert (await adkinetix.drv.trigger_mode.get_value()) == "Internal"
+
+    await setup_trigger_mode(DetectorTrigger.edge_trigger)
+    assert (await adkinetix.drv.trigger_mode.get_value()) == "Rising Edge"
+
+    await setup_trigger_mode(DetectorTrigger.constant_gate)
+    assert (await adkinetix.drv.trigger_mode.get_value()) == "Exp. Gate"
+
+    await setup_trigger_mode(DetectorTrigger.internal)
+    assert (await adkinetix.drv.trigger_mode.get_value()) == "Internal"
+
+    await setup_trigger_mode(DetectorTrigger.variable_gate)
+    assert (await adkinetix.drv.trigger_mode.get_value()) == "Exp. Gate"
+
+
+async def test_hints_from_hdf_writer(adkinetix: KinetixDetector):
+    assert adkinetix.hints == {"fields": ["adkinetix"]}
+
+
+async def test_can_read(adkinetix: KinetixDetector):
+    # Standard detector can be used as Readable
+    assert (await adkinetix.read()) == {}
+
+
+async def test_decribe_describes_writer_dataset(adkinetix: KinetixDetector):
+    set_sim_value(adkinetix._writer.hdf.file_path_exists, True)
+    set_sim_value(adkinetix._writer.hdf.capture, True)
+
+    assert await adkinetix.describe() == {}
+    await adkinetix.stage()
+    assert await adkinetix.describe() == {
+        "adkinetix": {
+            "source": "soft://adkinetix-hdf-full_file_name",
+            "shape": (0, 0),
+            "dtype": "array",
+            "external": "STREAM:",
+        }
+    }
+
+
+async def test_can_collect(
+    adkinetix: KinetixDetector, static_directory_provider: DirectoryProvider
+):
+    directory_info = static_directory_provider()
+    full_file_name = directory_info.root / directory_info.resource_dir / "foo.h5"
+    set_sim_value(adkinetix.hdf.full_file_name, str(full_file_name))
+    set_sim_value(adkinetix._writer.hdf.file_path_exists, True)
+    set_sim_value(adkinetix._writer.hdf.capture, True)
+    await adkinetix.stage()
+    docs = [(name, doc) async for name, doc in adkinetix.collect_asset_docs(1)]
+    assert len(docs) == 2
+    assert docs[0][0] == "stream_resource"
+    stream_resource = docs[0][1]
+    sr_uid = stream_resource["uid"]
+    assert stream_resource["data_key"] == "adkinetix"
+    assert stream_resource["spec"] == "AD_HDF5_SWMR_SLICE"
+    assert stream_resource["root"] == str(directory_info.root)
+    assert stream_resource["resource_path"] == str(
+        directory_info.resource_dir / "foo.h5"
+    )
+    assert stream_resource["path_semantics"] == "posix"
+    assert stream_resource["resource_kwargs"] == {
+        "path": "/entry/data/data",
+        "multiplier": 1,
+        "timestamps": "/entry/instrument/NDAttributes/NDArrayTimeStamp",
+    }
+    assert docs[1][0] == "stream_datum"
+    stream_datum = 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(adkinetix: KinetixDetector):
+    set_sim_value(adkinetix._writer.hdf.file_path_exists, True)
+    set_sim_value(adkinetix._writer.hdf.capture, True)
+    assert (await adkinetix.describe_collect()) == {}
+    await adkinetix.stage()
+    assert (await adkinetix.describe_collect()) == {
+        "adkinetix": {
+            "source": "soft://adkinetix-hdf-full_file_name",
+            "shape": (0, 0),
+            "dtype": "array",
+            "external": "STREAM:",
+        }
+    }