Note

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

ophyd_async.core.DetectorWriter#

class ophyd_async.core.DetectorWriter[source]#

Bases: ABC

Methods

`close`	Close writer and wait for it to be finished
`collect_stream_docs`	Create Stream docs up to given number written
`get_indices_written`	Get the number of indices written
`observe_indices_written`	Yield each index as it is written
`open`	Open writer and wait for it to be ready for data.

abstract async open(multiplier: int = 1) → Dict[str, DataKey][source]#

Open writer and wait for it to be ready for data.

Args:

multiplier: Each StreamDatum index corresponds to this many: written exposures

Returns:

Output for describe()

abstract observe_indices_written(timeout=10.0) → AsyncGenerator[int, None][source]#: Yield each index as it is written

abstract async get_indices_written() → int[source]#: Get the number of indices written

abstract collect_stream_docs(indices_written: int) → AsyncIterator[Tuple[Literal['stream_resource'], StreamResource] | Tuple[Literal['stream_datum'], StreamDatum]][source]#: Create Stream docs up to given number written

abstract async close() → None[source]#: Close writer and wait for it to be finished