Include chunk shape as a parameter in stream resource for HDF dataset

src/ophyd_async/core/_hdf_dataset.py

@@ -19,6 +19,10 @@ class HDFDataset:
     dtype_numpy: str = ""
     multiplier: int = 1
     swmr: bool = False
+    # Represents explicit chunk size written to disk.
+    # The first dimension represents the number of frames/points
+    # per equally-sized chunk in the first chunking dimension.
+    chunk_size: tuple[int, ...] = ()
 
 
 SLICE_NAME = "AD_HDF5_SWMR_SLICE"
@@ -65,6 +69,7 @@ def __init__(
                     "dataset": ds.dataset,
                     "swmr": ds.swmr,
                     "multiplier": ds.multiplier,
+                    "chunk_size": ds.chunk_size,
                 },
                 uid=None,
                 validate=True,

src/ophyd_async/epics/adcore/_core_io.py

@@ -135,4 +135,6 @@ def __init__(self, prefix: str, name="") -> None:
         self.array_size0 = epics_signal_r(int, prefix + "ArraySize0")
         self.array_size1 = epics_signal_r(int, prefix + "ArraySize1")
         self.create_directory = epics_signal_rw(int, prefix + "CreateDirectory")
+        self.num_frames_chunks = epics_signal_r(int, prefix + "NumFramesChunks_RBV")
+        self.chunk_size_auto = epics_signal_rw(bool, prefix + "ChunkSizeAuto")
         super().__init__(prefix, name)

src/ophyd_async/epics/adcore/_hdf_writer.py

@@ -55,6 +55,9 @@ async def open(self, multiplier: int = 1) -> Dict[str, DataKey]:
         # when directory path PV is processed.
         await self.hdf.create_directory.set(info.create_dir_depth)
 
+        # Make sure we are using chunk auto-sizing
+        await self.hdf.chunk_size_auto.set(True)
+
         await asyncio.gather(
             self.hdf.num_extra_dims.set(0),
             self.hdf.lazy_open.set(True),
@@ -83,6 +86,9 @@ async def open(self, multiplier: int = 1) -> Dict[str, DataKey]:
         self._multiplier = multiplier
         outer_shape = (multiplier,) if multiplier > 1 else ()
 
+        # Determine number of frames that will be saved per HDF chunk
+        frames_per_chunk = await self.hdf.num_frames_chunks.get_value()
+
         # Add the main data
         self._datasets = [
             HDFDataset(
@@ -91,6 +97,7 @@ async def open(self, multiplier: int = 1) -> Dict[str, DataKey]:
                 shape=detector_shape,
                 dtype_numpy=np_dtype,
                 multiplier=multiplier,
+                chunk_size=(frames_per_chunk, *detector_shape),
             )
         ]
         # And all the scalar datasets
@@ -117,6 +124,9 @@ async def open(self, multiplier: int = 1) -> Dict[str, DataKey]:
                             (),
                             np_datatype,
                             multiplier,
+                            # NDAttributes appear to always be configured with
+                            # this chunk size
+                            chunk_size=(16384,),
                         )
                     )
 

src/ophyd_async/fastcs/panda/_writer.py

@@ -102,7 +102,11 @@ async def _update_datasets(self) -> None:
 
         capture_table = await self.panda_data_block.datasets.get_value()
         self._datasets = [
-            HDFDataset(dataset_name, "/" + dataset_name, [1], multiplier=1)
+            # TODO: Update chunk size to read signal once available in IOC
+            # Currently PandA IOC sets chunk size to 1024 points per chunk
+            HDFDataset(
+                dataset_name, "/" + dataset_name, [1], multiplier=1, chunk_size=(1024,)
+            )
             for dataset_name in capture_table["name"]
         ]
 

tests/epics/adaravis/test_aravis.py

@@ -1,11 +1,9 @@
 import re
 
 import pytest
-from bluesky.run_engine import RunEngine
 
 from ophyd_async.core import (
     DetectorTrigger,
-    DeviceCollector,
     PathProvider,
     TriggerInfo,
     set_mock_value,
@@ -14,12 +12,8 @@
 
 
 @pytest.fixture
-async def test_adaravis(
-    RE: RunEngine,
-    static_path_provider: PathProvider,
-) -> adaravis.AravisDetector:
-    async with DeviceCollector(mock=True):
-        test_adaravis = adaravis.AravisDetector("ADARAVIS:", static_path_provider)
+async def test_adaravis(ad_standard_det_factory) -> adaravis.AravisDetector:
+    test_adaravis = await ad_standard_det_factory(adaravis.AravisDetector)
 
     return test_adaravis
 
@@ -72,7 +66,7 @@ def test_gpio_pin_limited(test_adaravis: adaravis.AravisDetector):
 
 
 async def test_hints_from_hdf_writer(test_adaravis: adaravis.AravisDetector):
-    assert test_adaravis.hints == {"fields": ["test_adaravis"]}
+    assert test_adaravis.hints == {"fields": ["test_adaravis1"]}
 
 
 async def test_can_read(test_adaravis: adaravis.AravisDetector):
@@ -90,9 +84,9 @@ async def test_decribe_describes_writer_dataset(
     await test_adaravis.stage()
     await test_adaravis.prepare(one_shot_trigger_info)
     assert await test_adaravis.describe() == {
-        "test_adaravis": {
-            "source": "mock+ca://ADARAVIS:HDF1:FullFileName_RBV",
-            "shape": (0, 0),
+        "test_adaravis1": {
+            "source": "mock+ca://ARAVIS1:HDF1:FullFileName_RBV",
+            "shape": (10, 10),
             "dtype": "array",
             "dtype_numpy": "|i1",
             "external": "STREAM:",
@@ -117,12 +111,13 @@ async def test_can_collect(
     assert docs[0][0] == "stream_resource"
     stream_resource = docs[0][1]
     sr_uid = stream_resource["uid"]
-    assert stream_resource["data_key"] == "test_adaravis"
+    assert stream_resource["data_key"] == "test_adaravis1"
     assert stream_resource["uri"] == "file://localhost" + str(full_file_name)
     assert stream_resource["parameters"] == {
         "dataset": "/entry/data/data",
         "swmr": False,
         "multiplier": 1,
+        "chunk_size": (1, 10, 10),
     }
     assert docs[1][0] == "stream_datum"
     stream_datum = docs[1][1]
@@ -140,9 +135,9 @@ async def test_can_decribe_collect(
     await test_adaravis.stage()
     await test_adaravis.prepare(one_shot_trigger_info)
     assert (await test_adaravis.describe_collect()) == {
-        "test_adaravis": {
-            "source": "mock+ca://ADARAVIS:HDF1:FullFileName_RBV",
-            "shape": (0, 0),
+        "test_adaravis1": {
+            "source": "mock+ca://ARAVIS1:HDF1:FullFileName_RBV",
+            "shape": (10, 10),
             "dtype": "array",
             "dtype_numpy": "|i1",
             "external": "STREAM:",

tests/epics/adcore/test_writers.py

@@ -47,6 +47,9 @@ async def hdf_writer_with_stats(
         hdf = adcore.NDFileHDFIO("HDF:")
         stats = adcore.NDPluginStatsIO("FOO:")
 
+    # Set number of frames per chunk to something reasonable
+    set_mock_value(hdf.num_frames_chunks, 2)
+
     return adcore.ADHDFWriter(
         hdf,
         static_path_provider,

tests/epics/adkinetix/test_kinetix.py

@@ -1,35 +1,28 @@
 import pytest
-from bluesky.run_engine import RunEngine
 
 from ophyd_async.core import (
     DetectorTrigger,
-    DeviceCollector,
     StaticPathProvider,
     set_mock_value,
 )
 from ophyd_async.core._detector import TriggerInfo
-from ophyd_async.epics import adkinetix
+from ophyd_async.epics.adkinetix import KinetixDetector
 
 
 @pytest.fixture
-async def test_adkinetix(
-    RE: RunEngine,
-    static_path_provider: StaticPathProvider,
-) -> adkinetix.KinetixDetector:
-    async with DeviceCollector(mock=True):
-        test_adkinetix = adkinetix.KinetixDetector("KINETIX:", static_path_provider)
-
-    return test_adkinetix
+async def test_adkinetix(ad_standard_det_factory):
+    kinetix = await ad_standard_det_factory(KinetixDetector)
+    return kinetix
 
 
 async def test_get_deadtime(
-    test_adkinetix: adkinetix.KinetixDetector,
+    test_adkinetix: KinetixDetector,
 ):
     # Currently Kinetix driver doesn't support getting deadtime.
     assert test_adkinetix._controller.get_deadtime(0) == 0.001
 
 
-async def test_trigger_modes(test_adkinetix: adkinetix.KinetixDetector):
+async def test_trigger_modes(test_adkinetix: KinetixDetector):
     set_mock_value(test_adkinetix.drv.trigger_mode, "Internal")
 
     async def setup_trigger_mode(trig_mode: DetectorTrigger):
@@ -53,17 +46,17 @@ async def setup_trigger_mode(trig_mode: DetectorTrigger):
     assert (await test_adkinetix.drv.trigger_mode.get_value()) == "Exp. Gate"
 
 
-async def test_hints_from_hdf_writer(test_adkinetix: adkinetix.KinetixDetector):
-    assert test_adkinetix.hints == {"fields": ["test_adkinetix"]}
+async def test_hints_from_hdf_writer(test_adkinetix: KinetixDetector):
+    assert test_adkinetix.hints == {"fields": ["test_adkinetix1"]}
 
 
-async def test_can_read(test_adkinetix: adkinetix.KinetixDetector):
+async def test_can_read(test_adkinetix: KinetixDetector):
     # Standard detector can be used as Readable
     assert (await test_adkinetix.read()) == {}
 
 
 async def test_decribe_describes_writer_dataset(
-    test_adkinetix: adkinetix.KinetixDetector, one_shot_trigger_info: TriggerInfo
+    test_adkinetix: KinetixDetector, one_shot_trigger_info: TriggerInfo
 ):
     set_mock_value(test_adkinetix._writer.hdf.file_path_exists, True)
     set_mock_value(test_adkinetix._writer.hdf.capture, True)
@@ -72,9 +65,9 @@ async def test_decribe_describes_writer_dataset(
     await test_adkinetix.stage()
     await test_adkinetix.prepare(one_shot_trigger_info)
     assert await test_adkinetix.describe() == {
-        "test_adkinetix": {
-            "source": "mock+ca://KINETIX:HDF1:FullFileName_RBV",
-            "shape": (0, 0),
+        "test_adkinetix1": {
+            "source": "mock+ca://KINETIX1:HDF1:FullFileName_RBV",
+            "shape": (10, 10),
             "dtype": "array",
             "dtype_numpy": "|i1",
             "external": "STREAM:",
@@ -83,7 +76,7 @@ async def test_decribe_describes_writer_dataset(
 
 
 async def test_can_collect(
-    test_adkinetix: adkinetix.KinetixDetector,
+    test_adkinetix: KinetixDetector,
     static_path_provider: StaticPathProvider,
     one_shot_trigger_info: TriggerInfo,
 ):
@@ -99,12 +92,13 @@ async def test_can_collect(
     assert docs[0][0] == "stream_resource"
     stream_resource = docs[0][1]
     sr_uid = stream_resource["uid"]
-    assert stream_resource["data_key"] == "test_adkinetix"
+    assert stream_resource["data_key"] == "test_adkinetix1"
     assert stream_resource["uri"] == "file://localhost" + str(full_file_name)
     assert stream_resource["parameters"] == {
         "dataset": "/entry/data/data",
         "swmr": False,
         "multiplier": 1,
+        "chunk_size": (1, 10, 10),
     }
     assert docs[1][0] == "stream_datum"
     stream_datum = docs[1][1]
@@ -114,17 +108,17 @@ async def test_can_collect(
 
 
 async def test_can_decribe_collect(
-    test_adkinetix: adkinetix.KinetixDetector, one_shot_trigger_info: TriggerInfo
+    test_adkinetix: KinetixDetector, one_shot_trigger_info: TriggerInfo
 ):
     set_mock_value(test_adkinetix._writer.hdf.file_path_exists, True)
     set_mock_value(test_adkinetix._writer.hdf.capture, True)
     assert (await test_adkinetix.describe_collect()) == {}
     await test_adkinetix.stage()
     await test_adkinetix.prepare(one_shot_trigger_info)
     assert (await test_adkinetix.describe_collect()) == {
-        "test_adkinetix": {
-            "source": "mock+ca://KINETIX:HDF1:FullFileName_RBV",
-            "shape": (0, 0),
+        "test_adkinetix1": {
+            "source": "mock+ca://KINETIX1:HDF1:FullFileName_RBV",
+            "shape": (10, 10),
             "dtype": "array",
             "dtype_numpy": "|i1",
             "external": "STREAM:",

tests/epics/advimba/test_vimba.py

@@ -1,9 +1,7 @@
 import pytest
-from bluesky.run_engine import RunEngine
 
 from ophyd_async.core import (
     DetectorTrigger,
-    DeviceCollector,
     PathProvider,
     set_mock_value,
 )
@@ -12,12 +10,8 @@
 
 
 @pytest.fixture
-async def test_advimba(
-    RE: RunEngine,
-    static_path_provider: PathProvider,
-) -> advimba.VimbaDetector:
-    async with DeviceCollector(mock=True):
-        test_advimba = advimba.VimbaDetector("VIMBA:", static_path_provider)
+async def test_advimba(ad_standard_det_factory) -> advimba.VimbaDetector:
+    test_advimba = await ad_standard_det_factory(advimba.VimbaDetector)
 
     return test_advimba
 
@@ -66,7 +60,7 @@ async def setup_trigger_mode(trig_mode: DetectorTrigger):
 
 
 async def test_hints_from_hdf_writer(test_advimba: advimba.VimbaDetector):
-    assert test_advimba.hints == {"fields": ["test_advimba"]}
+    assert test_advimba.hints == {"fields": ["test_advimba1"]}
 
 
 async def test_can_read(test_advimba: advimba.VimbaDetector):
@@ -84,9 +78,9 @@ async def test_decribe_describes_writer_dataset(
     await test_advimba.stage()
     await test_advimba.prepare(one_shot_trigger_info)
     assert await test_advimba.describe() == {
-        "test_advimba": {
-            "source": "mock+ca://VIMBA:HDF1:FullFileName_RBV",
-            "shape": (0, 0),
+        "test_advimba1": {
+            "source": "mock+ca://VIMBA1:HDF1:FullFileName_RBV",
+            "shape": (10, 10),
             "dtype": "array",
             "dtype_numpy": "|i1",
             "external": "STREAM:",
@@ -111,12 +105,13 @@ async def test_can_collect(
     assert docs[0][0] == "stream_resource"
     stream_resource = docs[0][1]
     sr_uid = stream_resource["uid"]
-    assert stream_resource["data_key"] == "test_advimba"
+    assert stream_resource["data_key"] == "test_advimba1"
     assert stream_resource["uri"] == "file://localhost" + str(full_file_name)
     assert stream_resource["parameters"] == {
         "dataset": "/entry/data/data",
         "swmr": False,
         "multiplier": 1,
+        "chunk_size": (1, 10, 10),
     }
     assert docs[1][0] == "stream_datum"
     stream_datum = docs[1][1]
@@ -134,9 +129,9 @@ async def test_can_decribe_collect(
     await test_advimba.stage()
     await test_advimba.prepare(one_shot_trigger_info)
     assert (await test_advimba.describe_collect()) == {
-        "test_advimba": {
-            "source": "mock+ca://VIMBA:HDF1:FullFileName_RBV",
-            "shape": (0, 0),
+        "test_advimba1": {
+            "source": "mock+ca://VIMBA1:HDF1:FullFileName_RBV",
+            "shape": (10, 10),
             "dtype": "array",
             "dtype_numpy": "|i1",
             "external": "STREAM:",

tests/epics/conftest.py

@@ -0,0 +1,33 @@
+import pytest
+from bluesky.run_engine import RunEngine
+
+from ophyd_async.core._detector import StandardDetector
+from ophyd_async.core._device import DeviceCollector
+from ophyd_async.core._mock_signal_utils import set_mock_value
+
+
+@pytest.fixture
+async def ad_standard_det_factory(
+    RE: RunEngine,
+    static_path_provider,
+) -> StandardDetector:
+    async def generate_ad_standard_det(ad_standard_detector_class, number=1):
+        detector_name = ad_standard_detector_class.__name__
+        if detector_name.endswith("Detector"):
+            detector_name = detector_name[:-8]
+
+        async with DeviceCollector(mock=True):
+            test_adstandard_det = ad_standard_detector_class(
+                f"{detector_name.upper()}{number}:",
+                static_path_provider,
+                name=f"test_ad{detector_name.lower()}{number}",
+            )
+
+        # Set number of frames per chunk and frame dimensions to something reasonable
+        set_mock_value(test_adstandard_det.hdf.num_frames_chunks, 1)
+        set_mock_value(test_adstandard_det.drv.array_size_x, 10)
+        set_mock_value(test_adstandard_det.drv.array_size_y, 10)
+
+        return test_adstandard_det
+
+    return generate_ad_standard_det