Multi-Dimensional Acquisition

The mda attribute on the CMMCorePlus object provides a high-level interface for running multi-dimensional microscopy experiments.

pymmcore_plus.mda.MDARunner

Object that executes a multi-dimensional experiment using an MDAEngine.

This object is available at CMMCorePlus.mda.

This is the main object that runs a multi-dimensional experiment; it does so by driving an acquisition engine that implements the PMDAEngine protocol. It emits signals at specific times during the experiment (see PMDASignaler for details on the signals that are available to connect to and when they are emitted).

The state machine modeled by this runner is as follows:

stateDiagram-v2
    [*] --> IDLE
    IDLE --> PREPARING : run()
    PREPARING --> running : sequence ready
    state running {
        WAITING --> ACQUIRING : next event ready
        ACQUIRING --> WAITING : event done
        WAITING --> PAUSED : <code>set_paused(True)</code>
        PAUSED --> WAITING : <code>set_paused(False)</code>
        ACQUIRING --> PAUSED : <code>set_paused(True)</code></br>(after event done)
    }
    running --> FINISHING : <code>cancel()</code>
    running --> FINISHING : all events exhausted
    FINISHING --> IDLE : cleanup done

You can query the current state of the runner using the status property, which returns a snapshot of the runner's current state, including the current phase of the acquisition, whether a cancel or pause has been requested, and the reason for finishing (once the acquisition is finished).

Source code in pymmcore_plus/mda/_runner.py

class MDARunner:
    """Object that executes a multi-dimensional experiment using an MDAEngine.

    This object is available at [`CMMCorePlus.mda`][pymmcore_plus.CMMCorePlus.mda].

    This is the main object that runs a multi-dimensional experiment; it does so by
    driving an acquisition engine that implements the
    [`PMDAEngine`][pymmcore_plus.mda.PMDAEngine] protocol.  It emits signals at specific
    times during the experiment (see
    [`PMDASignaler`][pymmcore_plus.mda.events.PMDASignaler] for details on the signals
    that are available to connect to and when they are emitted).

    The state machine modeled by this runner is as follows:

    ```mermaid
    stateDiagram-v2
        [*] --> IDLE
        IDLE --> PREPARING : run()
        PREPARING --> running : sequence ready
        state running {
            WAITING --> ACQUIRING : next event ready
            ACQUIRING --> WAITING : event done
            WAITING --> PAUSED : <code>set_paused(True)</code>
            PAUSED --> WAITING : <code>set_paused(False)</code>
            ACQUIRING --> PAUSED : <code>set_paused(True)</code></br>(after event done)
        }
        running --> FINISHING : <code>cancel()</code>
        running --> FINISHING : all events exhausted
        FINISHING --> IDLE : cleanup done
    ```

    You can query the current state of the runner using the
    [`status`][pymmcore_plus.mda.MDARunner.status] property, which returns a snapshot
    of the runner's current state, including the current phase of the acquisition,
    whether a cancel or pause has been requested, and the reason for finishing
    (once the acquisition is finished).
    """

    def __init__(self) -> None:
        self._engine: PMDAEngine | None = None
        self._signals = _get_auto_MDA_callback_class()()
        self._lock = threading.Lock()
        self._state: RunState = RunState.IDLE
        self._finish_reason: FinishReason | None = None
        self._cancel_requested: bool = False
        self._pause_requested: bool = False
        self._paused_time: float = 0
        self._pause_interval: float = 0.1  # sec to wait between checking pause state
        self._handlers: WeakSet[SupportsFrameReady] = WeakSet()
        self._sink: SinkProtocol | None = None
        self._sequence: MDASequence | None = None
        # timer for the full sequence, reset only once at the beginning of the sequence
        self._sequence_t0: float = 0.0
        # event clock, reset whenever `event.reset_event_timer` is True
        self._t0: float = 0.0

    def set_engine(self, engine: PMDAEngine) -> PMDAEngine | None:
        """Set the [`PMDAEngine`][pymmcore_plus.mda.PMDAEngine] to use for the MDA run."""  # noqa: E501
        # MagicMock on py312 no longer satisfies isinstance ... so we explicitly
        # allow it here just for the sake of testing.
        if not isinstance(engine, (PMDAEngine, MagicMock)):
            raise TypeError("Engine does not conform to the Engine protocol.")

        if self.is_running():  # pragma: no cover
            raise RuntimeError(
                "Cannot register a new engine when the current engine is running "
                "an acquisition. Please cancel the current engine's acquisition "
                "before registering"
            )

        old_engine, self._engine = self._engine, engine
        return old_engine

    # NOTE:
    # this return annotation is a lie, since the user can set it to their own engine.
    # but in MOST cases, this is the engine that will be used by default, so it's
    # convenient for IDEs to point to this rather than the abstract protocol.
    @property
    def engine(self) -> MDAEngine | None:
        """The [`PMDAEngine`][pymmcore_plus.mda.PMDAEngine] that is currently being used."""  # noqa: E501
        return self._engine  # type: ignore

    @property
    def events(self) -> PMDASignaler:
        """Signals that are emitted during the MDA run.

        See [`PMDASignaler`][pymmcore_plus.mda.PMDASignaler] for details on the
        signals that are available to connect to.
        """
        return self._signals

    @property
    def status(self) -> RunnerStatus:
        """Snapshot of the runner's current status."""
        with self._lock:
            return RunnerStatus(
                phase=self._state,
                finish_reason=self._finish_reason,
                cancel_requested=self._cancel_requested,
                pause_requested=self._pause_requested,
            )

    def is_running(self) -> bool:
        """Return True if an acquisition is currently underway.

        This will return True at any point between the emission of the
        [`sequenceStarted`][pymmcore_plus.mda.PMDASignaler.sequenceStarted] and
        [`sequenceFinished`][pymmcore_plus.mda.PMDASignaler.sequenceFinished] signals,
        including when the acquisition is currently paused.

        Returns
        -------
        bool
            Whether an acquisition is underway.
        """
        return self._state not in (RunState.IDLE, RunState.FINISHING)

    def is_paused(self) -> bool:
        """Return True if the acquisition is currently paused.

        Use `set_paused` to change the paused state.

        Returns
        -------
        bool
            Whether the current acquisition is paused.
        """
        return self._state == RunState.PAUSED

    def cancel(self) -> None:
        """Cancel the currently running acquisition.

        This is a no-op if no acquisition is currently running.
        If an acquisition is running then this will cancel the acquisition and
        a sequenceCanceled signal, followed by a sequenceFinished signal will
        be emitted.
        """
        with self._lock:
            if self._state in (RunState.IDLE, RunState.FINISHING):
                return
            self._paused_time = 0
            if self._state == RunState.ACQUIRING:
                # defer to event boundary
                self._cancel_requested = True
            else:
                # WAITING, PREPARING, or PAUSED → immediate transition
                self._finish_reason = FinishReason.CANCELED
                self._state = RunState.FINISHING

    @deprecated("Use `set_paused(paused)` instead.", category=DeprecationWarning)
    def toggle_pause(self) -> None:
        """Toggle the paused state of the current acquisition.

        !!!warning "Deprecated"
            Use [`set_paused`][pymmcore_plus.mda.MDARunner.set_paused] instead.
        """
        self.set_paused(not (self.is_paused() or self._pause_requested))

    def set_paused(self, paused: bool) -> None:
        """Set the paused state of the current acquisition.

        This is a no-op if the acquisition is already in the requested state,
        or if no acquisition is currently underway.

        Parameters
        ----------
        paused : bool
            Whether to pause (True) or unpause (False) the acquisition.
        """
        with self._lock:
            if self._state == RunState.WAITING:
                if not paused:
                    return
                self._state = RunState.PAUSED
            elif self._state == RunState.PAUSED:
                if paused:
                    return
                self._state = RunState.WAITING
            elif self._state == RunState.ACQUIRING:
                if self._pause_requested == paused:
                    return
                self._pause_requested = paused
            else:
                return
        self._signals.sequencePauseToggled.emit(paused)

    def run(
        self,
        events: Iterable[MDAEvent],
        *,
        output: SingleOutput | Sequence[SingleOutput] | None = None,
        overwrite: bool = False,
    ) -> None:
        """Run the multi-dimensional acquisition defined by `sequence`.

        Most users should not use this directly as it will block further
        execution. Instead, use the
        [`CMMCorePlus.run_mda`][pymmcore_plus.CMMCorePlus.run_mda] method which will
        run on a thread.

        !!!important

            If `output` is a sequence, it may contain nor more than one
            `str | Path | AcquisitionSettings`.

        Parameters
        ----------
        events : Iterable[MDAEvent]
            An iterable of `useq.MDAEvents` objects to execute.
        output : SingleOutput | Sequence[SingleOutput] | None, optional
            The output handler(s) to use.  If None, no output will be saved.

            The value may be either a single output or a sequence of outputs,
            where a "single output" can be any of the following:

            1. A string or `Path`. This value is passed directly to
                [`ome_writers.AcquisitionSettings`][], and the extension can be used
                to determine the file format to create:
                - `[.ome].zarr` will result in an OME-Zarr at the specified path
                - `[.ome].tiff` will result in an OME-TIFF at the specified, or, if
                  multiple positions are acquired, the output path will be treated as a
                  directory with individual OME-TIFF files for each position.

            1. An [`ome_writers.AcquisitionSettings`][], object which allows full
              control over many parameters.  See [ome_writers
              documentation](https://pymmcore-plus.github.io/ome-writers/usage/) for
              details.

            1. A handler object that implements the `DataHandler` protocol, currently
              meaning it has a `frameReady` method.  See `mda_listeners_connected`
              for more details.

            During the course of the sequence, the `get_view` method can be used to get
            an array-like view of the current data sink, if the sink supports it.

        overwrite : bool, optional
            Whether to overwrite existing files *when output is a `str` or `Path`*.
            Ignored in all other cases.  Default is False.
        """
        error = None
        sequence = events if isinstance(events, MDASequence) else GeneratorMDASequence()
        handlers, sink = self._coerce_outputs(output, overwrite=overwrite)
        self._sink = sink
        with self._handlers_connected(handlers):
            # NOTE: it's important that `_prepare_to_run` and `_finish_run` are
            # called inside the context manager, since the `mda_listeners_connected`
            # context manager expects to see both of those signals.
            try:
                engine = self._prepare_to_run(sequence)
                self._run(engine, events)
            except Exception as e:
                error = e
                with self._lock:
                    if self._finish_reason is None:
                        self._finish_reason = FinishReason.ERRORED
            with exceptions_logged():
                self._finish_run(sequence)
        if error is not None:
            raise error

    def get_view(self) -> SinkView | None:
        """Array-like view of the current data sink, if it exists."""
        if self._sink is None:  # pragma: no cover
            return None
        return self._sink.get_view()

    @deprecated(
        "`get_output_handlers` is deprecated, and no full replacement planned. "
        "Use `get_sink()` instead, to monitor the data as it is being acquired.",
        category=DeprecationWarning,
    )
    def get_output_handlers(self) -> tuple[SupportsFrameReady, ...]:
        """Return the data handlers that are currently connected.

        Output handlers are connected by passing them to the `output` parameter of the
        `run` method; the run method accepts objects with a `frameReady` method *or*
        strings representing paths.  If a string is passed, a handler will be created
        internally.

        This method returns a tuple of currently connected handlers, including those
        that were explicitly passed to `run()`, as well as those that were created based
        on file paths.  Internally, handlers are held by weak references, so if you want
        the handler to persist, you must keep a reference to it.  The only guaranteed
        API that the handler will have is the `frameReady` method, but it could be any
        user-defined object that implements that method.

        Handlers are cleared each time `run()` is called, (but not at the end
        of the sequence).

        Returns
        -------
        tuple[SupportsFrameReady, ...]
            Tuple of objects that (minimally) support the `frameReady` method.
        """
        return tuple(self._handlers)

    def seconds_elapsed(self) -> float:
        """Return the number of seconds since the start of the acquisition."""
        return time.perf_counter() - self._sequence_t0

    def event_seconds_elapsed(self) -> float:
        """Return the number of seconds on the "event clock".

        This is the time since either the start of the acquisition or the last
        event with `reset_event_timer` set to `True`.
        """
        return time.perf_counter() - self._t0

    @staticmethod
    def _coerce_outputs(
        output: SingleOutput | Sequence[SingleOutput] | None,
        overwrite: bool = False,
    ) -> tuple[list[SupportsFrameReady], SinkProtocol | None]:
        """Normalize and validate output into a list of frameReady handlers, and a sink.

        Returns
        -------
        tuple[list[SupportsFrameReady], SinkProtocol | None]
            A tuple of (handlers, sink).
        """
        sink: SinkProtocol | None = None
        handlers: list[SupportsFrameReady] = []

        if output is None:
            return handlers, sink

        if isinstance(output, (str, Path)) or not isinstance(output, Sequence):
            output = [output]

        for item in output:
            if isinstance(item, (str, Path, AcquisitionSettings)):
                if sink is not None:
                    raise NotImplementedError(
                        "Only one AcquisitionSettings object or path may be provided "
                        "as output.  Open a feature request if you would like to see "
                        "support for multiple data sinks."
                    )
                sink = _OmeWritersSink.from_output(item, overwrite=overwrite)
            else:
                if not callable(getattr(item, "frameReady", None)):
                    raise TypeError(
                        "Output handlers must have a callable frameReady method. "
                        f"Got {item} with type {type(item)}."
                    )
                handlers.append(item)
        return handlers, sink

    def _handlers_connected(
        self, handlers: Sequence[SupportsFrameReady]
    ) -> AbstractContextManager:
        """Context in which output handlers are connected to the frameReady signal."""
        self._handlers.clear()
        self._handlers.update(handlers)
        if not handlers:
            return nullcontext()
        return mda_listeners_connected(*handlers, mda_events=self._signals)

    def _run(self, engine: PMDAEngine, events: Iterable[MDAEvent]) -> None:
        """Main execution of events, inside the try/except block of `run`."""
        teardown_event = getattr(engine, "teardown_event", lambda e: None)
        if isinstance(events, Iterator):
            # if an iterator is passed directly, then we use that iterator
            # instead of the engine's event_iterator.  Directly passing an iterator
            # is an advanced use case, (for example, `iter(Queue(), None)` for event-
            # driven acquisition) and we don't want the engine to interfere with it.
            event_iterator = iter
        else:
            event_iterator = getattr(engine, "event_iterator", iter)
        _events: Iterator[MDAEvent] = event_iterator(events)
        self._reset_event_timer()
        self._sequence_t0 = self._t0

        _append: Callable | None = self._sink.append if self._sink is not None else None
        _skip: Callable | None = self._sink.skip if self._sink is not None else None
        _emit_event_started = self._signals.eventStarted.emit
        _emit_frame_ready = self._signals.frameReady.emit
        for event in _events:
            if event.reset_event_timer:
                self._reset_event_timer()

            if self._wait_until_event(event):
                break

            with self._lock:
                self._state = RunState.ACQUIRING

            _emit_event_started(event)
            logger.info("%s", event)

            try:
                engine.setup_event(event)
            except SkipEvent as exc:
                logger.info("%s", exc)
                if _skip is not None and exc.num_frames > 0:
                    _skip(frames=exc.num_frames * self._n_cameras)
                teardown_event(event)
            else:
                try:
                    runner_time_ms = self.seconds_elapsed() * 1000
                    # this is a bit of a hack to pass the time into the engine
                    # it is used for intra-event time calculations inside the
                    # engine. we pop it off after the event is executed.
                    event.metadata["runner_t0"] = self._sequence_t0
                    output = engine.exec_event(event) or ()
                    for payload in self._iter_exec_output(output):
                        if isinstance(payload, int):
                            if _skip is not None:
                                _skip(frames=payload)
                            continue
                        img, sub_event, meta = payload
                        sub_event.metadata.pop("runner_t0", None)
                        if "runner_time_ms" not in meta:
                            meta["runner_time_ms"] = runner_time_ms
                        if _append is not None:
                            _append(img, sub_event, meta)
                        with exceptions_logged():
                            _emit_frame_ready(img, sub_event, meta)
                finally:
                    teardown_event(event)

            # event boundary: resolve deferred flags
            with self._lock:
                if self._cancel_requested:
                    self._cancel_requested = False
                    self._finish_reason = FinishReason.CANCELED
                    self._state = RunState.FINISHING
                    break

                if self._pause_requested:
                    self._pause_requested = False
                    self._state = RunState.PAUSED
                    # signal was already emitted in set_paused()

                if self._state != RunState.PAUSED:
                    self._state = RunState.WAITING
        else:
            with self._lock:
                self._finish_reason = FinishReason.COMPLETED

    def _iter_exec_output(self, iterable: Iterable) -> Iterator[PImagePayload | int]:
        """Iterate over exec_event output, sending cancel/pause signals to generators.

        This allows the runner to communicate with generator-based engines
        (like exec_sequenced_event) without the engine needing to know about
        runner internals. Signals are sent via generator.send().

        Consecutive None payloads (missing frames) are coalesced into a single
        int representing the skip count.

        Works with any iterable - if it's not a generator or doesn't handle
        signals, they're simply ignored.
        """
        gen = iter(iterable)
        is_generator = isinstance(gen, types.GeneratorType)

        def _advance() -> PImagePayload | None:
            if not is_generator:  # pragma: no cover
                return next(gen)  # type: ignore[no-any-return]
            if self._cancel_requested or self._state == RunState.FINISHING:
                return gen.send("cancel")  # type: ignore
            return gen.send("pause" if self._pause_requested else None)  # type: ignore

        skip_count = 0
        try:
            item = next(gen)
            while True:
                if item is None:
                    skip_count += 1
                else:
                    if skip_count:
                        yield skip_count
                        skip_count = 0
                    yield item
                item = _advance()
        except StopIteration:
            if skip_count:
                yield skip_count

    def _prepare_to_run(self, sequence: MDASequence) -> PMDAEngine:
        """Set up for the MDA run.

        Parameters
        ----------
        sequence : MDASequence
            The sequence of events to run.
        """
        if not self._engine:  # pragma: no cover
            raise RuntimeError("No MDAEngine set.")

        with self._lock:
            self._state = RunState.PREPARING
            self._finish_reason = None
            self._cancel_requested = False
            self._pause_requested = False
        self._paused_time = 0.0
        self._sequence = sequence

        meta = self._engine.setup_sequence(sequence)

        # extract camera multiplier for sink skip accounting
        self._n_cameras = 1
        if meta and (infos := meta.get("image_infos")):
            self._n_cameras = infos[0].get("num_camera_adapter_channels", 1)

        if self._sink is not None:
            self._sink.setup(sequence, meta)

        with self._lock:
            if self._state != RunState.FINISHING:
                self._state = RunState.WAITING
        self._signals.sequenceStarted.emit(sequence, meta or {})
        logger.info("MDA Started: %s", sequence)
        return self._engine

    def _reset_event_timer(self) -> None:
        self._t0 = time.perf_counter()  # reference time, in seconds

    def _wait_until_event(self, event: MDAEvent) -> bool:
        """Wait until the event's min start time, checking for pauses/cancellations.

        Parameters
        ----------
        event : MDAEvent
            The event to wait for.

        Returns
        -------
        bool
            Whether the MDA was cancelled while waiting.
        """
        # NOTE: mypy narrows self._state after each check, but cancel() and
        # set_paused() mutate it from other threads, so all checks are reachable.
        if self._state == RunState.FINISHING:
            return True

        # pause loop (for deferred pause from ACQUIRING boundary)
        while self._state == RunState.PAUSED:
            self._paused_time += self._pause_interval
            time.sleep(self._pause_interval)
        if self._state == RunState.FINISHING:  # type: ignore[comparison-overlap]
            return True

        if event.min_start_time:
            go_at = event.min_start_time + self._paused_time
            remaining = go_at - self.event_seconds_elapsed()
            if remaining > 0.5:
                logger.info(
                    "Waiting %s until the next event",
                    _format_wait_time(remaining),
                )
            while remaining > 0:
                self._signals.awaitingEvent.emit(event, remaining)
                while self._state == RunState.PAUSED:  # type: ignore[comparison-overlap]
                    self._paused_time += self._pause_interval
                    remaining += self._pause_interval
                    time.sleep(self._pause_interval)
                if self._state == RunState.FINISHING:  # type: ignore[comparison-overlap]
                    return True
                time.sleep(min(remaining, 0.5))
                remaining = go_at - self.event_seconds_elapsed()

        return self._state == RunState.FINISHING  # type: ignore[comparison-overlap]

    def _finish_run(self, sequence: MDASequence) -> None:
        """To be called at the end of an acquisition.

        Parameters
        ----------
        sequence : MDASequence
            The sequence that was finished.
        """
        with self._lock:
            self._state = RunState.FINISHING
            if self._finish_reason is None:
                self._finish_reason = FinishReason.COMPLETED
            finish_reason = self._finish_reason

        if self._sink is not None:
            try:
                self._sink.close()
            except Exception as e:  # pragma: no cover
                logger.error("Error closing data sink: %s", e)

        if hasattr(self._engine, "teardown_sequence"):
            self._engine.teardown_sequence(sequence)  # type: ignore

        if finish_reason == FinishReason.CANCELED:
            logger.warning("MDA Canceled: %s", sequence)
            self._signals.sequenceCanceled.emit(sequence)

        logger.info("MDA Finished: %s", sequence)
        self._signals.sequenceFinished.emit(sequence)
        with self._lock:
            self._state = RunState.IDLE

cancel() -> None

Cancel the currently running acquisition.

This is a no-op if no acquisition is currently running. If an acquisition is running then this will cancel the acquisition and a sequenceCanceled signal, followed by a sequenceFinished signal will be emitted.

Source code in pymmcore_plus/mda/_runner.py

def cancel(self) -> None:
    """Cancel the currently running acquisition.

    This is a no-op if no acquisition is currently running.
    If an acquisition is running then this will cancel the acquisition and
    a sequenceCanceled signal, followed by a sequenceFinished signal will
    be emitted.
    """
    with self._lock:
        if self._state in (RunState.IDLE, RunState.FINISHING):
            return
        self._paused_time = 0
        if self._state == RunState.ACQUIRING:
            # defer to event boundary
            self._cancel_requested = True
        else:
            # WAITING, PREPARING, or PAUSED → immediate transition
            self._finish_reason = FinishReason.CANCELED
            self._state = RunState.FINISHING

engine() -> MDAEngine | None property

The PMDAEngine that is currently being used.

Source code in pymmcore_plus/mda/_runner.py

@property
def engine(self) -> MDAEngine | None:
    """The [`PMDAEngine`][pymmcore_plus.mda.PMDAEngine] that is currently being used."""  # noqa: E501
    return self._engine  # type: ignore

event_seconds_elapsed() -> float

Return the number of seconds on the "event clock".

This is the time since either the start of the acquisition or the last event with reset_event_timer set to True.

Source code in pymmcore_plus/mda/_runner.py

def event_seconds_elapsed(self) -> float:
    """Return the number of seconds on the "event clock".

    This is the time since either the start of the acquisition or the last
    event with `reset_event_timer` set to `True`.
    """
    return time.perf_counter() - self._t0

events() -> PMDASignaler property

Signals that are emitted during the MDA run.

See PMDASignaler for details on the signals that are available to connect to.

Source code in pymmcore_plus/mda/_runner.py

@property
def events(self) -> PMDASignaler:
    """Signals that are emitted during the MDA run.

    See [`PMDASignaler`][pymmcore_plus.mda.PMDASignaler] for details on the
    signals that are available to connect to.
    """
    return self._signals

get_output_handlers() -> tuple[SupportsFrameReady, ...]

Return the data handlers that are currently connected.

Output handlers are connected by passing them to the output parameter of the run method; the run method accepts objects with a frameReady method or strings representing paths. If a string is passed, a handler will be created internally.

This method returns a tuple of currently connected handlers, including those that were explicitly passed to run(), as well as those that were created based on file paths. Internally, handlers are held by weak references, so if you want the handler to persist, you must keep a reference to it. The only guaranteed API that the handler will have is the frameReady method, but it could be any user-defined object that implements that method.

Handlers are cleared each time run() is called, (but not at the end of the sequence).

Returns:

Type	Description
tuple[SupportsFrameReady, ...]	Tuple of objects that (minimally) support the frameReady method.

Source code in pymmcore_plus/mda/_runner.py

@deprecated(
    "`get_output_handlers` is deprecated, and no full replacement planned. "
    "Use `get_sink()` instead, to monitor the data as it is being acquired.",
    category=DeprecationWarning,
)
def get_output_handlers(self) -> tuple[SupportsFrameReady, ...]:
    """Return the data handlers that are currently connected.

    Output handlers are connected by passing them to the `output` parameter of the
    `run` method; the run method accepts objects with a `frameReady` method *or*
    strings representing paths.  If a string is passed, a handler will be created
    internally.

    This method returns a tuple of currently connected handlers, including those
    that were explicitly passed to `run()`, as well as those that were created based
    on file paths.  Internally, handlers are held by weak references, so if you want
    the handler to persist, you must keep a reference to it.  The only guaranteed
    API that the handler will have is the `frameReady` method, but it could be any
    user-defined object that implements that method.

    Handlers are cleared each time `run()` is called, (but not at the end
    of the sequence).

    Returns
    -------
    tuple[SupportsFrameReady, ...]
        Tuple of objects that (minimally) support the `frameReady` method.
    """
    return tuple(self._handlers)

get_view() -> SinkView | None

Array-like view of the current data sink, if it exists.

Source code in pymmcore_plus/mda/_runner.py

def get_view(self) -> SinkView | None:
    """Array-like view of the current data sink, if it exists."""
    if self._sink is None:  # pragma: no cover
        return None
    return self._sink.get_view()

is_paused() -> bool

Return True if the acquisition is currently paused.

Use set_paused to change the paused state.

Returns:

Type	Description
bool	Whether the current acquisition is paused.

Source code in pymmcore_plus/mda/_runner.py

def is_paused(self) -> bool:
    """Return True if the acquisition is currently paused.

    Use `set_paused` to change the paused state.

    Returns
    -------
    bool
        Whether the current acquisition is paused.
    """
    return self._state == RunState.PAUSED

is_running() -> bool

Return True if an acquisition is currently underway.

This will return True at any point between the emission of the sequenceStarted and sequenceFinished signals, including when the acquisition is currently paused.

Returns:

Type	Description
bool	Whether an acquisition is underway.

Source code in pymmcore_plus/mda/_runner.py

def is_running(self) -> bool:
    """Return True if an acquisition is currently underway.

    This will return True at any point between the emission of the
    [`sequenceStarted`][pymmcore_plus.mda.PMDASignaler.sequenceStarted] and
    [`sequenceFinished`][pymmcore_plus.mda.PMDASignaler.sequenceFinished] signals,
    including when the acquisition is currently paused.

    Returns
    -------
    bool
        Whether an acquisition is underway.
    """
    return self._state not in (RunState.IDLE, RunState.FINISHING)

run(events: Iterable[MDAEvent], *, output: SingleOutput | Sequence[SingleOutput] | None = None, overwrite: bool = False) -> None

Run the multi-dimensional acquisition defined by sequence.

Most users should not use this directly as it will block further execution. Instead, use the CMMCorePlus.run_mda method which will run on a thread.

Important

If output is a sequence, it may contain nor more than one str | Path | AcquisitionSettings.

Parameters:

Name	Type	Description	Default
events	Iterable[MDAEvent]	An iterable of useq.MDAEvents objects to execute.	required
output	SingleOutput | Sequence[SingleOutput] | None, optional	The output handler(s) to use. If None, no output will be saved. The value may be either a single output or a sequence of outputs, where a "single output" can be any of the following: A string or Path. This value is passed directly to ome_writers.AcquisitionSettings, and the extension can be used to determine the file format to create: [.ome].zarr will result in an OME-Zarr at the specified path [.ome].tiff will result in an OME-TIFF at the specified, or, if multiple positions are acquired, the output path will be treated as a directory with individual OME-TIFF files for each position. An ome_writers.AcquisitionSettings, object which allows full control over many parameters. See ome_writers documentation for details. A handler object that implements the DataHandler protocol, currently meaning it has a frameReady method. See mda_listeners_connected for more details. During the course of the sequence, the get_view method can be used to get an array-like view of the current data sink, if the sink supports it.	None

overwrite : bool, optional Whether to overwrite existing files when output is a str or Path. Ignored in all other cases. Default is False.

Source code in pymmcore_plus/mda/_runner.py

def run(
    self,
    events: Iterable[MDAEvent],
    *,
    output: SingleOutput | Sequence[SingleOutput] | None = None,
    overwrite: bool = False,
) -> None:
    """Run the multi-dimensional acquisition defined by `sequence`.

    Most users should not use this directly as it will block further
    execution. Instead, use the
    [`CMMCorePlus.run_mda`][pymmcore_plus.CMMCorePlus.run_mda] method which will
    run on a thread.

    !!!important

        If `output` is a sequence, it may contain nor more than one
        `str | Path | AcquisitionSettings`.

    Parameters
    ----------
    events : Iterable[MDAEvent]
        An iterable of `useq.MDAEvents` objects to execute.
    output : SingleOutput | Sequence[SingleOutput] | None, optional
        The output handler(s) to use.  If None, no output will be saved.

        The value may be either a single output or a sequence of outputs,
        where a "single output" can be any of the following:

        1. A string or `Path`. This value is passed directly to
            [`ome_writers.AcquisitionSettings`][], and the extension can be used
            to determine the file format to create:
            - `[.ome].zarr` will result in an OME-Zarr at the specified path
            - `[.ome].tiff` will result in an OME-TIFF at the specified, or, if
              multiple positions are acquired, the output path will be treated as a
              directory with individual OME-TIFF files for each position.

        1. An [`ome_writers.AcquisitionSettings`][], object which allows full
          control over many parameters.  See [ome_writers
          documentation](https://pymmcore-plus.github.io/ome-writers/usage/) for
          details.

        1. A handler object that implements the `DataHandler` protocol, currently
          meaning it has a `frameReady` method.  See `mda_listeners_connected`
          for more details.

        During the course of the sequence, the `get_view` method can be used to get
        an array-like view of the current data sink, if the sink supports it.

    overwrite : bool, optional
        Whether to overwrite existing files *when output is a `str` or `Path`*.
        Ignored in all other cases.  Default is False.
    """
    error = None
    sequence = events if isinstance(events, MDASequence) else GeneratorMDASequence()
    handlers, sink = self._coerce_outputs(output, overwrite=overwrite)
    self._sink = sink
    with self._handlers_connected(handlers):
        # NOTE: it's important that `_prepare_to_run` and `_finish_run` are
        # called inside the context manager, since the `mda_listeners_connected`
        # context manager expects to see both of those signals.
        try:
            engine = self._prepare_to_run(sequence)
            self._run(engine, events)
        except Exception as e:
            error = e
            with self._lock:
                if self._finish_reason is None:
                    self._finish_reason = FinishReason.ERRORED
        with exceptions_logged():
            self._finish_run(sequence)
    if error is not None:
        raise error

seconds_elapsed() -> float

Return the number of seconds since the start of the acquisition.

Source code in pymmcore_plus/mda/_runner.py

def seconds_elapsed(self) -> float:
    """Return the number of seconds since the start of the acquisition."""
    return time.perf_counter() - self._sequence_t0

set_engine(engine: PMDAEngine) -> PMDAEngine | None

Set the PMDAEngine to use for the MDA run.

Source code in pymmcore_plus/mda/_runner.py

def set_engine(self, engine: PMDAEngine) -> PMDAEngine | None:
    """Set the [`PMDAEngine`][pymmcore_plus.mda.PMDAEngine] to use for the MDA run."""  # noqa: E501
    # MagicMock on py312 no longer satisfies isinstance ... so we explicitly
    # allow it here just for the sake of testing.
    if not isinstance(engine, (PMDAEngine, MagicMock)):
        raise TypeError("Engine does not conform to the Engine protocol.")

    if self.is_running():  # pragma: no cover
        raise RuntimeError(
            "Cannot register a new engine when the current engine is running "
            "an acquisition. Please cancel the current engine's acquisition "
            "before registering"
        )

    old_engine, self._engine = self._engine, engine
    return old_engine

set_paused(paused: bool) -> None

Set the paused state of the current acquisition.

This is a no-op if the acquisition is already in the requested state, or if no acquisition is currently underway.

Parameters:

Name	Type	Description	Default
paused	bool	Whether to pause (True) or unpause (False) the acquisition.	required

Source code in pymmcore_plus/mda/_runner.py

def set_paused(self, paused: bool) -> None:
    """Set the paused state of the current acquisition.

    This is a no-op if the acquisition is already in the requested state,
    or if no acquisition is currently underway.

    Parameters
    ----------
    paused : bool
        Whether to pause (True) or unpause (False) the acquisition.
    """
    with self._lock:
        if self._state == RunState.WAITING:
            if not paused:
                return
            self._state = RunState.PAUSED
        elif self._state == RunState.PAUSED:
            if paused:
                return
            self._state = RunState.WAITING
        elif self._state == RunState.ACQUIRING:
            if self._pause_requested == paused:
                return
            self._pause_requested = paused
        else:
            return
    self._signals.sequencePauseToggled.emit(paused)

status() -> RunnerStatus property

Snapshot of the runner's current status.

Source code in pymmcore_plus/mda/_runner.py

@property
def status(self) -> RunnerStatus:
    """Snapshot of the runner's current status."""
    with self._lock:
        return RunnerStatus(
            phase=self._state,
            finish_reason=self._finish_reason,
            cancel_requested=self._cancel_requested,
            pause_requested=self._pause_requested,
        )

toggle_pause() -> None

Toggle the paused state of the current acquisition.

Deprecated

Use set_paused instead.

Source code in pymmcore_plus/mda/_runner.py

@deprecated("Use `set_paused(paused)` instead.", category=DeprecationWarning)
def toggle_pause(self) -> None:
    """Toggle the paused state of the current acquisition.

    !!!warning "Deprecated"
        Use [`set_paused`][pymmcore_plus.mda.MDARunner.set_paused] instead.
    """
    self.set_paused(not (self.is_paused() or self._pause_requested))

pymmcore_plus.mda.PMDAEngine

Protocol that all MDA engines must implement.

event_iterator(events: Iterable[MDAEvent]) -> Iterator[MDAEvent]

Wrapper on the event iterator.

Optional.

This can be used to wrap the event iterator to perform any event merging (e.g. if the engine supports HardwareSequencing) or event modification. The default implementation is just iter(events).

Be careful when using this method. It is powerful and can result in unexpected event iteration if used incorrectly.

exec_event(event: MDAEvent) -> Iterable[PImagePayload | None] abstractmethod

Execute event.

This method is called after setup_event and is responsible for executing the event. The default assumption is to acquire an image, but more elaborate events will be possible.

Yields (image, event, metadata) tuples for each acquired frame. May yield None for frames that could not be acquired (e.g. partial hardware failure during a triggered sequence); the runner will call sink.skip(frames=1) for each None.

setup_event(event: MDAEvent) -> None abstractmethod

Prepare state of system (hardware, etc.) for event.

This method is called before each event in the sequence. It is responsible for preparing the state of the system for the event. The engine should be in a state where it can call exec_event without any additional preparation. (This means that the engine should perform any waits or blocks required for system state changes to complete.)

If the engine cannot set up the event (e.g. hardware failure), it may raise SkipEvent(num_frames) to tell the runner to skip this event and inform the data sink of the missing frames. If SkipEvent is raised, exec_event will NOT be called for this event, but teardown_event will be.

setup_sequence(sequence: MDASequence) -> SummaryMetaV1 | None abstractmethod

Setup state of system (hardware, etc.) before an MDA is run.

This method is called once at the beginning of a sequence.

teardown_event(event: MDAEvent) -> None

Teardown state of system (hardware, etc.) after event.

Optional.

If the engine provides this function, it will be called after exec_event to perform any cleanup or teardown required after the event has been executed.

teardown_sequence(sequence: MDASequence) -> None

Perform any teardown required after the sequence has been executed.

Optional.

If the engine provides this function, it will be called after the last event in the sequence has been executed.

pymmcore_plus.mda.MDAEngine

The default MDAengine that ships with pymmcore-plus.

This implements the PMDAEngine protocol, and uses a CMMCorePlus instance to control the hardware.

It may be subclassed to provide custom behavior, or to override specific methods. https://pymmcore-plus.github.io/pymmcore-plus/guides/custom_engine/

Attributes:

Name	Type	Description
mmcore	CMMCorePlus	The CMMCorePlus instance to use for hardware control.
use_hardware_sequencing	bool	Whether to use hardware sequencing if possible. If True, the engine will attempt to combine MDAEvents into a single SequencedEvent if core.canSequenceEvents() reports that the events can be sequenced. This can be set after instantiation. By default, this is True, however in various testing and demo scenarios, you may wish to set it to False in order to avoid unexpected behavior.
force_set_xy_position	bool	Whether to always set the XY position, even if the target position is the same as the last commanded position (this does not query the stage for the current position). By default, this is True.
restore_initial_state	bool | None	Whether to restore the initial hardware state after the MDA sequence completes. If True, the engine will capture the initial state (positions, config groups, exposure settings) before the sequence starts and restore it after completion. If None (the default), restore_initial_state will be set to True if FocusDirection is known (i.e. not Unknown).
timeout_base	float	Minimum per-image timeout in seconds for sequenced acquisitions. The actual timeout used is max(timeout_base, exposure_s * timeout_multiplier). The deadline is reset after each received frame, so long sequences will not time out as long as frames keep arriving. By default, this is 5.0.
timeout_multiplier	float	Multiplier applied to the camera exposure time (in seconds) to compute the per-image timeout for sequenced acquisitions. By default, this is 5.0.
timeout_first_frame	float | None	Separate timeout in seconds for waiting for the first frame of a sequenced acquisition. This is useful when the camera is waiting for an external trigger that may arrive much later than the exposure time would suggest. If None, the standard computed timeout is used for all frames including the first. By default, this is 20.0.
timeout_action	Literal['raise', 'warn']	What to do when a sequenced acquisition times out. If "raise" (the default), a TimeoutError is raised. If "warn", a warning is issued and None is yielded for any missing frames.

Source code in pymmcore_plus/mda/_engine.py

class MDAEngine(PMDAEngine):
    """The default MDAengine that ships with pymmcore-plus.

    This implements the [`PMDAEngine`][pymmcore_plus.mda.PMDAEngine] protocol, and
    uses a [`CMMCorePlus`][pymmcore_plus.CMMCorePlus] instance to control the hardware.

    It may be subclassed to provide custom behavior, or to override specific methods.
    <https://pymmcore-plus.github.io/pymmcore-plus/guides/custom_engine/>

    Attributes
    ----------
    mmcore: CMMCorePlus
        The `CMMCorePlus` instance to use for hardware control.
    use_hardware_sequencing : bool
        Whether to use hardware sequencing if possible. If `True`, the engine will
        attempt to combine MDAEvents into a single `SequencedEvent` if
        [`core.canSequenceEvents()`][pymmcore_plus.CMMCorePlus.canSequenceEvents]
        reports that the events can be sequenced. This can be set after instantiation.
        By default, this is `True`, however in various testing and demo scenarios, you
        may wish to set it to `False` in order to avoid unexpected behavior.
    force_set_xy_position : bool
        Whether to always set the XY position, even if the target position is the same
        as the last commanded position (this does *not* query the stage for the
        current position). By default, this is `True`.
    restore_initial_state : bool | None
        Whether to restore the initial hardware state after the MDA sequence completes.
        If `True`, the engine will capture the initial state (positions,
        config groups, exposure settings) before the sequence starts and restore it
        after completion.  If `None` (the default), `restore_initial_state` will
        be set to `True` if FocusDirection is known (i.e. not Unknown).
    timeout_base : float
        Minimum per-image timeout in seconds for sequenced acquisitions. The actual
        timeout used is `max(timeout_base, exposure_s * timeout_multiplier)`.
        The deadline is reset after each received frame, so long sequences will not
        time out as long as frames keep arriving. By default, this is `5.0`.
    timeout_multiplier : float
        Multiplier applied to the camera exposure time (in seconds) to compute the
        per-image timeout for sequenced acquisitions. By default, this is `5.0`.
    timeout_first_frame : float | None
        Separate timeout in seconds for waiting for the *first* frame of a
        sequenced acquisition. This is useful when the camera is waiting for an
        external trigger that may arrive much later than the exposure time would
        suggest. If `None`, the standard computed timeout is used
        for all frames including the first. By default, this is `20.0`.
    timeout_action : Literal["raise", "warn"]
        What to do when a sequenced acquisition times out. If `"raise"` (the
        default), a `TimeoutError` is raised. If `"warn"`, a warning is issued
        and `None` is yielded for any missing frames.
    """

    def __init__(
        self,
        mmc: CMMCorePlus,
        *,
        use_hardware_sequencing: bool = True,
        force_set_xy_position: bool = True,
        restore_initial_state: bool | None = None,
        timeout_base: float = 5.0,
        timeout_multiplier: float = 5.0,
        timeout_first_frame: float | None = 20.0,
        timeout_action: Literal["raise", "warn"] = "raise",
    ) -> None:
        self._mmcore_ref = weakref.ref(mmc)
        self.use_hardware_sequencing: bool = use_hardware_sequencing
        self.force_set_xy_position: bool = force_set_xy_position
        self.restore_initial_state: bool | None = restore_initial_state
        self.timeout_base: float = timeout_base
        self.timeout_multiplier: float = timeout_multiplier
        self.timeout_first_frame: float | None = timeout_first_frame
        self.timeout_action: Literal["raise", "warn"] = timeout_action

        # whether to include position metadata when fetching on-frame metadata
        # omitted by default when performing triggered acquisition because it's slow.
        self._include_frame_position_metadata: IncludePositionArg = "unsequenced-only"

        # stored initial state for restoration (if restore_initial_state is True)
        self._initial_state: StateDict = {}

        # used to check if the hardware autofocus is engaged when the sequence begins.
        # if it is, we will re-engage it after the autofocus action (if successful).
        self._af_was_engaged: bool = False
        # used to store the success of the last _execute_autofocus call
        self._af_succeeded: bool = False

        # used for one_shot autofocus to store the z correction for each position index.
        # map of {position_index: z_correction}
        self._z_correction: dict[int | None, float] = {}

        # This is used to determine whether we need to re-enable autoshutter after
        # the sequence is done (assuming a event.keep_shutter_open was requested)
        # Note: getAutoShutter() is True when no config is loaded at all
        self._autoshutter_was_set: bool = mmc.getAutoShutter()

        self._last_config: tuple[str, str] = ("", "")
        self._last_xy_pos: tuple[float | None, float | None] = (None, None)

        # -----
        # The following values are stored during setup_sequence simply to speed up
        # retrieval of metadata during each frame.
        # sequence of (device, property) of all properties used in any of the presets
        # in the channel group.
        self._config_device_props: dict[str, Sequence[tuple[str, str]]] = {}

    @property
    def include_frame_position_metadata(self) -> IncludePositionArg:
        return self._include_frame_position_metadata

    @include_frame_position_metadata.setter
    def include_frame_position_metadata(self, value: IncludePositionArg) -> None:
        if value not in (True, False, "unsequenced-only"):  # pragma: no cover
            raise ValueError(
                "include_frame_position_metadata must be True, False, or "
                "'unsequenced-only'"
            )
        self._include_frame_position_metadata = value

    @property
    def mmcore(self) -> CMMCorePlus:
        """The `CMMCorePlus` instance to use for hardware control."""
        if (mmc := self._mmcore_ref()) is None:  # pragma: no cover
            raise RuntimeError("The CMMCorePlus instance has been garbage collected.")
        return mmc

    def _frame_timeout(self, event: MDAEvent) -> float:
        """Compute per-frame timeout in seconds for a sequenced acquisition."""
        exposure_s = (event.exposure or 0.0) / 1000.0
        return max(
            self.timeout_base,
            exposure_s * self.timeout_multiplier,
        )

    def _initial_timeout(self, event: MDAEvent) -> float:
        """Compute timeout for the first frame of a sequenced acquisition."""
        if self.timeout_first_frame is not None:
            return self.timeout_first_frame
        return self._frame_timeout(event)

    def _handle_timeout(self, timeout: float) -> None:
        """Handle a sequenced acquisition timeout per the configured action."""
        self.mmcore.stopSequenceAcquisition()
        msg = (
            f"Acquisition timed out after {timeout:.1f}s. "
            "Consider increasing 'timeout_base' or "
            "'timeout_multiplier' on the MDAEngine, or setting "
            "'timeout_first_frame' if the camera is waiting for an "
            "external trigger."
        )
        if self.timeout_action == "raise":
            raise TimeoutError(msg)
        logger.warning(msg)

    # ===================== Protocol Implementation =====================

    def setup_sequence(self, sequence: MDASequence) -> SummaryMetaV1 | None:
        """Setup the hardware for the entire sequence."""
        # clear z_correction for new sequence
        self._z_correction.clear()

        if not (core := self._mmcore_ref()):  # pragma: no cover
            from pymmcore_plus.core import CMMCorePlus

            core = CMMCorePlus.instance()
            self._mmcore_ref = weakref.ref(core)

        # stop any "continuous sequence acquisition" (live mode) that may be running.
        if core.isSequenceRunning():
            core.stopSequenceAcquisition()

        # just in case a non-programmatic changes have been made in the meantime
        # https://github.com/pymmcore-plus/pymmcore-plus/issues/503
        core._last_config = ("", "")  # noqa: SLF001
        core._last_xy_position.clear()  # noqa: SLF001

        self._update_config_device_props()
        # get if the autofocus is engaged at the start of the sequence
        self._af_was_engaged = core.isContinuousFocusLocked()

        # capture initial state if restoration is enabled
        if self.restore_initial_state is None:
            if fd := core.getFocusDevice():
                self.restore_initial_state = (
                    core.getFocusDirection(fd) != FocusDirection.Unknown
                )
            else:
                self.restore_initial_state = False

        if self.restore_initial_state:
            self._initial_state = self._capture_state()

        # Apply the setup event (ROI, properties, etc.) before computing
        # grid FOV sizes and summary metadata so they reflect the setup state.
        if sequence.setup is not None:
            self.setup_single_event(sequence.setup)

        if px_size := core.getPixelSizeUm():
            self._update_grid_fov_sizes(px_size, sequence)

        self._autoshutter_was_set = core.getAutoShutter()
        return self.get_summary_metadata(mda_sequence=sequence)

    def get_summary_metadata(self, mda_sequence: MDASequence | None) -> SummaryMetaV1:
        return summary_metadata(self.mmcore, mda_sequence=mda_sequence)

    def _update_grid_fov_sizes(self, px_size: float, sequence: MDASequence) -> None:
        *_, x_size, y_size = self.mmcore.getROI()
        fov_width = x_size * px_size
        fov_height = y_size * px_size

        if sequence.grid_plan:
            sequence.grid_plan.fov_width = fov_width
            sequence.grid_plan.fov_height = fov_height

        # set fov to any stage positions sequences
        for p in sequence.stage_positions:
            if p.sequence and p.sequence.grid_plan:
                p.sequence.grid_plan.fov_height = fov_height
                p.sequence.grid_plan.fov_width = fov_width

    def setup_event(self, event: MDAEvent) -> None:
        """Set the system hardware (XY, Z, channel, exposure) as defined in the event.

        Parameters
        ----------
        event : MDAEvent
            The event to use for the Hardware config
        """
        if isinstance(event, SequencedEvent):
            self.setup_sequenced_event(event)
        else:
            self.setup_single_event(event)
        self.mmcore.waitForSystem()

    def exec_event(self, event: MDAEvent) -> Iterable[PImagePayload | None]:
        """Execute an individual event and return the image data."""
        action = getattr(event, "action", None)
        core = self.mmcore
        if isinstance(action, HardwareAutofocus):
            # skip if no autofocus device is found
            if not core.getAutoFocusDevice():
                logger.warning("No autofocus device found. Cannot execute autofocus.")
                return

            try:
                # execute hardware autofocus
                new_correction = self._execute_autofocus(action)
                self._af_succeeded = True
            except RuntimeError as e:
                logger.warning("Hardware autofocus failed. %s", e)
                self._af_succeeded = False
            else:
                # store correction for this position index
                p_idx = event.index.get("p", None)
                self._z_correction[p_idx] = new_correction + self._z_correction.get(
                    p_idx, 0.0
                )
            return

        # don't try to execute any other action types. Mostly, this is just
        # CustomAction, which is a user-defined action that the engine doesn't know how
        # to handle.  But may include other actions in the future, and this ensures
        # backwards compatibility.
        if not isinstance(action, (AcquireImage, type(None))):
            return

        # if the autofocus was engaged at the start of the sequence AND autofocus action
        # did not fail, re-engage it. NOTE: we need to do that AFTER the runner calls
        # `setup_event`, so we can't do it inside the exec_event autofocus action above.
        if self._af_was_engaged and self._af_succeeded:
            core.enableContinuousFocus(True)

        if isinstance(event, SequencedEvent):
            yield from self.exec_sequenced_event(event)
        else:
            yield from self.exec_single_event(event)

    def event_iterator(self, events: Iterable[MDAEvent]) -> Iterator[MDAEvent]:
        """Event iterator that merges events for hardware sequencing if possible.

        This wraps `for event in events: ...` inside `MDARunner.run()` and combines
        sequenceable events into an instance of `SequencedEvent` if
        `self.use_hardware_sequencing` is `True`.
        """
        if not self.use_hardware_sequencing:
            yield from events
            return

        yield from iter_sequenced_events(self.mmcore, events)

    # ===================== Regular Events =====================

    def setup_single_event(self, event: MDAEvent) -> None:
        """Setup hardware for a single (non-sequenced) event.

        This method is not part of the PMDAEngine protocol (it is called by
        `setup_event`, which *is* part of the protocol), but it is made public
        in case a user wants to subclass this engine and override this method.
        """
        if event.keep_shutter_open:
            ...

        self._set_event_xy_position(event)

        if event.z_pos is not None:
            self._set_event_z(event)
        if event.slm_image is not None:
            self._set_event_slm_image(event)

        self._set_event_channel(event)

        mmcore = self.mmcore
        if event.exposure is not None:
            try:
                mmcore.setExposure(event.exposure)
            except Exception as e:
                logger.warning("Failed to set exposure. %s", e)
        if event.properties is not None:
            self._set_event_properties(event.properties)
        if event.roi is not None:
            self._set_event_roi(event)
        if (
            # (if autoshutter wasn't set at the beginning of the sequence
            # then it never matters...)
            self._autoshutter_was_set
            # if we want to leave the shutter open after this event, and autoshutter
            # is currently enabled...
            and event.keep_shutter_open
            and mmcore.getAutoShutter()
        ):
            # we have to disable autoshutter and open the shutter
            mmcore.setAutoShutter(False)
            mmcore.setShutterOpen(True)

    def exec_single_event(self, event: MDAEvent) -> Iterator[PImagePayload | None]:
        """Execute a single (non-triggered) event and return the image data.

        This method is not part of the PMDAEngine protocol (it is called by
        `exec_event`, which *is* part of the protocol), but it is made public
        in case a user wants to subclass this engine and override this method.
        """
        if event.slm_image is not None:
            self._exec_event_slm_image(event.slm_image)

        mmcore = self.mmcore
        try:
            mmcore.snapImage()
            # taking event time after snapImage includes exposure time
            # not sure that's what we want, but it's currently consistent with the
            # timing of the sequenced event runner (where Elapsed_Time_ms is taken after
            # the image is acquired, not before the exposure starts)
            t0 = event.metadata.get("runner_t0") or time.perf_counter()
            event_time_ms = (time.perf_counter() - t0) * 1000
        except Exception as e:
            logger.warning("Failed to snap image. %s", e)
            return
        if not event.keep_shutter_open and self._autoshutter_was_set:
            mmcore.setShutterOpen(False)

        # most cameras will only have a single channel
        # but Multi-camera may have multiple, and we need to retrieve a buffer for each
        n_cam_channels = mmcore.getNumberOfCameraChannels()
        for cam in range(n_cam_channels):
            meta = self.get_frame_metadata(
                event,
                runner_time_ms=event_time_ms,
                camera_device=mmcore.getPhysicalCameraDevice(cam),
                include_position=self._include_frame_position_metadata is not False,
            )
            # add cam index when using multi-camera, matching sequenced path
            sub_event = event
            if n_cam_channels > 1:
                sub_event = event.model_copy(
                    update={"index": {**event.index, "cam": cam}}
                )
            # Note, the third element is actually a MutableMapping, but mypy doesn't
            # see TypedDict as a subclass of MutableMapping yet.
            # https://github.com/python/mypy/issues/4976
            yield ImagePayload(mmcore.getImage(cam), sub_event, meta)  # type: ignore[misc]

    def get_frame_metadata(
        self,
        event: MDAEvent,
        prop_values: tuple[PropertyValue, ...] | None = None,
        runner_time_ms: float = 0.0,
        include_position: bool = True,
        camera_device: str | None = None,
    ) -> FrameMetaV1:
        if prop_values is None and (ch := event.channel):
            prop_values = self._get_current_props(ch.group)
        else:
            prop_values = ()
        return frame_metadata(
            self.mmcore,
            cached=True,
            runner_time_ms=runner_time_ms,
            camera_device=camera_device,
            property_values=prop_values,
            mda_event=event,
            include_position=include_position,
        )

    def teardown_event(self, event: MDAEvent) -> None:
        """Teardown state of system (hardware, etc.) after `event`."""
        # autoshutter was set at the beginning of the sequence, and this event
        # doesn't want to leave the shutter open.  Re-enable autoshutter.
        core = self.mmcore
        if not event.keep_shutter_open and self._autoshutter_was_set:
            core.setAutoShutter(True)
        # FIXME: this may not be hitting as intended...
        # https://github.com/pymmcore-plus/pymmcore-plus/pull/353#issuecomment-2159176491
        if isinstance(event, SequencedEvent):
            if event.exposure_sequence:
                core.stopExposureSequence(core.getCameraDevice())
            if event.x_sequence:
                core.stopXYStageSequence(core.getXYStageDevice())
            if event.z_sequence:
                core.stopStageSequence(core.getFocusDevice())
            for dev, prop in event.property_sequences:
                core.stopPropertySequence(dev, prop)

    def teardown_sequence(self, sequence: MDASequence) -> None:
        """Perform any teardown required after the sequence has been executed."""
        # restore initial state if enabled and state was captured
        if self.restore_initial_state and self._initial_state:
            self._restore_initial_state()

    def _capture_state(self) -> StateDict:
        """Capture the current hardware state for later restoration."""
        state: StateDict = {}
        if (core := self._mmcore_ref()) is None:
            return state

        try:
            # capture XY position
            if core.getXYStageDevice():
                state["xy_position"] = core.getXYPosition()
        except Exception as e:
            logger.warning("Failed to capture XY position: %s", e)

        try:
            # capture Z position
            if core.getFocusDevice():
                state["z_position"] = core.getZPosition()
        except Exception as e:
            logger.warning("Failed to capture Z position: %s", e)

        try:
            state["exposure"] = core.getExposure()
        except Exception as e:
            logger.warning("Failed to capture exposure setting: %s", e)

        # capture config group states
        try:
            state_groups = state.setdefault("config_groups", {})
            for group in core.getAvailableConfigGroups():
                if current_config := core.getCurrentConfig(group):
                    state_groups[group] = current_config
        except Exception as e:
            logger.warning("Failed to get available config groups: %s", e)

        # capture autoshutter state
        try:
            state["autoshutter"] = core.getAutoShutter()
        except Exception as e:
            logger.warning("Failed to capture autoshutter state: %s", e)

        # capture ROI
        try:
            state["roi"] = tuple(core.getROI())  # type: ignore[typeddict-item]
        except Exception as e:
            logger.warning("Failed to capture ROI: %s", e)

        return state

    def _restore_initial_state(self) -> None:
        """Restore the hardware state that was captured before the sequence."""
        if not self._initial_state or (core := self._mmcore_ref()) is None:
            return

        # !!! We need to be careful about the order of Z and XY restoration:
        #
        # If FocusDirection is Unknown, we cannot safely restore Z *or* XY stage
        # positions: we simply refuse and warn.
        #
        # If focus_dir is TowardSample, and we are restoring a Z-position that is
        # *lower* than the current position or
        # if focus_dir is AwayFromSample, and we are restoring a Z-position that is
        # *higher* than the current position, then we need to move Z *before* moving XY,
        # otherwise we may crash the objective into the sample.
        # Otherwise, we should move XY first, then Z.
        target_z = self._initial_state.get("z_position")
        move_z_first = False
        focus_dir = FocusDirection.Unknown
        if target_z is not None and (focus_device := core.getFocusDevice()):
            focus_dir = core.getFocusDirection(focus_device)
            cur_z = core.getZPosition()
            # focus_dir TowardSample => increasing position brings obj. closer to sample
            if cur_z > target_z:
                if focus_dir == FocusDirection.TowardSample:
                    move_z_first = True
            elif focus_dir == FocusDirection.AwayFromSample:
                move_z_first = True

        if focus_dir == FocusDirection.Unknown:
            _warn_focus_dir(focus_device)
        else:

            def _move_z() -> None:
                if target_z is not None:
                    try:
                        if core.getFocusDevice():
                            core.setZPosition(target_z)
                    except Exception as e:
                        logger.warning("Failed to restore Z position: %s", e)

            if move_z_first:
                _move_z()

            # restore XY position
            if "xy_position" in self._initial_state:
                try:
                    if core.getXYStageDevice():
                        core.setXYPosition(*self._initial_state["xy_position"])
                except Exception as e:
                    logger.warning("Failed to restore XY position: %s", e)

            if not move_z_first:
                _move_z()

        # restore exposure
        if "exposure" in self._initial_state:
            try:
                core.setExposure(self._initial_state["exposure"])
            except Exception as e:
                logger.warning("Failed to restore exposure setting: %s", e)

        # restore config group states
        for key, value in self._initial_state.get("config_groups", {}).items():
            try:
                core.setConfig(key, value)
            except Exception as e:
                logger.warning(
                    "Failed to restore config group %s to %s: %s", key, value, e
                )

        # restore autoshutter state
        if "autoshutter" in self._initial_state:
            try:
                core.setAutoShutter(self._initial_state["autoshutter"])
            except Exception as e:
                logger.warning("Failed to restore autoshutter state: %s", e)

        # restore ROI
        if "roi" in self._initial_state:
            core.clearROI()
            core.setROI(*self._initial_state["roi"])

        core.waitForSystem()
        # clear the state after restoration
        self._initial_state = {}

    # ===================== Sequenced Events =====================

    def _load_sequenced_event(self, event: SequencedEvent) -> None:
        """Load a `SequencedEvent` into the core.

        `SequencedEvent` is a special pymmcore-plus specific subclass of
        `useq.MDAEvent`.
        """
        core = self.mmcore
        if event.exposure_sequence:
            cam_device = core.getCameraDevice()
            with suppress(RuntimeError):
                core.stopExposureSequence(cam_device)
            core.loadExposureSequence(cam_device, event.exposure_sequence)
        if event.x_sequence:  # y_sequence is implied and will be the same length
            stage = core.getXYStageDevice()
            with suppress(RuntimeError):
                core.stopXYStageSequence(stage)
            core.loadXYStageSequence(stage, event.x_sequence, event.y_sequence)
        if event.z_sequence:
            zstage = core.getFocusDevice()
            with suppress(RuntimeError):
                core.stopStageSequence(zstage)
            core.loadStageSequence(zstage, event.z_sequence)
        if event.slm_sequence:
            slm = core.getSLMDevice()
            with suppress(RuntimeError):
                core.stopSLMSequence(slm)
            core.loadSLMSequence(slm, event.slm_sequence)  # type: ignore[arg-type]
        if event.property_sequences:
            for (dev, prop), value_sequence in event.property_sequences.items():
                with suppress(RuntimeError):
                    core.stopPropertySequence(dev, prop)
                core.loadPropertySequence(dev, prop, value_sequence)

    def setup_sequenced_event(self, event: SequencedEvent) -> None:
        """Setup hardware for a sequenced (triggered) event.

        This method is not part of the PMDAEngine protocol (it is called by
        `setup_event`, which *is* part of the protocol), but it is made public
        in case a user wants to subclass this engine and override this method.
        """
        core = self.mmcore

        self._load_sequenced_event(event)

        # this is probably not necessary.  loadSequenceEvent will have already
        # set all the config properties individually/manually.  However, without
        # the call below, we won't be able to query `core.getCurrentConfig()`
        # not sure that's necessary; and this is here for tests to pass for now,
        # but this could be removed.
        self._set_event_channel(event)

        if event.properties:
            self._set_event_properties(event.properties)

        if event.slm_image:
            self._set_event_slm_image(event)

        if event.roi:
            self._set_event_roi(event)

        if core.isSequenceRunning():
            self._await_sequence_acquisition()

        # start sequences or set non-sequenced values
        if event.x_sequence:
            core.startXYStageSequence(core.getXYStageDevice())
        else:
            self._set_event_xy_position(event)

        if event.z_sequence:
            core.startStageSequence(core.getFocusDevice())
        elif event.z_pos is not None:
            self._set_event_z(event)

        if event.exposure_sequence:
            core.startExposureSequence(core.getCameraDevice())
        elif event.exposure is not None:
            core.setExposure(event.exposure)

        if event.property_sequences:
            for dev, prop in event.property_sequences:
                core.startPropertySequence(dev, prop)

    def _await_sequence_acquisition(
        self, timeout: float = 5.0, poll_interval: float = 0.2
    ) -> None:
        tot = 0.0
        core = self.mmcore
        core.stopSequenceAcquisition()
        while core.isSequenceRunning():
            time.sleep(poll_interval)
            tot += poll_interval
            if tot >= timeout:
                raise TimeoutError("Failed to stop running sequence")

    def post_sequence_started(self, event: SequencedEvent) -> None:
        """Perform any actions after startSequenceAcquisition has been called.

        This method is available to subclasses in case they need to perform any
        actions after a hardware-triggered sequence has been started (i.e. after
        core.startSequenceAcquisition has been called).

        The default implementation does nothing.
        """

    def exec_sequenced_event(self, event: SequencedEvent) -> EventPayloadGenerator:
        """Execute a sequenced (triggered) event and return the image data.

        This method is not part of the PMDAEngine protocol (it is called by
        `exec_event`, which *is* part of the protocol), but it is made public
        in case a user wants to subclass this engine and override this method.
        """
        n_events = len(event.events)
        if t0 := event.metadata.get("runner_t0"):
            t0_ms = (time.perf_counter() - t0) * 1000
        else:
            t0_ms = 0.0

        if event.slm_image is not None:
            self._exec_event_slm_image(event.slm_image)

        core = self.mmcore
        n_channels = core.getNumberOfCameraChannels()

        # Start sequence
        # Note that the overload of startSequenceAcquisition that takes a camera
        # label does NOT automatically initialize a circular buffer.  So if this call
        # is changed to accept the camera in the future, that should be kept in mind.
        core.startSequenceAcquisition(n_events, 0, True)
        self.post_sequence_started(event)

        if n_channels == 1:
            yield from self._exec_single_camera_sequence(event, t0_ms)
        else:
            yield from self._exec_multi_camera_sequence(event, t0_ms, n_channels)

    def _exec_single_camera_sequence(
        self, event: SequencedEvent, t0_ms: float
    ) -> EventPayloadGenerator:
        """Execute single-camera sequenced event (fast path, no coordination needed)."""
        core = self.mmcore
        count = 0
        pause_warned = False
        frame_timeout = self._frame_timeout(event)
        timeout = self._initial_timeout(event)
        deadline = time.monotonic() + timeout

        # Pop frames while sequence is running, then drain remaining buffer
        while time.monotonic() < deadline:
            if remaining := core.getRemainingImageCount():
                # Reset deadline for next frame if we have remaining images
                timeout = frame_timeout
                deadline = time.monotonic() + timeout
                img, mm_meta = core.popNextImageAndMD()
                signal = yield self._create_seqimg_payload_from_popped(
                    img,
                    mm_meta,
                    event=event.events[count],
                    channel=0,
                    event_t0=t0_ms,
                    remaining=remaining - 1,
                )
                count += 1
                if signal == "cancel":
                    core.stopSequenceAcquisition()
                    return
                if signal == "pause" and not pause_warned:
                    pause_warned = True
                    logger.warning(
                        "MDA: Pause has been requested, but sequenced acquisition "
                        "cannot be yet paused, only canceled."
                    )
            elif core.isSequenceRunning():
                # Still acquiring, buffer temporarily empty - wait
                time.sleep(0.001)
            else:
                # Done acquiring and buffer empty - exit
                break
        else:
            # Deadline exceeded
            self._handle_timeout(timeout)

        if core.isBufferOverflowed():  # pragma: no cover
            raise MemoryError("Buffer overflowed")

        # Yield None for each missing frame so the runner can tell the sink
        n_expected = len(event.events)
        if count != n_expected:
            logger.warning(
                "Unexpected number of images returned from sequence. "
                "Expected %s, got %s",
                n_expected,
                count,
            )
            for _ in range(n_expected - count):
                yield None

    def _exec_multi_camera_sequence(
        self,
        event: SequencedEvent,
        t0_ms: float,
        n_channels: int,
    ) -> EventPayloadGenerator:
        """Execute multi-camera sequenced event with frame coordination."""
        core = self.mmcore
        n_events = len(event.events)
        coordinator = _MultiCameraCoordinator(core, n_channels, n_events)
        pause_warned = False
        frame_timeout = self._frame_timeout(event)
        timeout = self._initial_timeout(event)
        deadline = time.monotonic() + timeout

        # Unified loop: pop while running, then drain remaining buffer
        while time.monotonic() < deadline:
            if core.getRemainingImageCount():
                # Reset deadline whenever we receive data
                timeout = frame_timeout
                deadline = time.monotonic() + timeout
                for payload in coordinator.pop_and_process(
                    event.events,
                    t0_ms,
                    self._create_seqimg_payload_from_popped,
                ):
                    signal = yield payload
                    if signal == "cancel":
                        core.stopSequenceAcquisition()
                        return
                    if signal == "pause" and not pause_warned:
                        pause_warned = True
                        logger.warning(
                            "MDA: Pause has been requested, but sequenced "
                            "acquisition cannot be yet paused, only canceled."
                        )
            elif core.isSequenceRunning():
                time.sleep(0.001)
            else:
                break
        else:
            # Deadline exceeded
            self._handle_timeout(timeout)

        if core.isBufferOverflowed():  # pragma: no cover
            raise MemoryError("Buffer overflowed")

        # Flush buffered frames and validate count
        for payload in coordinator.flush_remaining():
            signal = yield payload
            if signal == "cancel":
                return
        for _missing in range(coordinator.validate_count()):
            yield None

    def _create_seqimg_payload_from_popped(
        self,
        img: NDArray,
        mm_meta: Metadata,
        event: MDAEvent,
        channel: int,
        *,
        event_t0: float = 0.0,
        remaining: int = 0,
    ) -> PImagePayload:
        """Create ImagePayload from already-popped image and metadata.

        This is used by exec_sequenced_event to properly handle multi-camera
        acquisitions where images arrive asynchronously.
        """
        core = self.mmcore
        try:
            seq_time = float(mm_meta.get(Keyword.Elapsed_Time_ms))
        except Exception:
            seq_time = 0.0
        try:
            # note, when present in circular buffer meta, this key is called "Camera".
            # It's NOT actually Keyword.CoreCamera (but it's the same value)
            # it is hardcoded in various places in mmCoreAndDevices, see:
            # see: https://github.com/micro-manager/mmCoreAndDevices/pull/468
            camera_device = mm_meta.GetSingleTag("Camera").GetValue()
        except Exception:
            camera_device = core.getPhysicalCameraDevice(channel)

        # TODO: determine whether we want to try to populate changing property values
        # during the course of a triggered sequence
        meta = self.get_frame_metadata(
            event,
            prop_values=(),
            runner_time_ms=event_t0 + seq_time,
            camera_device=camera_device,
            include_position=self._include_frame_position_metadata is True,
        )
        meta["hardware_triggered"] = True
        meta["images_remaining_in_buffer"] = remaining
        meta["camera_metadata"] = dict(mm_meta)

        # https://github.com/python/mypy/issues/4976
        return ImagePayload(img, event, meta)  # type: ignore[return-value]

    # ===================== EXTRA =====================

    def _execute_autofocus(self, action: HardwareAutofocus) -> float:
        """Perform the hardware autofocus.

        Returns the change in ZPosition that occurred during the autofocus event.
        """
        core = self.mmcore
        # switch off autofocus device if it is on
        core.enableContinuousFocus(False)

        if action.autofocus_motor_offset is not None:
            # set the autofocus device offset
            # if name is given explicitly, use it, otherwise use setAutoFocusOffset
            # (see docs for setAutoFocusOffset for additional details)
            if name := getattr(action, "autofocus_device_name", None):
                core.setPosition(name, action.autofocus_motor_offset)
            else:
                core.setAutoFocusOffset(action.autofocus_motor_offset)
            core.waitForSystem()

        @retry(exceptions=RuntimeError, tries=action.max_retries, logger=logger.warning)
        def _perform_full_focus(previous_z: float) -> float:
            core.fullFocus()
            core.waitForSystem()
            return core.getZPosition() - previous_z

        return _perform_full_focus(core.getZPosition())

    def _set_event_xy_position(self, event: MDAEvent) -> None:
        event_x, event_y = event.x_pos, event.y_pos
        # If neither coordinate is provided, do nothing.
        if event_x is None and event_y is None:
            return

        core = self.mmcore
        # skip if no XY stage device is found
        if not core.getXYStageDevice():
            logger.warning("No XY stage device found. Cannot set XY position.")
            return

        # Retrieve the last commanded XY position.
        last_x, last_y = core._last_xy_position.get(None) or (None, None)  # noqa: SLF001
        if (
            not self.force_set_xy_position
            and (event_x is None or event_x == last_x)
            and (event_y is None or event_y == last_y)
        ):
            return

        if event_x is None or event_y is None:
            cur_x, cur_y = core.getXYPosition()
            event_x = cur_x if event_x is None else event_x
            event_y = cur_y if event_y is None else event_y

        try:
            core.setXYPosition(event_x, event_y)
        except Exception as e:
            logger.warning("Failed to set XY position. %s", e)

    def _set_event_channel(self, event: MDAEvent) -> None:
        if (ch := event.channel) is None:
            return

        # comparison with _last_config is a fast/rough check ... which may miss subtle
        # differences if device properties have been individually set in the meantime.
        # could also compare to the system state, with:
        # data = self.mmcore.getConfigData(ch.group, ch.config)
        # if self.mmcore.getSystemStateCache().isConfigurationIncluded(data):
        #     ...
        if (ch.group, ch.config) != self.mmcore._last_config:  # noqa: SLF001
            try:
                self.mmcore.setConfig(ch.group, ch.config)
            except Exception as e:
                logger.warning("Failed to set channel. %s", e)

    def _set_event_z(self, event: MDAEvent) -> None:
        # skip if no Z stage device is found
        if not self.mmcore.getFocusDevice():
            logger.warning("No Z stage device found. Cannot set Z position.")
            return

        p_idx = event.index.get("p", None)
        correction = self._z_correction.setdefault(p_idx, 0.0)
        self.mmcore.setZPosition(cast("float", event.z_pos) + correction)

    def _set_event_properties(self, properties: Sequence[useq.PropertyTuple]) -> None:
        """Set device properties, using setPosition for stage devices."""
        core = self.mmcore
        for dev, prop, value in properties:
            try:
                if prop == Keyword.Position:
                    dev_type = core.getDeviceType(dev)
                    if dev_type == DeviceType.XYStage:
                        if not isinstance(value, (list, tuple)):
                            logger.warning(
                                "Expected XY position to be a list or tuple, got %s. "
                                "Ignoring.",
                                type(value),
                            )
                            continue
                        x, y, *_ = value
                        core.setXYPosition(dev, float(x), float(y))
                    elif dev_type == DeviceType.Stage:
                        core.setPosition(dev, float(value))
                    else:
                        core.setProperty(dev, prop, value)
                else:
                    core.setProperty(dev, prop, value)
            except Exception as e:
                logger.warning(
                    "Failed to set property %s of device %s. %s", prop, dev, e
                )

    def _set_event_roi(self, event: MDAEvent) -> None:
        """Set the camera ROI for this event.

        This method is not part of the PMDAEngine protocol (it is called by
        `setup_single_event`), but it is made public in case a user wants to
        subclass this engine and override ROI behavior.
        """
        if (roi := event.roi) is None:
            return
        try:
            self.mmcore.clearROI()
            self.mmcore.setROI(roi.offset_x, roi.offset_y, roi.width, roi.height)
        except Exception as e:
            logger.warning("Failed to set ROI. %s", e)

    def _set_event_slm_image(self, event: MDAEvent) -> None:
        if not event.slm_image:
            return
        core = self.mmcore
        try:
            # Get the SLM device
            if not (
                slm_device := event.slm_image.device or core.getSLMDevice()
            ):  # pragma: no cover
                raise ValueError("No SLM device found or specified.")

            # cast to numpy array
            slm_array = np.asarray(event.slm_image)
            # if it's a single value, we can just set all pixels to that value
            if slm_array.ndim == 0:
                value = slm_array.item()
                if isinstance(value, bool):
                    dev_name = core.getDeviceName(slm_device)
                    on_value = _SLM_DEVICES_PIXEL_ON_VALUES.get(dev_name, 1)
                    value = on_value if value else 0
                core.setSLMPixelsTo(slm_device, int(value))
            elif slm_array.size == 3:
                # if it's a 3-valued array, we assume it's RGB
                r, g, b = slm_array.astype(int)
                core.setSLMPixelsTo(slm_device, r, g, b)
            elif slm_array.ndim in (2, 3):
                # if it's a 2D/3D array, we assume it's an image
                # where 3D is RGB with shape (h, w, 3)
                if slm_array.ndim == 3 and slm_array.shape[2] != 3:
                    raise ValueError(  # pragma: no cover
                        "SLM image must be 2D or 3D with 3 channels (RGB)."
                    )
                # convert boolean on/off values to pixel values
                if slm_array.dtype == bool:
                    dev_name = core.getDeviceName(slm_device)
                    on_value = _SLM_DEVICES_PIXEL_ON_VALUES.get(dev_name, 1)
                    slm_array = np.where(slm_array, on_value, 0).astype(np.uint8)
                core.setSLMImage(slm_device, slm_array)
            if event.slm_image.exposure:
                core.setSLMExposure(slm_device, event.slm_image.exposure)
        except Exception as e:
            logger.warning("Failed to set SLM Image: %s", e)

    def _exec_event_slm_image(self, img: useq.SLMImage) -> None:
        if slm_device := (img.device or self.mmcore.getSLMDevice()):
            try:
                self.mmcore.displaySLMImage(slm_device)
            except Exception as e:
                logger.warning("Failed to set SLM Image: %s", e)

    def _update_config_device_props(self) -> None:
        # store devices/props that make up each config group for faster lookup
        self._config_device_props.clear()
        core = self.mmcore
        for grp in core.getAvailableConfigGroups():
            for preset in core.getAvailableConfigs(grp):
                # ordered/unique list of (device, property) tuples for each group
                self._config_device_props[grp] = tuple(
                    {(i[0], i[1]): None for i in core.getConfigData(grp, preset)}
                )

    def _get_current_props(self, *groups: str) -> tuple[PropertyValue, ...]:
        """Faster version of core.getConfigGroupState(group).

        MMCore does some excess iteration that we want to avoid here. It calls
        GetAvailableConfigs and then calls getConfigData for *every* preset in the
        group, (not only the one being requested).  We go straight to cached data
        for the group we want.
        """
        return tuple(
            {
                "dev": dev,
                "prop": prop,
                "val": self.mmcore.getPropertyFromCache(dev, prop),
            }
            for group in groups
            if (dev_props := self._config_device_props.get(group))
            for dev, prop in dev_props
        )

force_set_xy_position: bool = force_set_xy_position instance-attribute

restore_initial_state: bool | None = restore_initial_state instance-attribute

timeout_action: Literal['raise', 'warn'] = timeout_action instance-attribute

timeout_base: float = timeout_base instance-attribute

timeout_first_frame: float | None = timeout_first_frame instance-attribute

timeout_multiplier: float = timeout_multiplier instance-attribute

use_hardware_sequencing: bool = use_hardware_sequencing instance-attribute

event_iterator(events: Iterable[MDAEvent]) -> Iterator[MDAEvent]

Event iterator that merges events for hardware sequencing if possible.

This wraps for event in events: ... inside MDARunner.run() and combines sequenceable events into an instance of SequencedEvent if self.use_hardware_sequencing is True.

Source code in pymmcore_plus/mda/_engine.py

def event_iterator(self, events: Iterable[MDAEvent]) -> Iterator[MDAEvent]:
    """Event iterator that merges events for hardware sequencing if possible.

    This wraps `for event in events: ...` inside `MDARunner.run()` and combines
    sequenceable events into an instance of `SequencedEvent` if
    `self.use_hardware_sequencing` is `True`.
    """
    if not self.use_hardware_sequencing:
        yield from events
        return

    yield from iter_sequenced_events(self.mmcore, events)

exec_event(event: MDAEvent) -> Iterable[PImagePayload | None]

Execute an individual event and return the image data.

Source code in pymmcore_plus/mda/_engine.py

def exec_event(self, event: MDAEvent) -> Iterable[PImagePayload | None]:
    """Execute an individual event and return the image data."""
    action = getattr(event, "action", None)
    core = self.mmcore
    if isinstance(action, HardwareAutofocus):
        # skip if no autofocus device is found
        if not core.getAutoFocusDevice():
            logger.warning("No autofocus device found. Cannot execute autofocus.")
            return

        try:
            # execute hardware autofocus
            new_correction = self._execute_autofocus(action)
            self._af_succeeded = True
        except RuntimeError as e:
            logger.warning("Hardware autofocus failed. %s", e)
            self._af_succeeded = False
        else:
            # store correction for this position index
            p_idx = event.index.get("p", None)
            self._z_correction[p_idx] = new_correction + self._z_correction.get(
                p_idx, 0.0
            )
        return

    # don't try to execute any other action types. Mostly, this is just
    # CustomAction, which is a user-defined action that the engine doesn't know how
    # to handle.  But may include other actions in the future, and this ensures
    # backwards compatibility.
    if not isinstance(action, (AcquireImage, type(None))):
        return

    # if the autofocus was engaged at the start of the sequence AND autofocus action
    # did not fail, re-engage it. NOTE: we need to do that AFTER the runner calls
    # `setup_event`, so we can't do it inside the exec_event autofocus action above.
    if self._af_was_engaged and self._af_succeeded:
        core.enableContinuousFocus(True)

    if isinstance(event, SequencedEvent):
        yield from self.exec_sequenced_event(event)
    else:
        yield from self.exec_single_event(event)

exec_sequenced_event(event: SequencedEvent) -> EventPayloadGenerator

Execute a sequenced (triggered) event and return the image data.

This method is not part of the PMDAEngine protocol (it is called by exec_event, which is part of the protocol), but it is made public in case a user wants to subclass this engine and override this method.

Source code in pymmcore_plus/mda/_engine.py

def exec_sequenced_event(self, event: SequencedEvent) -> EventPayloadGenerator:
    """Execute a sequenced (triggered) event and return the image data.

    This method is not part of the PMDAEngine protocol (it is called by
    `exec_event`, which *is* part of the protocol), but it is made public
    in case a user wants to subclass this engine and override this method.
    """
    n_events = len(event.events)
    if t0 := event.metadata.get("runner_t0"):
        t0_ms = (time.perf_counter() - t0) * 1000
    else:
        t0_ms = 0.0

    if event.slm_image is not None:
        self._exec_event_slm_image(event.slm_image)

    core = self.mmcore
    n_channels = core.getNumberOfCameraChannels()

    # Start sequence
    # Note that the overload of startSequenceAcquisition that takes a camera
    # label does NOT automatically initialize a circular buffer.  So if this call
    # is changed to accept the camera in the future, that should be kept in mind.
    core.startSequenceAcquisition(n_events, 0, True)
    self.post_sequence_started(event)

    if n_channels == 1:
        yield from self._exec_single_camera_sequence(event, t0_ms)
    else:
        yield from self._exec_multi_camera_sequence(event, t0_ms, n_channels)

exec_single_event(event: MDAEvent) -> Iterator[PImagePayload | None]

Execute a single (non-triggered) event and return the image data.

This method is not part of the PMDAEngine protocol (it is called by exec_event, which is part of the protocol), but it is made public in case a user wants to subclass this engine and override this method.

Source code in pymmcore_plus/mda/_engine.py

def exec_single_event(self, event: MDAEvent) -> Iterator[PImagePayload | None]:
    """Execute a single (non-triggered) event and return the image data.

    This method is not part of the PMDAEngine protocol (it is called by
    `exec_event`, which *is* part of the protocol), but it is made public
    in case a user wants to subclass this engine and override this method.
    """
    if event.slm_image is not None:
        self._exec_event_slm_image(event.slm_image)

    mmcore = self.mmcore
    try:
        mmcore.snapImage()
        # taking event time after snapImage includes exposure time
        # not sure that's what we want, but it's currently consistent with the
        # timing of the sequenced event runner (where Elapsed_Time_ms is taken after
        # the image is acquired, not before the exposure starts)
        t0 = event.metadata.get("runner_t0") or time.perf_counter()
        event_time_ms = (time.perf_counter() - t0) * 1000
    except Exception as e:
        logger.warning("Failed to snap image. %s", e)
        return
    if not event.keep_shutter_open and self._autoshutter_was_set:
        mmcore.setShutterOpen(False)

    # most cameras will only have a single channel
    # but Multi-camera may have multiple, and we need to retrieve a buffer for each
    n_cam_channels = mmcore.getNumberOfCameraChannels()
    for cam in range(n_cam_channels):
        meta = self.get_frame_metadata(
            event,
            runner_time_ms=event_time_ms,
            camera_device=mmcore.getPhysicalCameraDevice(cam),
            include_position=self._include_frame_position_metadata is not False,
        )
        # add cam index when using multi-camera, matching sequenced path
        sub_event = event
        if n_cam_channels > 1:
            sub_event = event.model_copy(
                update={"index": {**event.index, "cam": cam}}
            )
        # Note, the third element is actually a MutableMapping, but mypy doesn't
        # see TypedDict as a subclass of MutableMapping yet.
        # https://github.com/python/mypy/issues/4976
        yield ImagePayload(mmcore.getImage(cam), sub_event, meta)  # type: ignore[misc]

get_frame_metadata(event: MDAEvent, prop_values: tuple[PropertyValue, ...] | None = None, runner_time_ms: float = 0.0, include_position: bool = True, camera_device: str | None = None) -> FrameMetaV1

Source code in pymmcore_plus/mda/_engine.py

def get_frame_metadata(
    self,
    event: MDAEvent,
    prop_values: tuple[PropertyValue, ...] | None = None,
    runner_time_ms: float = 0.0,
    include_position: bool = True,
    camera_device: str | None = None,
) -> FrameMetaV1:
    if prop_values is None and (ch := event.channel):
        prop_values = self._get_current_props(ch.group)
    else:
        prop_values = ()
    return frame_metadata(
        self.mmcore,
        cached=True,
        runner_time_ms=runner_time_ms,
        camera_device=camera_device,
        property_values=prop_values,
        mda_event=event,
        include_position=include_position,
    )

get_summary_metadata(mda_sequence: MDASequence | None) -> SummaryMetaV1

Source code in pymmcore_plus/mda/_engine.py

def get_summary_metadata(self, mda_sequence: MDASequence | None) -> SummaryMetaV1:
    return summary_metadata(self.mmcore, mda_sequence=mda_sequence)

include_frame_position_metadata() -> IncludePositionArg property writable

Source code in pymmcore_plus/mda/_engine.py

@property
def include_frame_position_metadata(self) -> IncludePositionArg:
    return self._include_frame_position_metadata

mmcore() -> CMMCorePlus property

The CMMCorePlus instance to use for hardware control.

Source code in pymmcore_plus/mda/_engine.py

@property
def mmcore(self) -> CMMCorePlus:
    """The `CMMCorePlus` instance to use for hardware control."""
    if (mmc := self._mmcore_ref()) is None:  # pragma: no cover
        raise RuntimeError("The CMMCorePlus instance has been garbage collected.")
    return mmc

post_sequence_started(event: SequencedEvent) -> None

Perform any actions after startSequenceAcquisition has been called.

This method is available to subclasses in case they need to perform any actions after a hardware-triggered sequence has been started (i.e. after core.startSequenceAcquisition has been called).

The default implementation does nothing.

Source code in pymmcore_plus/mda/_engine.py

def post_sequence_started(self, event: SequencedEvent) -> None:
    """Perform any actions after startSequenceAcquisition has been called.

    This method is available to subclasses in case they need to perform any
    actions after a hardware-triggered sequence has been started (i.e. after
    core.startSequenceAcquisition has been called).

    The default implementation does nothing.
    """

setup_event(event: MDAEvent) -> None

Set the system hardware (XY, Z, channel, exposure) as defined in the event.

Parameters:

Name	Type	Description	Default
event	MDAEvent	The event to use for the Hardware config	required

Source code in pymmcore_plus/mda/_engine.py

def setup_event(self, event: MDAEvent) -> None:
    """Set the system hardware (XY, Z, channel, exposure) as defined in the event.

    Parameters
    ----------
    event : MDAEvent
        The event to use for the Hardware config
    """
    if isinstance(event, SequencedEvent):
        self.setup_sequenced_event(event)
    else:
        self.setup_single_event(event)
    self.mmcore.waitForSystem()

setup_sequence(sequence: MDASequence) -> SummaryMetaV1 | None

Setup the hardware for the entire sequence.

Source code in pymmcore_plus/mda/_engine.py

def setup_sequence(self, sequence: MDASequence) -> SummaryMetaV1 | None:
    """Setup the hardware for the entire sequence."""
    # clear z_correction for new sequence
    self._z_correction.clear()

    if not (core := self._mmcore_ref()):  # pragma: no cover
        from pymmcore_plus.core import CMMCorePlus

        core = CMMCorePlus.instance()
        self._mmcore_ref = weakref.ref(core)

    # stop any "continuous sequence acquisition" (live mode) that may be running.
    if core.isSequenceRunning():
        core.stopSequenceAcquisition()

    # just in case a non-programmatic changes have been made in the meantime
    # https://github.com/pymmcore-plus/pymmcore-plus/issues/503
    core._last_config = ("", "")  # noqa: SLF001
    core._last_xy_position.clear()  # noqa: SLF001

    self._update_config_device_props()
    # get if the autofocus is engaged at the start of the sequence
    self._af_was_engaged = core.isContinuousFocusLocked()

    # capture initial state if restoration is enabled
    if self.restore_initial_state is None:
        if fd := core.getFocusDevice():
            self.restore_initial_state = (
                core.getFocusDirection(fd) != FocusDirection.Unknown
            )
        else:
            self.restore_initial_state = False

    if self.restore_initial_state:
        self._initial_state = self._capture_state()

    # Apply the setup event (ROI, properties, etc.) before computing
    # grid FOV sizes and summary metadata so they reflect the setup state.
    if sequence.setup is not None:
        self.setup_single_event(sequence.setup)

    if px_size := core.getPixelSizeUm():
        self._update_grid_fov_sizes(px_size, sequence)

    self._autoshutter_was_set = core.getAutoShutter()
    return self.get_summary_metadata(mda_sequence=sequence)

setup_sequenced_event(event: SequencedEvent) -> None

Setup hardware for a sequenced (triggered) event.

This method is not part of the PMDAEngine protocol (it is called by setup_event, which is part of the protocol), but it is made public in case a user wants to subclass this engine and override this method.

Source code in pymmcore_plus/mda/_engine.py

def setup_sequenced_event(self, event: SequencedEvent) -> None:
    """Setup hardware for a sequenced (triggered) event.

    This method is not part of the PMDAEngine protocol (it is called by
    `setup_event`, which *is* part of the protocol), but it is made public
    in case a user wants to subclass this engine and override this method.
    """
    core = self.mmcore

    self._load_sequenced_event(event)

    # this is probably not necessary.  loadSequenceEvent will have already
    # set all the config properties individually/manually.  However, without
    # the call below, we won't be able to query `core.getCurrentConfig()`
    # not sure that's necessary; and this is here for tests to pass for now,
    # but this could be removed.
    self._set_event_channel(event)

    if event.properties:
        self._set_event_properties(event.properties)

    if event.slm_image:
        self._set_event_slm_image(event)

    if event.roi:
        self._set_event_roi(event)

    if core.isSequenceRunning():
        self._await_sequence_acquisition()

    # start sequences or set non-sequenced values
    if event.x_sequence:
        core.startXYStageSequence(core.getXYStageDevice())
    else:
        self._set_event_xy_position(event)

    if event.z_sequence:
        core.startStageSequence(core.getFocusDevice())
    elif event.z_pos is not None:
        self._set_event_z(event)

    if event.exposure_sequence:
        core.startExposureSequence(core.getCameraDevice())
    elif event.exposure is not None:
        core.setExposure(event.exposure)

    if event.property_sequences:
        for dev, prop in event.property_sequences:
            core.startPropertySequence(dev, prop)

setup_single_event(event: MDAEvent) -> None

Setup hardware for a single (non-sequenced) event.

This method is not part of the PMDAEngine protocol (it is called by setup_event, which is part of the protocol), but it is made public in case a user wants to subclass this engine and override this method.

Source code in pymmcore_plus/mda/_engine.py

def setup_single_event(self, event: MDAEvent) -> None:
    """Setup hardware for a single (non-sequenced) event.

    This method is not part of the PMDAEngine protocol (it is called by
    `setup_event`, which *is* part of the protocol), but it is made public
    in case a user wants to subclass this engine and override this method.
    """
    if event.keep_shutter_open:
        ...

    self._set_event_xy_position(event)

    if event.z_pos is not None:
        self._set_event_z(event)
    if event.slm_image is not None:
        self._set_event_slm_image(event)

    self._set_event_channel(event)

    mmcore = self.mmcore
    if event.exposure is not None:
        try:
            mmcore.setExposure(event.exposure)
        except Exception as e:
            logger.warning("Failed to set exposure. %s", e)
    if event.properties is not None:
        self._set_event_properties(event.properties)
    if event.roi is not None:
        self._set_event_roi(event)
    if (
        # (if autoshutter wasn't set at the beginning of the sequence
        # then it never matters...)
        self._autoshutter_was_set
        # if we want to leave the shutter open after this event, and autoshutter
        # is currently enabled...
        and event.keep_shutter_open
        and mmcore.getAutoShutter()
    ):
        # we have to disable autoshutter and open the shutter
        mmcore.setAutoShutter(False)
        mmcore.setShutterOpen(True)

teardown_event(event: MDAEvent) -> None

Teardown state of system (hardware, etc.) after event.

Source code in pymmcore_plus/mda/_engine.py

def teardown_event(self, event: MDAEvent) -> None:
    """Teardown state of system (hardware, etc.) after `event`."""
    # autoshutter was set at the beginning of the sequence, and this event
    # doesn't want to leave the shutter open.  Re-enable autoshutter.
    core = self.mmcore
    if not event.keep_shutter_open and self._autoshutter_was_set:
        core.setAutoShutter(True)
    # FIXME: this may not be hitting as intended...
    # https://github.com/pymmcore-plus/pymmcore-plus/pull/353#issuecomment-2159176491
    if isinstance(event, SequencedEvent):
        if event.exposure_sequence:
            core.stopExposureSequence(core.getCameraDevice())
        if event.x_sequence:
            core.stopXYStageSequence(core.getXYStageDevice())
        if event.z_sequence:
            core.stopStageSequence(core.getFocusDevice())
        for dev, prop in event.property_sequences:
            core.stopPropertySequence(dev, prop)

teardown_sequence(sequence: MDASequence) -> None

Perform any teardown required after the sequence has been executed.

Source code in pymmcore_plus/mda/_engine.py

def teardown_sequence(self, sequence: MDASequence) -> None:
    """Perform any teardown required after the sequence has been executed."""
    # restore initial state if enabled and state was captured
    if self.restore_initial_state and self._initial_state:
        self._restore_initial_state()