Rename multiplier to frames_per_event and move to first dim of shape

src/ophyd_async/core/_detector.py

@@ -22,7 +22,7 @@
     WritesStreamAssets,
 )
 from event_model import DataKey
-from pydantic import BaseModel, Field, NonNegativeInt, computed_field
+from pydantic import BaseModel, Field, NonNegativeInt, PositiveInt, computed_field
 
 from ._device import Device, DeviceConnector
 from ._protocol import AsyncConfigurable, AsyncReadable
@@ -65,19 +65,21 @@ 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: PositiveInt = 1
 
     @computed_field
     @cached_property
     def total_number_of_triggers(self) -> int:
         return (
-            sum(self.number_of_triggers)
+            sum(self.number_of_triggers) * self.frames_per_event
             if isinstance(self.number_of_triggers, list)
-            else self.number_of_triggers
+            else self.number_of_triggers * self.frames_per_event
         )
 
 
@@ -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:
@@ -339,10 +345,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,6 @@ def __init__(
                 parameters={
                     "dataset": ds.dataset,
                     "swmr": ds.swmr,
-                    "multiplier": ds.multiplier,
                     "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
@@ -183,7 +182,8 @@ async def collect_stream_docs(
                     uri=uri,
                     data_key=self._name_provider(),
                     parameters={
-                        # Assume that we always write 1 frame per file/chunk
+                        # Assume that we always write 1 frame per file/chunk, this
+                        # may change to self._frames_per_event in the future
                         "chunk_shape": (1, *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,8 @@ 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),
                 dtype_numpy=np_dtype,
-                multiplier=multiplier,
                 chunk_shape=(frames_per_chunk, *detector_shape),
             )
         ]
@@ -112,9 +112,8 @@ async def open(self, multiplier: int = 1) -> dict[str, DataKey]:
                         HDFDataset(
                             datakey,
                             f"/entry/instrument/NDAttributes/{datakey}",
-                            (),
+                            (frames_per_event,),
                             np_datatype,
-                            multiplier,
                             # NDAttributes appear to always be configured with
                             # this chunk size
                             chunk_shape=(16384,),
@@ -124,8 +123,8 @@ 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)),
-                dtype="array" if ds.shape else "number",
+                shape=list(ds.shape),
+                dtype="array" if len(ds.shape) > 1 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,11 +34,11 @@ 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"""
+        self._frames_per_event = frames_per_event
 
         # Ensure flushes are immediate
         await self.panda_data_block.flush_period.set(0)
@@ -68,10 +68,6 @@ 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:
-            raise ValueError(
-                "All PandA datasets should be scalar, multiplier should be 1"
-            )
 
         return await self._describe()
 
@@ -85,7 +81,7 @@ async def _describe(self) -> dict[str, DataKey]:
             ds.data_key: DataKey(
                 source=self.panda_data_block.hdf_directory.source,
                 shape=list(ds.shape),
-                dtype="array" if ds.shape != [1] else "number",
+                dtype="array" if len(ds.shape) > 1 else "number",
                 # PandA data should always be written as Float64
                 dtype_numpy="<f8",
                 external="STREAM:",
@@ -105,7 +101,10 @@ async def _update_datasets(self) -> None:
             # 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,)
+                dataset_name,
+                "/" + dataset_name,
+                shape=(self._frames_per_event,),
+                chunk_shape=(1024,),
             )
             for dataset_name in capture_table.name
         ]
@@ -124,15 +123,20 @@ async def _update_datasets(self) -> None:
     # StandardDetector behavior
     async def wait_for_index(self, index: int, timeout: float | None = DEFAULT_TIMEOUT):
         def matcher(value: int) -> bool:
-            return value >= index
+            # Index is already divided by frames_per_event, so we need to also
+            # divide the value by frames_per_event to get the correct index
+            return value // self._frames_per_event >= index
 
         matcher.__name__ = f"index_at_least_{index}"
         await wait_for_value(
             self.panda_data_block.num_captured, matcher, timeout=timeout
         )
 
     async def get_indices_written(self) -> int:
-        return await self.panda_data_block.num_captured.get_value()
+        return (
+            await self.panda_data_block.num_captured.get_value()
+            // self._frames_per_event
+        )
 
     async def observe_indices_written(
         self, timeout=DEFAULT_TIMEOUT
@@ -141,7 +145,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 // self._frames_per_event
 
     async def collect_stream_docs(
         self, indices_written: int