Rename multiplier to frames_per_event and move to first dim of shape

src/ophyd_async/core/_detector.py

@@ -21,7 +21,7 @@
     Triggerable,
     WritesStreamAssets,
 )
-from event_model import DataKey
+from event_model import DataKey  # type: ignore
 from pydantic import BaseModel, Field, NonNegativeInt, computed_field
 
 from ._device import Device, DeviceConnector
@@ -65,11 +65,13 @@ class TriggerInfo(BaseModel):
     livetime: float | None = Field(default=None, ge=0)
     #: What is the maximum timeout on waiting for a frame
     frame_timeout: float | None = Field(default=None, gt=0)
-    #: How many triggers make up a single StreamDatum index, to allow multiple frames
-    #: from a faster detector to be zipped with a single frame from a slow detector
-    #: e.g. if num=10 and multiplier=5 then the detector will take 10 frames,
-    #: but publish 2 indices, and describe() will show a shape of (5, h, w)
-    multiplier: int = 1
+    #: The number of triggers that are grouped into a single StreamDatum index.
+    #: A frames_per_event > 1 can be useful to have frames from a faster detector
+    #: able to be zipped with a single frame from a slower detector. E.g. if
+    #: number_of_triggers=10 and frames_per_event=5 then the detector will take
+    #: 10 frames, but publish 2 StreamDatum indices, and describe() will show a
+    #: shape of (5, h, w) for each.
+    frames_per_event: NonNegativeInt = 1
 
     @computed_field
     @cached_property
@@ -107,8 +109,8 @@ async def prepare(self, trigger_info: TriggerInfo) -> None:
                 exposure time.
                 deadtime Defaults to None. This is the minimum deadtime between
                 triggers.
-                multiplier The number of triggers grouped into a single StreamDatum
-                index.
+                frames_per_event The number of triggers grouped into a single
+                StreamDatum index
         """
 
     @abstractmethod
@@ -133,13 +135,17 @@ class DetectorWriter(ABC):
     (e.g. an HDF5 file)"""
 
     @abstractmethod
-    async def open(self, multiplier: int = 1) -> dict[str, DataKey]:
+    async def open(self, frames_per_event: int = 1) -> dict[str, DataKey]:
         """Open writer and wait for it to be ready for data.
 
         Args:
-            multiplier: Each StreamDatum index corresponds to this many
-                written exposures
-
+            frames_per_event: The number of triggers that are grouped into a single
+                StreamDatum index. A frames_per_event > 1 can be useful to have
+                frames from a faster detector able to be zipped with a single frame
+                from a slower detector. E.g. if number_of_triggers=10 and
+                frames_per_event=5 then the detector will take 10 frames, but publish
+                2 StreamDatum indices, and describe() will show a shape of (5, h, w)
+                for each.
         Returns:
             Output for ``describe()``
         """
@@ -148,7 +154,7 @@ async def open(self, multiplier: int = 1) -> dict[str, DataKey]:
     def observe_indices_written(
         self, timeout=DEFAULT_TIMEOUT
     ) -> AsyncGenerator[int, None]:
-        """Yield the index of each frame (or equivalent data point) as it is written"""
+        """Yield the index of each frame (or batch of frames) as it is written"""
 
     @abstractmethod
     async def get_indices_written(self) -> int:
@@ -327,10 +333,12 @@ async def prepare(self, value: TriggerInfo) -> None:
             if isinstance(self._trigger_info.number_of_triggers, list)
             else [self._trigger_info.number_of_triggers]
         )
-        self._initial_frame = await self._writer.get_indices_written()
+        # Open the writer and prepare the controller.
         self._describe, _ = await asyncio.gather(
-            self._writer.open(value.multiplier), self._controller.prepare(value)
+            self._writer.open(value.frames_per_event), self._controller.prepare(value)
         )
+        # Get the initial frame index from the writer.
+        self._initial_frame = await self._writer.get_indices_written()
         if value.trigger != DetectorTrigger.INTERNAL:
             await self._controller.arm()
             self._fly_start = time.monotonic()

src/ophyd_async/core/_hdf_dataset.py

@@ -3,7 +3,7 @@
 from pathlib import Path
 from urllib.parse import urlunparse
 
-from event_model import (
+from event_model import (  # type: ignore
     ComposeStreamResource,
     ComposeStreamResourceBundle,
     StreamDatum,
@@ -18,7 +18,6 @@ class HDFDataset:
     dataset: str
     shape: Sequence[int] = field(default_factory=tuple)
     dtype_numpy: str = ""
-    multiplier: int = 1
     swmr: bool = False
     # Represents explicit chunk size written to disk.
     chunk_shape: tuple[int, ...] = ()
@@ -67,7 +66,7 @@ def __init__(
                 parameters={
                     "dataset": ds.dataset,
                     "swmr": ds.swmr,
-                    "multiplier": ds.multiplier,
+                    "shape": ds.shape,
                     "chunk_shape": ds.chunk_shape,
                 },
                 uid=None,

src/ophyd_async/epics/adcore/_core_writer.py

@@ -5,7 +5,7 @@
 from urllib.parse import urlunparse
 
 from bluesky.protocols import Hints, StreamAsset
-from event_model import (
+from event_model import (  # type: ignore
     ComposeStreamResource,
     DataKey,
     StreamRange,
@@ -59,7 +59,6 @@ def __init__(
         self._emitted_resource = None
 
         self._capture_status: AsyncStatus | None = None
-        self._multiplier = 1
         self._filename_template = "%s%s_%6.6d"
 
     @classmethod
@@ -123,10 +122,10 @@ async def begin_capture(self) -> None:
             self.fileio.capture, True, wait_for_set_completion=False
         )
 
-    async def open(self, multiplier: int = 1) -> dict[str, DataKey]:
+    async def open(self, frames_per_event: int = 1) -> dict[str, DataKey]:
         self._emitted_resource = None
         self._last_emitted = 0
-        self._multiplier = multiplier
+        self._frames_per_event = frames_per_event
         frame_shape = await self._dataset_describer.shape()
         dtype_numpy = await self._dataset_describer.np_datatype()
 
@@ -135,7 +134,7 @@ async def open(self, multiplier: int = 1) -> dict[str, DataKey]:
         describe = {
             self._name_provider(): DataKey(
                 source=self._name_provider(),
-                shape=list(frame_shape),
+                shape=[frames_per_event, *frame_shape],
                 dtype="array",
                 dtype_numpy=dtype_numpy,
                 external="STREAM:",
@@ -148,11 +147,11 @@ async def observe_indices_written(
     ) -> AsyncGenerator[int, None]:
         """Wait until a specific index is ready to be collected"""
         async for num_captured in observe_value(self.fileio.num_captured, timeout):
-            yield num_captured // self._multiplier
+            yield num_captured // self._frames_per_event
 
     async def get_indices_written(self) -> int:
         num_captured = await self.fileio.num_captured.get_value()
-        return num_captured // self._multiplier
+        return num_captured // self._frames_per_event
 
     async def collect_stream_docs(
         self, indices_written: int
@@ -181,10 +180,17 @@ async def collect_stream_docs(
                 self._emitted_resource = bundler_composer(
                     mimetype=self._mimetype,
                     uri=uri,
+                    # TODO: This is confusing, I expected this to be of type `DataKey`
+                    # but it is a string. Naming could be improved maybe?
                     data_key=self._name_provider(),
+                    # Q: What are the parameters used for? Extra info?
                     parameters={
-                        # Assume that we always write 1 frame per file/chunk
-                        "chunk_shape": (1, *frame_shape),
+                        # Assume that we always write self._frames_per_event
+                        # frames per file/chunk
+                        # TODO: Validate this assumption and that it should
+                        # not be self._frames_per_event
+                        "chunk_shape": (self._frames_per_event, *frame_shape),
+                        "shape": (self._frames_per_event, *frame_shape),
                         # Include file template for reconstruction in consolidator
                         "template": file_template,
                     },

src/ophyd_async/epics/adcore/_hdf_writer.py

@@ -4,7 +4,7 @@
 from xml.etree import ElementTree as ET
 
 from bluesky.protocols import Hints, StreamAsset
-from event_model import DataKey
+from event_model import DataKey  # type: ignore
 
 from ophyd_async.core import (
     DEFAULT_TIMEOUT,
@@ -48,7 +48,7 @@ def __init__(
         self._file: HDFFile | None = None
         self._include_file_number = False
 
-    async def open(self, multiplier: int = 1) -> dict[str, DataKey]:
+    async def open(self, frames_per_event: int = 1) -> dict[str, DataKey]:
         self._file = None
 
         # Setting HDF writer specific signals
@@ -74,8 +74,9 @@ async def open(self, multiplier: int = 1) -> dict[str, DataKey]:
         name = self._name_provider()
         detector_shape = await self._dataset_describer.shape()
         np_dtype = await self._dataset_describer.np_datatype()
-        self._multiplier = multiplier
-        outer_shape = (multiplier,) if multiplier > 1 else ()
+
+        # Used by the base class
+        self._frames_per_event = frames_per_event
 
         # Determine number of frames that will be saved per HDF chunk
         frames_per_chunk = await self.fileio.num_frames_chunks.get_value()
@@ -85,9 +86,10 @@ async def open(self, multiplier: int = 1) -> dict[str, DataKey]:
             HDFDataset(
                 data_key=name,
                 dataset="/entry/data/data",
-                shape=detector_shape,
+                shape=(frames_per_event, *detector_shape)
+                if frames_per_event > 1 or detector_shape
+                else (),
                 dtype_numpy=np_dtype,
-                multiplier=multiplier,
                 chunk_shape=(frames_per_chunk, *detector_shape),
             )
         ]
@@ -112,9 +114,8 @@ async def open(self, multiplier: int = 1) -> dict[str, DataKey]:
                         HDFDataset(
                             datakey,
                             f"/entry/instrument/NDAttributes/{datakey}",
-                            (),
+                            (frames_per_event,) if frames_per_event > 1 else (),
                             np_datatype,
-                            multiplier,
                             # NDAttributes appear to always be configured with
                             # this chunk size
                             chunk_shape=(16384,),
@@ -124,7 +125,7 @@ async def open(self, multiplier: int = 1) -> dict[str, DataKey]:
         describe = {
             ds.data_key: DataKey(
                 source=self.fileio.full_file_name.source,
-                shape=list(outer_shape + tuple(ds.shape)),
+                shape=list(ds.shape),
                 dtype="array" if ds.shape else "number",
                 dtype_numpy=ds.dtype_numpy,
                 external="STREAM:",

src/ophyd_async/epics/eiger/_odin_io.py

@@ -2,7 +2,7 @@
 from collections.abc import AsyncGenerator, AsyncIterator
 
 from bluesky.protocols import StreamAsset
-from event_model import DataKey
+from event_model import DataKey  # type: ignore
 
 from ophyd_async.core import (
     DEFAULT_TIMEOUT,
@@ -77,8 +77,9 @@ def __init__(
         self._name_provider = name_provider
         super().__init__()
 
-    async def open(self, multiplier: int = 1) -> dict[str, DataKey]:
+    async def open(self, frames_per_event: int = 1) -> dict[str, DataKey]:
         info = self._path_provider(device_name=self._name_provider())
+        self._frames_per_event = frames_per_event
 
         await asyncio.gather(
             self._drv.file_path.set(str(info.directory_path)),
@@ -101,7 +102,7 @@ async def _describe(self) -> dict[str, DataKey]:
         return {
             "data": DataKey(
                 source=self._drv.file_name.source,
-                shape=list(data_shape),
+                shape=[self._frames_per_event, *data_shape],
                 dtype="array",
                 # TODO: Use correct type based on eiger https://github.com/bluesky/ophyd-async/issues/529
                 dtype_numpy="<u2",
@@ -113,10 +114,10 @@ async def observe_indices_written(
         self, timeout=DEFAULT_TIMEOUT
     ) -> AsyncGenerator[int, None]:
         async for num_captured in observe_value(self._drv.num_captured, timeout):
-            yield num_captured
+            yield num_captured // self._frames_per_event
 
     async def get_indices_written(self) -> int:
-        return await self._drv.num_captured.get_value()
+        return await self._drv.num_captured.get_value() // self._frames_per_event
 
     def collect_stream_docs(self, indices_written: int) -> AsyncIterator[StreamAsset]:
         # TODO: Correctly return stream https://github.com/bluesky/ophyd-async/issues/530

src/ophyd_async/fastcs/panda/_writer.py

@@ -3,8 +3,8 @@
 from pathlib import Path
 
 from bluesky.protocols import StreamAsset
-from event_model import DataKey
-from p4p.client.thread import Context
+from event_model import DataKey  # type: ignore
+from p4p.client.thread import Context  # type: ignore
 
 from ophyd_async.core import (
     DEFAULT_TIMEOUT,
@@ -34,10 +34,9 @@ def __init__(
         self._name_provider = name_provider
         self._datasets: list[HDFDataset] = []
         self._file: HDFFile | None = None
-        self._multiplier = 1
 
     # Triggered on PCAP arm
-    async def open(self, multiplier: int = 1) -> dict[str, DataKey]:
+    async def open(self, frames_per_event: int = 1) -> dict[str, DataKey]:
         """Retrieve and get descriptor of all PandA signals marked for capture"""
 
         # Ensure flushes are immediate
@@ -68,9 +67,9 @@ async def open(self, multiplier: int = 1) -> dict[str, DataKey]:
 
         # Wait for it to start, stashing the status that tells us when it finishes
         await self.panda_data_block.capture.set(True)
-        if multiplier > 1:
+        if frames_per_event > 1:
             raise ValueError(
-                "All PandA datasets should be scalar, multiplier should be 1"
+                "All PandA datasets should be scalar, frames_per_event should be 1"
             )
 
         return await self._describe()
@@ -84,8 +83,9 @@ async def _describe(self) -> dict[str, DataKey]:
         describe = {
             ds.data_key: DataKey(
                 source=self.panda_data_block.hdf_directory.source,
-                shape=list(ds.shape),
-                dtype="array" if ds.shape != [1] else "number",
+                # frames_per_event is always 1 for PandA
+                shape=[1, *ds.shape] if ds.shape else [],
+                dtype="array" if ds.shape else "number",
                 # PandA data should always be written as Float64
                 dtype_numpy="<f8",
                 external="STREAM:",
@@ -104,9 +104,7 @@ async def _update_datasets(self) -> None:
         self._datasets = [
             # 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_shape=(1024,)
-            )
+            HDFDataset(dataset_name, "/" + dataset_name, shape=(), chunk_shape=(1024,))
             for dataset_name in capture_table.name
         ]
 
@@ -141,7 +139,7 @@ async def observe_indices_written(
         async for num_captured in observe_value(
             self.panda_data_block.num_captured, timeout
         ):
-            yield num_captured // self._multiplier
+            yield num_captured
 
     async def collect_stream_docs(
         self, indices_written: int

src/ophyd_async/sim/_pattern_detector/_pattern_detector_writer.py

@@ -1,6 +1,6 @@
 from collections.abc import AsyncGenerator, AsyncIterator
 
-from event_model import DataKey
+from event_model import DataKey  # type: ignore
 
 from ophyd_async.core import DEFAULT_TIMEOUT, DetectorWriter, NameProvider, PathProvider
 
@@ -20,9 +20,10 @@ def __init__(
         self.path_provider = path_provider
         self.name_provider = name_provider
 
-    async def open(self, multiplier: int = 1) -> dict[str, DataKey]:
+    async def open(self, frames_per_event: int = 1) -> dict[str, DataKey]:
+        self._frames_per_event = frames_per_event
         return await self.pattern_generator.open_file(
-            self.path_provider, self.name_provider(), multiplier
+            self.path_provider, self.name_provider(), frames_per_event
         )
 
     async def close(self) -> None:
@@ -35,7 +36,7 @@ async def observe_indices_written(
         self, timeout=DEFAULT_TIMEOUT
     ) -> AsyncGenerator[int, None]:
         async for index in self.pattern_generator.observe_indices_written(timeout):
-            yield index
+            yield index // self._frames_per_event
 
     async def get_indices_written(self) -> int:
-        return self.pattern_generator.image_counter
+        return self.pattern_generator.image_counter // self._frames_per_event