Ophyd async is included on a provisional basis until the v1.0 release and may change API on minor release numbers before then

Make a StandardDetector#

StandardDetector is an abstract class to assist in creating Device classes for hardware that writes its own data e.g. an AreaDetector implementation, or a PandA writing motor encoder positions to file. The StandardDetector is a simple compound device, with 2 standard components:

  • DetectorWriter to handle data persistence, i/o and pass information about data to the RunEngine (usually an instance of ADHDFWriter)

  • DetectorControl with logic for arming and disarming the detector. This will be unique to the StandardDetector implementation.

Writing an AreaDetector StandardDetector#

For an AreaDetector implementation of the StandardDetector, two entity objects which are subdevices of the StandardDetector are used to map to AreaDetector plugins:

  • An NDPluginFile instance (for ADHDFWriter an instance of NDFileHDFIO)

  • An ADBaseIO instance mapping to NDArray for the “driver” of the detector implementation

Define a FooDriver if the NDArray requires fields in addition to those on ADBaseIO to be exposed. It should extend ADBaseIO. Enumeration fields should be named to prevent namespace collision, i.e. for a Signal named “TriggerSource” use the enum “FooTriggerSource”

class FooDriver(adcore.ADBaseIO):
    def __init__(self, prefix: str, name: str = "") -> None:
        self.trigger_mode = epics_signal_rw_rbv(str, prefix + "TriggerMode")
        super().__init__(prefix, name)

Define a FooController with handling for converting the standard pattern of ophyd_async.core.DetectorControl.arm() and ophyd_async.core.DetectorControl.disarm() to required state of FooDriver e.g. setting a compatible “FooTriggerSource” for a given DetectorTrigger, or raising an exception if incompatible with the DetectorTrigger.

The ophyd_async.core.DetectorControl.get_deadtime() method is used when constructing sequence tables for hardware controlled scanning. Details on how to calculate the deadtime may be only available from technical manuals or otherwise complex. If it requires fetching from signals, it is recommended to cache the value during the StandardDetector `prepare` method.

class FooController(DetectorControl):
    def __init__(self, driver: FooDriver) -> None:
        self._drv = driver

    def get_deadtime(self, exposure: float) -> float:
        # FooDetector deadtime handling
        return 0.001

    async def arm(
        num: int,
        trigger: DetectorTrigger = DetectorTrigger.internal,
        exposure: Optional[float] = None,
    ) -> AsyncStatus:
        await asyncio.gather(
        if exposure is not None:
            await self._drv.acquire_time.set(exposure)
        return await adcore.start_acquiring_driver_and_ensure_status(self._drv)

    async def disarm(self):
        await adcore.stop_busy_record(self._drv.acquire, False, timeout=1)

FooDetector ties the Driver, Controller and data persistence layer together. The example FooDetector writes h5 files using the standard NDPlugin. It additionally supports the HasHints protocol which is optional but recommended.

Its initialiser assumes the NSLS-II AreaDetector plugin EPICS address suffixes as defaults but allows overriding: this pattern is recommended for consistency. If the FooDriver signals that should be read as configuration, they should be added to the “config_sigs” passed to the super.

class FooDetector(StandardDetector, HasHints):
    _controller: FooController
    _writer: adcore.ADHDFWriter

    def __init__(
        prefix: str,
        path_provider: PathProvider,
        # Must be children to pick up connect
        self.drv = FooDriver(prefix + drv_suffix)
        self.hdf = adcore.NDFileHDFIO(prefix + hdf_suffix)


    def hints(self) -> Hints:
        return self._writer.hints

Writing a non-AreaDetector StandardDetector#

A non-AreaDetector StandardDetector should implement DetectorControl and DetectorWriter directly. Here we construct a DetectorControl that co-ordinates signals on a PandA PositionCapture block - a child device “pcap” of the StandardDetector implementation, analogous to the FooDriver.

class PandaPcapController(DetectorControl):
    def __init__(self, pcap: PcapBlock) -> None:
        self.pcap = pcap

    def get_deadtime(self, exposure: float) -> float:
        return 0.000000008

    async def arm(
        num: int,
        trigger: DetectorTrigger = DetectorTrigger.constant_gate,
        exposure: Optional[float] = None,
    ) -> AsyncStatus:
        assert trigger in (
            trigger == DetectorTrigger.variable_gate,
        ), "Only constant_gate and variable_gate triggering is supported on the PandA"
        await asyncio.gather(self.pcap.arm.set(True))
        await wait_for_value(, True, timeout=1)
        return AsyncStatus(wait_for_value(, False, timeout=None))

    async def disarm(self) -> AsyncStatus:
        await asyncio.gather(self.pcap.arm.set(False))
        await wait_for_value(, False, timeout=1)
        return AsyncStatus(wait_for_value(, False, timeout=None))

The PandA may write a number of fields, and the PandaHDFWriter co-ordinates those, configures the filewriter and describes the data for the RunEngine.

class PandaHDFWriter(DetectorWriter):
    _ctxt: Optional[Context] = None

    def __init__(
        prefix: str,
        path_provider: PathProvider,
        name_provider: NameProvider,
        panda_device: CommonPandaBlocks,
    ) -> None:
        self.panda_device = panda_device
        self._prefix = prefix
        self._path_provider = path_provider
        self._name_provider = name_provider
        self._datasets: List[HDFDataset] = []
        self._file: Optional[HDFFile] = None
        self._multiplier = 1

    # Triggered on PCAP arm
    async def open(self, multiplier: int = 1) -> Dict[str, DataKey]:
        """Retrieve and get descriptor of all PandA signals marked for capture"""

        # Ensure flushes are immediate

        self._file = None
        info = self._path_provider(
        # Set the initial values
        await asyncio.gather(
            # TODO: Set create_dir_depth once available

        # Wait for it to start, stashing the status that tells us when it finishes
        if multiplier > 1:
            raise ValueError(
                "All PandA datasets should be scalar, multiplier should be 1"

        return await self._describe()

    async def _describe(self) -> Dict[str, DataKey]:
        Return a describe based on the datasets PV

        await self._update_datasets()
        describe = {
            ds.data_key: DataKey(
                dtype="array" if ds.shape != [1] else "number",
                dtype_numpy="<f8",  # PandA data should always be written as Float64
            for ds in self._datasets
        return describe

    async def _update_datasets(self) -> None:
        Load data from the datasets PV on the panda, update internal
        representation of datasets that the panda will write.

        capture_table = await
        self._datasets = [
            HDFDataset(dataset_name, "/" + dataset_name, [1], multiplier=1)
            for dataset_name in capture_table["name"]

    # Next few functions are exactly the same as AD writer. Could move as default
    # StandardDetector behavior
    async def wait_for_index(
        self, index: int, timeout: Optional[float] = DEFAULT_TIMEOUT
        def matcher(value: int) -> bool:
            return value >= index

        matcher.__name__ = f"index_at_least_{index}"
        await wait_for_value(
  , matcher, timeout=timeout

    async def get_indices_written(self) -> int:
        return await

    async def observe_indices_written(
        self, timeout=DEFAULT_TIMEOUT
    ) -> AsyncGenerator[int, None]:
        """Wait until a specific index is ready to be collected"""
        async for num_captured in observe_value(
  , timeout
            yield num_captured // self._multiplier

    async def collect_stream_docs(
        self, indices_written: int
    ) -> AsyncIterator[StreamAsset]:
        # TODO: fail if we get dropped frames
        if indices_written:
            if not self._file:
                self._file = HDFFile(
                    / Path(await,
                for doc in self._file.stream_resources():
                    yield "stream_resource", doc
            for doc in self._file.stream_data(indices_written):
                yield "stream_datum", doc

    # Could put this function as default for StandardDetector
    async def close(self):
            False, wait=True, timeout=DEFAULT_TIMEOUT

The PandA StandardDetector implementation simply ties the component parts and its child devices together.

class HDFPanda(CommonPandaBlocks, StandardDetector):
    def __init__(
        prefix: str,
        path_provider: PathProvider,
        config_sigs: Sequence[SignalR] = (),
        name: str = "",
        self._prefix = prefix

        controller = PandaPcapController(pcap=self.pcap)
        writer = PandaHDFWriter(
            name_provider=lambda: name,

    async def connect(
        self, mock: bool = False, timeout: float = DEFAULT_TIMEOUT
    ) -> None:
        await fill_pvi_entries(self, self._prefix + "PVI", timeout=timeout, mock=mock)
        await super().connect(mock=mock, timeout=timeout)