How to write a camera for continuous acquisition¶

In continuous acquisition, a camera takes frames one after another whether or not a plan runs, so a live view can show them at any time. On this page you write such a camera as an ophyd-async device, and plans read its latest frame into their events or write its frames to a file without stopping the acquisition. The plans that read are those of bluesky, and the one that writes is a few lines you add. The last section does the same for a camera served by an areaDetector IOC.

Prerequisites¶

You need a service that runs the camera, and How to write a service shows how to write one and declare it. The device on this page expects the service to serve these process variables under its prefix, over PVAccess:

Process variable	Holds
`Acquire`	`true` while the camera acquires. A write to it completes at once, without waiting for the acquisition to end.
`Exposure`	The exposure time, in seconds.
`Image`	The latest frame, as an `NTNDArray`. It holds a frame of the right shape from the start.
`ImageCount`	The number of frames taken so far. It goes up by one each time `Image` changes.
`Capture`	`true` while frames go to the file. The service sets it back to `false` after `NumCapture` frames.
`NumCapture`	The number of frames to write, or 0 to write until `Capture` is set to `false`.
`FilePath`	The `file://` address of the next file. Writing it sets `NumCaptured` to 0.
`NumCaptured`	The number of frames written to the current file.

The service writes the frames into the dataset /data of an HDF5 file. PVAccess carries a frame together with its shape, which Channel Access can't, so the session speaks PVAccess; see Name the transport.

Write the camera¶

The camera reads the latest frame when a plan asks for one. A plan starts the acquisition if it was stopped and leaves it as it found it, which a small class does for the camera and, further down, for the writer:

@dataclass
class AcquireForPlan:
    """Start a camera for a plan, and leave it as the plan found it."""

    acquire: SignalRW[bool]
    was_acquiring: bool = False

    async def start(self) -> None:
        self.was_acquiring = await self.acquire.get_value()
        if not self.was_acquiring:
            await self.acquire.set(True)

    async def restore(self) -> None:
        if not self.was_acquiring:
            await self.acquire.set(False)

The camera itself reads the service's process variables:

class Camera(StandardReadable):
    """A camera that acquires continuously: plans read its latest frame."""

    def __init__(self, prefix: str, name: str = "") -> None:
        with self.add_children_as_readables(
            StandardReadableFormat.HINTED_UNCACHED_SIGNAL
        ):
            self.image = epics_signal_r(np.ndarray, f"pva://{prefix}Image")
        with self.add_children_as_readables(StandardReadableFormat.CONFIG_SIGNAL):
            self.exposure = epics_signal_rw(float, f"pva://{prefix}Exposure")
        self.acquire = epics_signal_rw(bool, f"pva://{prefix}Acquire")
        self.image_count = epics_signal_r(int, f"pva://{prefix}ImageCount")
        self.acquisition = AcquireForPlan(self.acquire)
        super().__init__(name=name)

    @AsyncStatus.wrap
    async def stage(self) -> None:
        await self.acquisition.start()
        await super().stage()

    @AsyncStatus.wrap
    async def unstage(self) -> None:
        await super().unstage()
        await self.acquisition.restore()

    @AsyncStatus.wrap
    async def trigger(self) -> None:
        seen = await self.image_count.get_value(cached=False)
        # frame seen + 1 may have started its exposure before this call
        await wait_for_value(self.image_count, lambda n: n >= seen + 2, timeout=10)

HINTED_UNCACHED_SIGNAL makes image a hint of the camera, so a plan reads it and a live view finds it without knowing its name. UNCACHED makes each read ask the service for the frame, instead of returning the last value the device received.
CONFIG_SIGNAL records the exposure time once, in the description of the run, so the events don't repeat it.
stage and unstage are what a plan calls before and after it uses the camera; see stage.
trigger waits for the second frame after it is called. The first one may have started its exposure before the call, while a motor was still moving, so it could show the sample in the wrong place.

Write the writer¶

The writer saves frames to a file while the acquisition goes on. It is an ophyd-async StandardDetector, which takes three parts: one prepares the number of frames, one starts and stops the capture, and one says where the frames go:

@dataclass
class CaptureTrigger(DetectorTriggerLogic):
    num_capture: SignalRW[int]

    async def prepare_internal(
        self, num: int, livetime: float, deadtime: float
    ) -> None:
        await self.num_capture.set(num)


@dataclass
class CaptureAcquire(DetectorAcquireLogic):
    capture: SignalRW[bool]
    num_capture: SignalRW[int]
    acquisition: AcquireForPlan

    async def ensure_ready(self) -> None:
        await self.acquisition.start()
        await self.capture.set(False)

    async def start_acquiring(self) -> None:
        await self.capture.set(True)

    async def wait_for_idle(self) -> None:
        if await self.num_capture.get_value():
            await wait_for_value(self.capture, False, timeout=None)
        else:
            await self.capture.set(False)

    async def ensure_stopped(self) -> None:
        await self.capture.set(False)
        await self.acquisition.restore()


@dataclass
class CaptureData(DetectorDataLogic):
    image: SignalR[np.ndarray]
    file_path: SignalRW[str]
    num_captured: SignalR[int]
    path_provider: PathProvider

    async def prepare_unbounded(self, datakey_name: str) -> StreamResourceDataProvider:
        info = self.path_provider(datakey_name)
        uri = f"{info.directory_uri}{info.filename}.h5"
        await self.file_path.set(uri)
        await wait_for_value(self.num_captured, 0, timeout=5)
        frame = (await self.image.describe())[self.image.name]
        shape = tuple(size or 0 for size in frame["shape"])
        return StreamResourceDataProvider(
            uri=uri,
            resources=[
                StreamResourceInfo(
                    data_key=datakey_name,
                    shape=shape,
                    chunk_shape=(1, *shape),
                    dtype_numpy=frame["dtype_numpy"],
                    parameters={"dataset": "/data"},
                )
            ],
            mimetype="application/x-hdf5",
            collections_written_signal=self.num_captured,
        )

CaptureTrigger sets NumCapture to the number of frames the plan asks for.
CaptureAcquire starts the acquisition when the writer is staged, and turns off a capture an earlier plan left on. It puts the acquisition back as it was when the writer is unstaged, through the same AcquireForPlan.
wait_for_idle waits until the service has written the frames asked for, or stops the capture at once when none were asked for.
CaptureData sets the file, then reads the shape and type of a frame from image, so the documents of the run describe what the file holds. It tells bluesky the frames are in /data of an HDF5 file, and how many are written so far.

The writer builds the three parts from its own process variables:

class CameraWriter(StandardDetector):
    """Write frames to a file while the camera acquires, and leave it acquiring."""

    def __init__(
        self, prefix: str, path_provider: PathProvider, name: str = ""
    ) -> None:
        self.acquire = epics_signal_rw(bool, f"pva://{prefix}Acquire")
        self.image = epics_signal_r(np.ndarray, f"pva://{prefix}Image")
        self.capture = epics_signal_rw(bool, f"pva://{prefix}Capture")
        self.num_capture = epics_signal_rw(int, f"pva://{prefix}NumCapture")
        self.file_path = epics_signal_rw(str, f"pva://{prefix}FilePath")
        self.num_captured = epics_signal_r(int, f"pva://{prefix}NumCaptured")
        super().__init__(name=name)
        self.add_detector_logics(
            CaptureTrigger(self.num_capture),
            CaptureAcquire(
                self.capture, self.num_capture, AcquireForPlan(self.acquire)
            ),
            CaptureData(self.image, self.file_path, self.num_captured, path_provider),
        )

A writer inside the camera is never staged

The plans of bluesky stage the device at the top of the tree each device belongs to. A writer held as an attribute of the camera is never staged itself, so a second write reuses what the first one prepared and times out. Declare the writer as a device of its own.

Declare them¶

Declare the camera and the writer beside the service, which gives both its prefix:

class MyApp(QtSession):
    config: ClassVar[dict[str, Any]] = {"services": {"transport": "pv-access"}}
    camera_service: Annotated[
        AsService, Launch("mylab.services.camera", ready="camera ready", prefix="CAM:")
    ]
    camera: Annotated[AsDevice[Camera], Declare(service="camera_service")]
    camera_writer: Annotated[AsDevice[CameraWriter], Declare(service="camera_service")]
    live: AsPresenter[LiveFrames]

    def wire(self) -> Iterator[Link]:
        for device in self.devices.values():
            for signal in hinted_arrays(device):
                yield signal, self.live.show


if __name__ == "__main__":
    MyApp().run()

wire links every hinted signal that holds an array to the live view, which the next section writes.

Show the frames live¶

A live view finds the frames through hints, so it works for any camera, including one whose image signal has another name. The function below returns the hinted signals of a device whose values are arrays:

def hinted_arrays(device: Device) -> list[SignalR[Any]]:
    """Return the hinted signals of `device` that hold arrays, such as images."""
    hinted = (
        set(device.hints.get("fields", [])) if isinstance(device, HasHints) else set()
    )
    return [
        signal
        for signal in walk_devices(device).values()
        if isinstance(signal, SignalR)
        and signal.name in hinted
        and signal.datatype is np.ndarray
    ]

The camera sends every frame it takes, and a screen needs far fewer, so the presenter passes a frame on only when enough time has gone by since the last one of the same signal:

class LiveFrames:
    """Pass on the latest frame of every camera, at most `rate` times a second."""

    sig_frame = Signal(str, object)

    def __init__(self, name: str, rate: float = 10.0) -> None:
        self.name = name
        self.period = 1 / rate
        self.shown: dict[str, float] = {}

    @slot
    def show(self, reading: dict[str, Reading[Any]]) -> None:
        now = time.monotonic()
        for key, value in reading.items():
            if now - self.shown.get(key, 0.0) >= self.period:
                self.shown[key] = now
                self.sig_frame.emit(key, value["value"])

A view connected to sig_frame gets the name of the signal and the frame. Frames arrive only while the camera acquires. A plan starts the acquisition for as long as it runs, and outside a plan you set acquire, for instance from a button of a view of your own.

Run plans¶

Plans of bluesky read the camera as they would any detector. Each one stages the camera, so it leaves the acquisition as it found it:

To	Run
Take frames in events	`bp.count([camera], num=3)`
Take a frame at each position of a motor	`bp.rel_scan([camera], motor, 0, 2, 3)`
Read the camera in a loop of your own	`bps.trigger_and_read([camera, motor])` inside `bpp.stage_wrapper(..., [camera])`
Write frames to a file only	`write_frames([camera_writer], 5)`

trigger_and_read doesn't stage the camera, so a loop of your own needs the stage_wrapper around it. bluesky has no plan that prepares a writer and declares its stream before collecting, which ophyd-async requires, so writing takes this plan:

def write_frames(
    writers: Sequence[StandardDetector], frames: int
) -> MsgGenerator[None]:
    """Write `frames` frames of each writer to a file while its camera runs."""

    def capture() -> MsgGenerator[None]:
        yield from bps.declare_stream(*writers, name="primary", collect=True)
        yield from bps.kickoff_all(*writers, wait=True)
        yield from bps.collect_while_completing(writers, writers, flush_period=0.5)

    def prepare_and_capture() -> MsgGenerator[None]:
        for writer in writers:
            yield from bps.prepare(
                writer, TriggerInfo(number_of_events=frames), wait=True
            )
        yield from bpp.run_wrapper(capture())

    yield from bpp.stage_wrapper(prepare_and_capture(), writers)

collect_while_completing sends the frames written so far every half second, so a long write shows progress in the documents of the run. bp.fly collects without declaring the stream first, and fails with Single nested data keys should be pre-declared.

Stage a device inside the plan that uses it

Run as a plan of its own, bps.stage(camera) leaves the camera staged, and the RunEngine unstages it when that plan ends without waiting for it to finish. The next plan then stages the camera while it is still being unstaged. Stage a device in the plan that uses it.

Use an areaDetector camera¶

An areaDetector IOC already does what the service above does, and ophyd-async drives one in continuous acquisition with ContAcqDetector. The device is a subclass of it, with two additions.

Needs an ophyd-async release after 0.21.3

ophyd-async 0.21.3 never sends the trigger that lets frames through to the file, so every plan on this section's camera times out. This section was checked against the main branch of ophyd-async on 24 October 2026, before a release with the fix was out.

The IOC needs these plugins, under the names ContAcqDetector looks for:

cam1:, the driver, which acquires in continuous mode.
CB1:, a circular buffer fed by the driver, which holds frames back until a plan asks for them.
HDF1:, an HDF5 writer fed by CB1:.
Pva1:, which serves the latest frame of the driver over PVAccess as Pva1:Image.

The device reads Pva1:Image as its hinted signal, and starts the acquisition when it is staged:

class SimCamera(ContAcqDetector):
    """An areaDetector camera that acquires continuously."""

    def __init__(
        self, prefix: str, path_provider: PathProvider, name: str = ""
    ) -> None:
        self.image = epics_signal_r(np.ndarray, f"pva://{prefix}Pva1:Image")
        super().__init__(prefix, ADWriterFactory.hdf(path_provider), name=name)
        self.add_detector_logics(PluginSignalDataLogic(self.driver, self.image))

    @AsyncStatus.wrap
    async def stage(self) -> None:
        self.was_acquiring = await self.driver.acquire.get_value()
        if not self.was_acquiring:
            await self.driver.image_mode.set(ADImageMode.CONTINUOUS)
            # in continuous mode the put completes only when acquisition stops
            await set_and_wait_for_value(
                self.driver.acquire, True, wait_for_set_completion=False
            )
        await super().stage()

    @AsyncStatus.wrap
    async def unstage(self) -> None:
        await super().unstage()
        if not self.was_acquiring:
            await self.driver.acquire.set(False)

PluginSignalDataLogic adds image to what the camera reads and to its hints, so the plans and the live view above find it as they find the image of the other camera.
The driver's acquire waits, by default, for the acquisition to end, which in continuous mode never happens. set_and_wait_for_value with wait_for_set_completion=False returns once the driver reports it is acquiring.
stage starts the acquisition only when it was stopped, because a second start while the first is still waiting stalls the connection to the IOC; see the warning below.

Declare the IOC as an attached service, and link the hinted arrays to the live view as before:

class MyApp(QtSession):
    detector_ioc: Annotated[AsService, Attach("SIM:")]
    camera: Annotated[AsDevice[SimCamera], Declare(service="detector_ioc")]
    live: AsPresenter[LiveFrames]

    def wire(self) -> Iterator[Link]:
        for device in self.devices.values():
            for signal in hinted_arrays(device):
                yield signal, self.live.show


if __name__ == "__main__":
    MyApp().run()

The camera is its own writer, so write_frames([camera], 5) writes frames during the acquisition, and bp.count, bp.rel_scan and the loop of the table run as they do above. Each of them also writes the frames it reads to the HDF5 file, since the camera always writes through HDF1:.

A second start stalls the connection

A Channel Access write that waits for its completion stays open until the IOC reports it complete, which a start in continuous mode never does. While one is open on a process variable, a second waiting write to the same one stops every reply on that connection, and the plans on the camera time out. Start the acquisition only when it is stopped, and stop it with driver.acquire.set(False), which doesn't wait.

The frame travels over PVAccess and everything else over Channel Access. If the IOC is on another network, the address you give Attach reaches Channel Access only, so also list the IOC in EPICS_PVA_ADDR_LIST.

The examples in full¶

The camera, the writer and the live view

"""The session of the guide "How to write a camera for continuous acquisition"."""

from __future__ import annotations

import time
from collections.abc import Iterator, Sequence  # noqa: TC003
from dataclasses import dataclass
from typing import Annotated, Any, ClassVar

import bluesky.plan_stubs as bps
import bluesky.preprocessors as bpp
import numpy as np
from bluesky.protocols import HasHints, Reading
from bluesky.utils import MsgGenerator  # noqa: TC002
from ophyd_async.core import (
    AsyncStatus,
    DetectorAcquireLogic,
    DetectorDataLogic,
    DetectorTriggerLogic,
    Device,
    PathProvider,
    SignalR,
    SignalRW,
    StandardDetector,
    StandardReadable,
    StandardReadableFormat,
    StreamResourceDataProvider,
    StreamResourceInfo,
    TriggerInfo,
    wait_for_value,
    walk_devices,
)
from ophyd_async.epics.core import epics_signal_r, epics_signal_rw
from psygnal import Signal

from redsun import AsDevice, AsPresenter, AsService, Declare, Launch, Link, slot
from redsun.qt import QtSession


@dataclass
class AcquireForPlan:
    """Start a camera for a plan, and leave it as the plan found it."""

    acquire: SignalRW[bool]
    was_acquiring: bool = False

    async def start(self) -> None:
        self.was_acquiring = await self.acquire.get_value()
        if not self.was_acquiring:
            await self.acquire.set(True)

    async def restore(self) -> None:
        if not self.was_acquiring:
            await self.acquire.set(False)


@dataclass
class CaptureTrigger(DetectorTriggerLogic):
    num_capture: SignalRW[int]

    async def prepare_internal(
        self, num: int, livetime: float, deadtime: float
    ) -> None:
        await self.num_capture.set(num)


@dataclass
class CaptureAcquire(DetectorAcquireLogic):
    capture: SignalRW[bool]
    num_capture: SignalRW[int]
    acquisition: AcquireForPlan

    async def ensure_ready(self) -> None:
        await self.acquisition.start()
        await self.capture.set(False)

    async def start_acquiring(self) -> None:
        await self.capture.set(True)

    async def wait_for_idle(self) -> None:
        if await self.num_capture.get_value():
            await wait_for_value(self.capture, False, timeout=None)
        else:
            await self.capture.set(False)

    async def ensure_stopped(self) -> None:
        await self.capture.set(False)
        await self.acquisition.restore()


@dataclass
class CaptureData(DetectorDataLogic):
    image: SignalR[np.ndarray]
    file_path: SignalRW[str]
    num_captured: SignalR[int]
    path_provider: PathProvider

    async def prepare_unbounded(self, datakey_name: str) -> StreamResourceDataProvider:
        info = self.path_provider(datakey_name)
        uri = f"{info.directory_uri}{info.filename}.h5"
        await self.file_path.set(uri)
        await wait_for_value(self.num_captured, 0, timeout=5)
        frame = (await self.image.describe())[self.image.name]
        shape = tuple(size or 0 for size in frame["shape"])
        return StreamResourceDataProvider(
            uri=uri,
            resources=[
                StreamResourceInfo(
                    data_key=datakey_name,
                    shape=shape,
                    chunk_shape=(1, *shape),
                    dtype_numpy=frame["dtype_numpy"],
                    parameters={"dataset": "/data"},
                )
            ],
            mimetype="application/x-hdf5",
            collections_written_signal=self.num_captured,
        )


class CameraWriter(StandardDetector):
    """Write frames to a file while the camera acquires, and leave it acquiring."""

    def __init__(
        self, prefix: str, path_provider: PathProvider, name: str = ""
    ) -> None:
        self.acquire = epics_signal_rw(bool, f"pva://{prefix}Acquire")
        self.image = epics_signal_r(np.ndarray, f"pva://{prefix}Image")
        self.capture = epics_signal_rw(bool, f"pva://{prefix}Capture")
        self.num_capture = epics_signal_rw(int, f"pva://{prefix}NumCapture")
        self.file_path = epics_signal_rw(str, f"pva://{prefix}FilePath")
        self.num_captured = epics_signal_r(int, f"pva://{prefix}NumCaptured")
        super().__init__(name=name)
        self.add_detector_logics(
            CaptureTrigger(self.num_capture),
            CaptureAcquire(
                self.capture, self.num_capture, AcquireForPlan(self.acquire)
            ),
            CaptureData(self.image, self.file_path, self.num_captured, path_provider),
        )


class Camera(StandardReadable):
    """A camera that acquires continuously: plans read its latest frame."""

    def __init__(self, prefix: str, name: str = "") -> None:
        with self.add_children_as_readables(
            StandardReadableFormat.HINTED_UNCACHED_SIGNAL
        ):
            self.image = epics_signal_r(np.ndarray, f"pva://{prefix}Image")
        with self.add_children_as_readables(StandardReadableFormat.CONFIG_SIGNAL):
            self.exposure = epics_signal_rw(float, f"pva://{prefix}Exposure")
        self.acquire = epics_signal_rw(bool, f"pva://{prefix}Acquire")
        self.image_count = epics_signal_r(int, f"pva://{prefix}ImageCount")
        self.acquisition = AcquireForPlan(self.acquire)
        super().__init__(name=name)

    @AsyncStatus.wrap
    async def stage(self) -> None:
        await self.acquisition.start()
        await super().stage()

    @AsyncStatus.wrap
    async def unstage(self) -> None:
        await super().unstage()
        await self.acquisition.restore()

    @AsyncStatus.wrap
    async def trigger(self) -> None:
        seen = await self.image_count.get_value(cached=False)
        # frame seen + 1 may have started its exposure before this call
        await wait_for_value(self.image_count, lambda n: n >= seen + 2, timeout=10)


def write_frames(
    writers: Sequence[StandardDetector], frames: int
) -> MsgGenerator[None]:
    """Write `frames` frames of each writer to a file while its camera runs."""

    def capture() -> MsgGenerator[None]:
        yield from bps.declare_stream(*writers, name="primary", collect=True)
        yield from bps.kickoff_all(*writers, wait=True)
        yield from bps.collect_while_completing(writers, writers, flush_period=0.5)

    def prepare_and_capture() -> MsgGenerator[None]:
        for writer in writers:
            yield from bps.prepare(
                writer, TriggerInfo(number_of_events=frames), wait=True
            )
        yield from bpp.run_wrapper(capture())

    yield from bpp.stage_wrapper(prepare_and_capture(), writers)


def hinted_arrays(device: Device) -> list[SignalR[Any]]:
    """Return the hinted signals of `device` that hold arrays, such as images."""
    hinted = (
        set(device.hints.get("fields", [])) if isinstance(device, HasHints) else set()
    )
    return [
        signal
        for signal in walk_devices(device).values()
        if isinstance(signal, SignalR)
        and signal.name in hinted
        and signal.datatype is np.ndarray
    ]


class LiveFrames:
    """Pass on the latest frame of every camera, at most `rate` times a second."""

    sig_frame = Signal(str, object)

    def __init__(self, name: str, rate: float = 10.0) -> None:
        self.name = name
        self.period = 1 / rate
        self.shown: dict[str, float] = {}

    @slot
    def show(self, reading: dict[str, Reading[Any]]) -> None:
        now = time.monotonic()
        for key, value in reading.items():
            if now - self.shown.get(key, 0.0) >= self.period:
                self.shown[key] = now
                self.sig_frame.emit(key, value["value"])


class MyApp(QtSession):
    config: ClassVar[dict[str, Any]] = {"services": {"transport": "pv-access"}}
    camera_service: Annotated[
        AsService, Launch("mylab.services.camera", ready="camera ready", prefix="CAM:")
    ]
    camera: Annotated[AsDevice[Camera], Declare(service="camera_service")]
    camera_writer: Annotated[AsDevice[CameraWriter], Declare(service="camera_service")]
    live: AsPresenter[LiveFrames]

    def wire(self) -> Iterator[Link]:
        for device in self.devices.values():
            for signal in hinted_arrays(device):
                yield signal, self.live.show


if __name__ == "__main__":
    MyApp().run()

The areaDetector camera

"""The areaDetector session of "How to write a camera for continuous acquisition"."""

from __future__ import annotations

from collections.abc import Iterator  # noqa: TC003
from typing import Annotated

import numpy as np
from ophyd_async.core import AsyncStatus, PathProvider, set_and_wait_for_value
from ophyd_async.epics.adcore import (
    ADImageMode,
    ADWriterFactory,
    ContAcqDetector,
    PluginSignalDataLogic,
)
from ophyd_async.epics.core import epics_signal_r

from docs.examples.continuous_camera import LiveFrames, hinted_arrays
from redsun import AsDevice, AsPresenter, AsService, Attach, Declare, Link
from redsun.qt import QtSession


class SimCamera(ContAcqDetector):
    """An areaDetector camera that acquires continuously."""

    def __init__(
        self, prefix: str, path_provider: PathProvider, name: str = ""
    ) -> None:
        self.image = epics_signal_r(np.ndarray, f"pva://{prefix}Pva1:Image")
        super().__init__(prefix, ADWriterFactory.hdf(path_provider), name=name)
        self.add_detector_logics(PluginSignalDataLogic(self.driver, self.image))

    @AsyncStatus.wrap
    async def stage(self) -> None:
        self.was_acquiring = await self.driver.acquire.get_value()
        if not self.was_acquiring:
            await self.driver.image_mode.set(ADImageMode.CONTINUOUS)
            # in continuous mode the put completes only when acquisition stops
            await set_and_wait_for_value(
                self.driver.acquire, True, wait_for_set_completion=False
            )
        await super().stage()

    @AsyncStatus.wrap
    async def unstage(self) -> None:
        await super().unstage()
        if not self.was_acquiring:
            await self.driver.acquire.set(False)


class MyApp(QtSession):
    detector_ioc: Annotated[AsService, Attach("SIM:")]
    camera: Annotated[AsDevice[SimCamera], Declare(service="detector_ioc")]
    live: AsPresenter[LiveFrames]

    def wire(self) -> Iterator[Link]:
        for device in self.devices.values():
            for signal in hinted_arrays(device):
                yield signal, self.live.show


if __name__ == "__main__":
    MyApp().run()