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).

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).
    """

    def __init__(self) -> None:
        self._engine: PMDAEngine | None = None
        self._signals = _get_auto_MDA_callback_class()()
        self._running = False
        self._paused = False
        self._paused_time: float = 0
        self._pause_interval: float = 0.1  # sec to wait between checking pause state

        self._canceled = False
        self._sequence: MDASequence | None = None
        self._reset_timer()

    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

    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._running

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

        Use `toggle_pause` to change the paused state.

        Returns
        -------
        bool
            Whether the current acquisition is paused.
        """
        return self._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.
        """
        self._canceled = True
        self._paused_time = 0

    def toggle_pause(self) -> None:
        """Toggle the paused state of the current acquisition.

        To get whether the acquisition is currently paused use the
        [`is_paused`][pymmcore_plus.mda.MDARunner.is_paused] method. This method is a
        no-op if no acquisition is currently underway.
        """
        if self.is_running():
            self._paused = not self._paused
            self._signals.sequencePauseToggled.emit(self._paused)

    def run(
        self,
        events: Iterable[MDAEvent],
        *,
        output: SingleOutput | Sequence[SingleOutput] | None = None,
    ) -> 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.

        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:

            - A string or Path to a directory to save images to. A handler will be
                created automatically based on the extension of the path.
                - `.zarr` files will be handled by `OMEZarrWriter`
                - `.ome.tiff` files will be handled by `OMETiffWriter`
                - A directory with no extension will be handled by `ImageSequenceWriter`
            - A handler object that implements the `DataHandler` protocol, currently
                meaning it has a `frameReady` method.  See `mda_listeners_connected`
                for more details.
        """
        error = None
        sequence = events if isinstance(events, MDASequence) else GeneratorMDASequence()
        with self._outputs_connected(output):
            # 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 exceptions_logged():
                self._finish_run(sequence)
        if error is not None:
            raise error

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

    def _outputs_connected(
        self, output: SingleOutput | Sequence[SingleOutput] | None
    ) -> ContextManager:
        """Context in which output handlers are connected to the frameReady signal."""
        if output is None:
            return nullcontext()

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

        # convert all items to handler objects
        handlers: list[Any] = []
        for item in output:
            if isinstance(item, (str, Path)):
                handlers.append(self._handler_for_path(item))
            else:
                # TODO: better check for valid handler protocol
                # quick hack for now.
                if not hasattr(item, "frameReady"):
                    raise TypeError(
                        "Output handlers must have a frameReady method. "
                        f"Got {item} with type {type(item)}."
                    )
                handlers.append(item)

        return mda_listeners_connected(*handlers, mda_events=self._signals)

    def _handler_for_path(self, path: str | Path) -> object:
        """Convert a string or Path into a handler object.

        This method picks from the built-in handlers based on the extension of the path.
        """
        path = str(Path(path).expanduser().resolve())
        if path.endswith(".zarr"):
            from pymmcore_plus.mda.handlers import OMEZarrWriter

            return OMEZarrWriter(path)

        if path.endswith((".tiff", ".tif")):
            from pymmcore_plus.mda.handlers import OMETiffWriter

            return OMETiffWriter(path)

        # FIXME: ugly hack for the moment to represent a non-existent directory
        # there are many features that ImageSequenceWriter supports, and it's unclear
        # how to infer them all from a single string.
        if not (Path(path).suffix or Path(path).exists()):
            from pymmcore_plus.mda.handlers import ImageSequenceWriter

            return ImageSequenceWriter(path)

        raise ValueError(f"Could not infer a writer handler for path: '{path}'")

    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)
        event_iterator = getattr(engine, "event_iterator", iter)
        _events: Iterator[MDAEvent] = event_iterator(events)
        self._reset_timer()

        for event in _events:
            # If cancelled break out of the loop
            if self._wait_until_event(event) or not self._running:
                break

            self._signals.eventStarted.emit(event)
            logger.info("%s", event)
            engine.setup_event(event)

            try:
                elapsed_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
                # we pop it off after the event is executed.
                event.metadata["runner_t0"] = self._t0
                output = engine.exec_event(event) or ()  # in case output is None
                for payload in output:
                    img, event, meta = payload
                    event.metadata.pop("runner_t0", None)
                    if "runner_time_ms" not in meta:
                        meta["runner_time_ms"] = elapsed_ms
                    with exceptions_logged():
                        self._signals.frameReady.emit(img, event, meta)
            finally:
                teardown_event(event)

    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.")

        self._running = True
        self._paused = False
        self._paused_time = 0.0
        self._sequence = sequence

        meta = self._engine.setup_sequence(sequence)
        self._signals.sequenceStarted.emit(sequence, meta or {})
        logger.info("MDA Started: %s", sequence)
        return self._engine

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

    def _check_canceled(self) -> bool:
        """Return True if the cancel method has been called and emit relevant signals.

        If cancelled, this relies on the `self._sequence` being the current sequence
        in order to emit a `sequenceCanceled` signal.

        Returns
        -------
        bool
            Whether the MDA has been canceled.
        """
        if self._canceled:
            logger.warning("MDA Canceled: %s", self._sequence)
            self._signals.sequenceCanceled.emit(self._sequence)
            self._canceled = False
            return True
        return False

    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.
        """
        if not self.is_running():
            return False  # pragma: no cover
        if self._check_canceled():
            return True
        while self.is_paused() and not self._canceled:
            self._paused_time += self._pause_interval  # fixme: be more precise
            time.sleep(self._pause_interval)

            if self._check_canceled():
                return True

        # FIXME: this is actually the only place where the runner assumes our event is
        # an MDAevent.  For everything else, the engine is technically the only thing
        # that cares about the event time.
        # So this whole method could potentially be moved to the engine.
        if event.min_start_time:
            go_at = event.min_start_time + self._paused_time
            # We need to enter a loop here checking paused and canceled.
            # otherwise you'll potentially wait a long time to cancel
            remaining_wait_time = go_at - self.seconds_elapsed()
            while remaining_wait_time > 0:
                self._signals.awaitingEvent.emit(event, remaining_wait_time)
                while self._paused and not self._canceled:
                    self._paused_time += self._pause_interval  # fixme: be more precise
                    remaining_wait_time += self._pause_interval
                    time.sleep(self._pause_interval)

                if self._canceled:
                    break
                time.sleep(min(remaining_wait_time, 0.5))
                remaining_wait_time = go_at - self.seconds_elapsed()

        # check canceled again in case it was canceled
        # during the waiting loop
        return self._check_canceled()

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

        Parameters
        ----------
        sequence : MDASequence
            The sequence that was finished.
        """
        self._running = False
        self._canceled = False

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

        logger.info("MDA Finished: %s", sequence)
        self._signals.sequenceFinished.emit(sequence)

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.
    """
    self._canceled = True
    self._paused_time = 0

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

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

is_paused() -> bool

Return True if the acquisition is currently paused.

Use toggle_pause 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 `toggle_pause` to change the paused state.

    Returns
    -------
    bool
        Whether the current acquisition is paused.
    """
    return self._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._running

run(events: Iterable[MDAEvent], *, output: SingleOutput | Sequence[SingleOutput] | None = None) -> 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.

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 to a directory to save images to. A handler will be created automatically based on the extension of the path. .zarr files will be handled by OMEZarrWriter .ome.tiff files will be handled by OMETiffWriter A directory with no extension will be handled by ImageSequenceWriter A handler object that implements the DataHandler protocol, currently meaning it has a frameReady method. See mda_listeners_connected for more details.	None

Source code in pymmcore_plus/mda/_runner.py

def run(
    self,
    events: Iterable[MDAEvent],
    *,
    output: SingleOutput | Sequence[SingleOutput] | None = None,
) -> 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.

    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:

        - A string or Path to a directory to save images to. A handler will be
            created automatically based on the extension of the path.
            - `.zarr` files will be handled by `OMEZarrWriter`
            - `.ome.tiff` files will be handled by `OMETiffWriter`
            - A directory with no extension will be handled by `ImageSequenceWriter`
        - A handler object that implements the `DataHandler` protocol, currently
            meaning it has a `frameReady` method.  See `mda_listeners_connected`
            for more details.
    """
    error = None
    sequence = events if isinstance(events, MDASequence) else GeneratorMDASequence()
    with self._outputs_connected(output):
        # 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 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._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

toggle_pause() -> None

Toggle the paused state of the current acquisition.

To get whether the acquisition is currently paused use the is_paused method. This method is a no-op if no acquisition is currently underway.

Source code in pymmcore_plus/mda/_runner.py

def toggle_pause(self) -> None:
    """Toggle the paused state of the current acquisition.

    To get whether the acquisition is currently paused use the
    [`is_paused`][pymmcore_plus.mda.MDARunner.is_paused] method. This method is a
    no-op if no acquisition is currently underway.
    """
    if self.is_running():
        self._paused = not self._paused
        self._signals.sequencePauseToggled.emit(self._paused)

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] 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.

The protocol for the returned object is still under development. However, if the returned object has an image attribute, then the MDARunner will emit a frameReady signal

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.)

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 False, in order to avoid unexpected behavior, particularly in testing and demo scenarios. But in many "real world" scenarios, this can be set to True to improve performance.

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 `False`, in order to avoid unexpected behavior, particularly
        in testing and demo scenarios.  But in many "real world" scenarios, this can be
        set to `True` to improve performance.
    """

    def __init__(self, mmc: CMMCorePlus, use_hardware_sequencing: bool = False) -> None:
        self._mmc = mmc
        self.use_hardware_sequencing = use_hardware_sequencing

        # 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 = self._mmc.getAutoShutter()

        # -----
        # 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 mmcore(self) -> CMMCorePlus:
        """The `CMMCorePlus` instance to use for hardware control."""
        return self._mmc

    # ===================== 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 self._mmc:  # pragma: no cover
            from pymmcore_plus.core import CMMCorePlus

            self._mmc = CMMCorePlus.instance()

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

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

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

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

    def _update_grid_fov_sizes(self, px_size: float, sequence: MDASequence) -> None:
        *_, x_size, y_size = self._mmc.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._mmc.waitForSystem()

    def exec_event(self, event: MDAEvent) -> Iterable[PImagePayload]:
        """Execute an individual event and return the image data."""
        action = getattr(event, "action", None)
        if isinstance(action, HardwareAutofocus):
            # skip if no autofocus device is found
            if not self._mmc.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

        # 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:
            self._mmc.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

        seq: list[MDAEvent] = []
        for event in events:
            # if the sequence is empty or the current event can be sequenced with the
            # previous event, add it to the sequence
            if not seq or self._mmc.canSequenceEvents(seq[-1], event, len(seq)):
                seq.append(event)
            else:
                # otherwise, yield a SequencedEvent if the sequence has accumulated
                # more than one event, otherwise yield the single event
                yield seq[0] if len(seq) == 1 else SequencedEvent.create(seq)
                # add this current event and start a new sequence
                seq = [event]
        # yield any remaining events
        if seq:
            yield seq[0] if len(seq) == 1 else SequencedEvent.create(seq)

    # ===================== 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:
            ...

        if event.x_pos is not None or event.y_pos is not None:
            self._set_event_position(event)
        if event.z_pos is not None:
            self._set_event_z(event)

        if event.channel is not None:
            try:
                # possible speedup by setting manually.
                self._mmc.setConfig(event.channel.group, event.channel.config)
            except Exception as e:
                logger.warning("Failed to set channel. %s", e)
        if event.exposure is not None:
            try:
                self._mmc.setExposure(event.exposure)
            except Exception as e:
                logger.warning("Failed to set exposure. %s", e)

        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 self._mmc.getAutoShutter()
        ):
            # we have to disable autoshutter and open the shutter
            self._mmc.setAutoShutter(False)
            self._mmc.setShutterOpen(True)

    def exec_single_event(self, event: MDAEvent) -> Iterator[PImagePayload]:
        """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.
        """
        try:
            self._mmc.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:
            self._mmc.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
        for cam in range(self._mmc.getNumberOfCameraChannels()):
            meta = self.get_frame_metadata(
                event,
                runner_time_ms=event_time_ms,
                camera_device=self._mmc.getPhysicalCameraDevice(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(self._mmc.getImage(cam), event, meta)  # type: ignore[misc]

    def get_frame_metadata(
        self,
        event: MDAEvent,
        prop_values: tuple[PropertyValue, ...] | None = None,
        runner_time_ms: float = 0.0,
        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._mmc,
            cached=True,
            runner_time_ms=runner_time_ms,
            camera_device=camera_device,
            property_values=prop_values,
            mda_event=event,
        )

    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._mmc
        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(self._mmc.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):
                core.stopPropertySequence(dev, prop)

    def teardown_sequence(self, sequence: MDASequence) -> None:
        """Perform any teardown required after the sequence has been executed."""
        pass

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

    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._mmc
        cam_device = self._mmc.getCameraDevice()

        if event.exposure_sequence:
            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 prop_seqs := event.property_sequences(core):
            for (dev, prop), value_sequence in prop_seqs.items():
                with suppress(RuntimeError):
                    core.stopPropertySequence(dev, prop)
                core.loadPropertySequence(dev, prop, value_sequence)

        # TODO: SLM

        # preparing a Sequence while another is running is dangerous.
        if core.isSequenceRunning():
            self._await_sequence_acquisition()
        core.prepareSequenceAcquisition(cam_device)

        # start sequences or set non-sequenced values
        if event.x_sequence:
            core.startXYStageSequence(stage)
        elif event.x_pos is not None or event.y_pos is not None:
            self._set_event_position(event)

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

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

        if prop_seqs:
            for dev, prop in prop_seqs:
                core.startPropertySequence(dev, prop)
        elif event.channel is not None:
            core.setConfig(event.channel.group, event.channel.config)

    def _await_sequence_acquisition(
        self, timeout: float = 5.0, poll_interval: float = 0.2
    ) -> None:
        tot = 0.0
        self._mmc.stopSequenceAcquisition()
        while self._mmc.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) -> Iterable[PImagePayload]:
        """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)

        t0 = event.metadata.get("runner_t0") or time.perf_counter()
        event_t0_ms = (time.perf_counter() - t0) * 1000

        # 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.
        self._mmc.startSequenceAcquisition(
            n_events,
            0,  # intervalMS  # TODO: add support for this
            True,  # stopOnOverflow
        )
        self.post_sequence_started(event)

        n_channels = self._mmc.getNumberOfCameraChannels()
        count = 0
        iter_events = product(event.events, range(n_channels))
        # block until the sequence is done, popping images in the meantime
        while self._mmc.isSequenceRunning():
            if remaining := self._mmc.getRemainingImageCount():
                yield self._next_seqimg_payload(
                    *next(iter_events), remaining=remaining - 1, event_t0=event_t0_ms
                )
                count += 1
            else:
                time.sleep(0.001)

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

        while remaining := self._mmc.getRemainingImageCount():
            yield self._next_seqimg_payload(
                *next(iter_events), remaining=remaining - 1, event_t0=event_t0_ms
            )
            count += 1

        # necessary?
        expected_images = n_events * n_channels
        if count != expected_images:
            logger.warning(
                "Unexpected number of images returned from sequence. "
                "Expected %s, got %s",
                expected_images,
                count,
            )

    def _next_seqimg_payload(
        self,
        event: MDAEvent,
        channel: int = 0,
        *,
        event_t0: float = 0.0,
        remaining: int = 0,
    ) -> PImagePayload:
        """Grab next image from the circular buffer and return it as an ImagePayload."""
        _slice = 0  # ?
        img, mm_meta = self._mmc.popNextImageAndMD(channel, _slice)
        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 = self._mmc.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,
        )
        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.
        """
        # switch off autofocus device if it is on
        self._mmc.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):
                self._mmc.setPosition(name, action.autofocus_motor_offset)
            else:
                self._mmc.setAutoFocusOffset(action.autofocus_motor_offset)
            self._mmc.waitForSystem()

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

        return _perform_full_focus(self._mmc.getZPosition())

    def _set_event_position(self, event: MDAEvent) -> None:
        # skip if no XY stage device is found
        if not self._mmc.getXYStageDevice():
            logger.warning("No XY stage device found. Cannot set XY position.")
            return

        x = event.x_pos if event.x_pos is not None else self._mmc.getXPosition()
        y = event.y_pos if event.y_pos is not None else self._mmc.getYPosition()
        self._mmc.setXYPosition(x, y)

    def _set_event_z(self, event: MDAEvent) -> None:
        # skip if no Z stage device is found
        if not self._mmc.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._mmc.setZPosition(cast("float", event.z_pos) + correction)

    def _update_config_device_props(self) -> None:
        # store devices/props that make up each config group for faster lookup
        self._config_device_props.clear()
        for grp in self._mmc.getAvailableConfigGroups():
            for preset in self._mmc.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 self._mmc.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._mmc.getPropertyFromCache(dev, prop),
            }
            for group in groups
            if (dev_props := self._config_device_props.get(group))
            for dev, prop in dev_props
        )

use_hardware_sequencing = 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

    seq: list[MDAEvent] = []
    for event in events:
        # if the sequence is empty or the current event can be sequenced with the
        # previous event, add it to the sequence
        if not seq or self._mmc.canSequenceEvents(seq[-1], event, len(seq)):
            seq.append(event)
        else:
            # otherwise, yield a SequencedEvent if the sequence has accumulated
            # more than one event, otherwise yield the single event
            yield seq[0] if len(seq) == 1 else SequencedEvent.create(seq)
            # add this current event and start a new sequence
            seq = [event]
    # yield any remaining events
    if seq:
        yield seq[0] if len(seq) == 1 else SequencedEvent.create(seq)

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

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]:
    """Execute an individual event and return the image data."""
    action = getattr(event, "action", None)
    if isinstance(action, HardwareAutofocus):
        # skip if no autofocus device is found
        if not self._mmc.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

    # 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:
        self._mmc.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) -> Iterable[PImagePayload]

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) -> Iterable[PImagePayload]:
    """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)

    t0 = event.metadata.get("runner_t0") or time.perf_counter()
    event_t0_ms = (time.perf_counter() - t0) * 1000

    # 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.
    self._mmc.startSequenceAcquisition(
        n_events,
        0,  # intervalMS  # TODO: add support for this
        True,  # stopOnOverflow
    )
    self.post_sequence_started(event)

    n_channels = self._mmc.getNumberOfCameraChannels()
    count = 0
    iter_events = product(event.events, range(n_channels))
    # block until the sequence is done, popping images in the meantime
    while self._mmc.isSequenceRunning():
        if remaining := self._mmc.getRemainingImageCount():
            yield self._next_seqimg_payload(
                *next(iter_events), remaining=remaining - 1, event_t0=event_t0_ms
            )
            count += 1
        else:
            time.sleep(0.001)

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

    while remaining := self._mmc.getRemainingImageCount():
        yield self._next_seqimg_payload(
            *next(iter_events), remaining=remaining - 1, event_t0=event_t0_ms
        )
        count += 1

    # necessary?
    expected_images = n_events * n_channels
    if count != expected_images:
        logger.warning(
            "Unexpected number of images returned from sequence. "
            "Expected %s, got %s",
            expected_images,
            count,
        )

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

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]:
    """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.
    """
    try:
        self._mmc.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:
        self._mmc.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
    for cam in range(self._mmc.getNumberOfCameraChannels()):
        meta = self.get_frame_metadata(
            event,
            runner_time_ms=event_time_ms,
            camera_device=self._mmc.getPhysicalCameraDevice(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(self._mmc.getImage(cam), event, meta)  # type: ignore[misc]

get_frame_metadata(event: MDAEvent, prop_values: tuple[PropertyValue, ...] | None = None, runner_time_ms: float = 0.0, 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,
    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._mmc,
        cached=True,
        runner_time_ms=runner_time_ms,
        camera_device=camera_device,
        property_values=prop_values,
        mda_event=event,
    )

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._mmc, mda_sequence=mda_sequence)

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."""
    return self._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._mmc.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 self._mmc:  # pragma: no cover
        from pymmcore_plus.core import CMMCorePlus

        self._mmc = CMMCorePlus.instance()

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

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

    self._autoshutter_was_set = self._mmc.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._mmc
    cam_device = self._mmc.getCameraDevice()

    if event.exposure_sequence:
        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 prop_seqs := event.property_sequences(core):
        for (dev, prop), value_sequence in prop_seqs.items():
            with suppress(RuntimeError):
                core.stopPropertySequence(dev, prop)
            core.loadPropertySequence(dev, prop, value_sequence)

    # TODO: SLM

    # preparing a Sequence while another is running is dangerous.
    if core.isSequenceRunning():
        self._await_sequence_acquisition()
    core.prepareSequenceAcquisition(cam_device)

    # start sequences or set non-sequenced values
    if event.x_sequence:
        core.startXYStageSequence(stage)
    elif event.x_pos is not None or event.y_pos is not None:
        self._set_event_position(event)

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

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

    if prop_seqs:
        for dev, prop in prop_seqs:
            core.startPropertySequence(dev, prop)
    elif event.channel is not None:
        core.setConfig(event.channel.group, event.channel.config)

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:
        ...

    if event.x_pos is not None or event.y_pos is not None:
        self._set_event_position(event)
    if event.z_pos is not None:
        self._set_event_z(event)

    if event.channel is not None:
        try:
            # possible speedup by setting manually.
            self._mmc.setConfig(event.channel.group, event.channel.config)
        except Exception as e:
            logger.warning("Failed to set channel. %s", e)
    if event.exposure is not None:
        try:
            self._mmc.setExposure(event.exposure)
        except Exception as e:
            logger.warning("Failed to set exposure. %s", e)

    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 self._mmc.getAutoShutter()
    ):
        # we have to disable autoshutter and open the shutter
        self._mmc.setAutoShutter(False)
        self._mmc.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._mmc
    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(self._mmc.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):
            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."""
    pass