feat: add live MDA status line

src/pymmcore_gui/actions/widget_actions.py

@@ -2,24 +2,24 @@
 
 from __future__ import annotations
 
-from collections.abc import Callable
-from typing import TYPE_CHECKING, Annotated, TypeVar, cast
+from collections.abc import Callable, Mapping
+from typing import TYPE_CHECKING, Annotated, Any, TypeVar, cast
 
 import pymmcore_widgets as pmmw
 from pymmcore_plus import CMMCorePlus
+from useq import AcquireImage, HardwareAutofocus, MDAEvent, MDASequence
 
 from pymmcore_gui._qt.QtAds import CDockWidget, DockWidgetArea, SideBarLocation
-from pymmcore_gui._qt.QtCore import Qt
+from pymmcore_gui._qt.QtCore import QObject, Qt, Signal
 from pymmcore_gui._qt.QtGui import QAction
-from pymmcore_gui._qt.QtWidgets import QDialog, QVBoxLayout, QWidget
+from pymmcore_gui._qt.QtWidgets import QDialog, QLabel, QVBoxLayout, QWidget
 
 from ._action_info import ActionKey, WidgetActionInfo, _ensure_isinstance
 
 if TYPE_CHECKING:
     from pathlib import Path
 
     from pymmcore_gui._main_window import MicroManagerGUI
-    from pymmcore_gui._qt.QtCore import QObject
     from pymmcore_gui.widgets._exception_log import ExceptionLog
     from pymmcore_gui.widgets._mm_console import MMConsole
     from pymmcore_gui.widgets._stage_control import StagesControlWidget
@@ -29,6 +29,10 @@
 CT = TypeVar("CT", bound=Callable[[QWidget], QWidget])
 
 
+class _MDAStatusEmitter(QObject):
+    statusRequested = Signal(str)
+
+
 class WidgetAction(ActionKey):
     """Widget Actions toggle/create singleton widgets."""
 
@@ -110,6 +114,37 @@ def __init__(
         ) -> None:
             super().__init__(parent=parent, mmcore=mmcore)
             self._hide_tiff_sequence()
+            self._status_emitter = _MDAStatusEmitter(self)
+            self._active_sequence: MDASequence | None = None
+            self._frame_total = 0
+            self._frame_index = 0
+            self._last_event: MDAEvent | None = None
+            self._is_paused = False
+            self._was_canceled = False
+
+            self._status_label = QLabel("Idle", self)
+            self._status_label.setObjectName("mdaStatusLabel")
+            self._status_label.setWordWrap(True)
+            self._status_label.setTextInteractionFlags(
+                Qt.TextInteractionFlag.TextSelectableByMouse
+            )
+            self.statusRequested.connect(self._status_label.setText)
+
+            layout = cast("QVBoxLayout", self.layout())
+            layout.addWidget(self._status_label)
+
+            events = self._mmc.mda.events
+            events.sequenceStarted.connect(self._on_sequence_started)
+            events.eventStarted.connect(self._on_event_started)
+            events.frameReady.connect(self._on_frame_ready)
+            events.awaitingEvent.connect(self._on_awaiting_event)
+            events.sequencePauseToggled.connect(self._on_pause_toggled)
+            events.sequenceCanceled.connect(self._on_sequence_canceled)
+            events.sequenceFinished.connect(self._on_sequence_finished)
+
+        @property
+        def statusRequested(self) -> Any:
+            return self._status_emitter.statusRequested
 
         def _hide_tiff_sequence(self) -> None:
             """Remove the 'tiff-sequence' option from the save widget's writer combo."""
@@ -125,6 +160,123 @@ def prepare_mda(self) -> bool | str | Path | None:
                 output = "memory"
             return output
 
+        def _axis_value(self, event: MDAEvent | None, axis: str) -> int | None:
+            if event is None:
+                return None
+            for key, value in event.index.items():
+                if str(key) == axis:
+                    return int(value) + 1
+            return None
+
+        def _axis_total(self, axis: str) -> int | None:
+            if self._active_sequence is None:
+                return None
+            for key, value in self._active_sequence.sizes.items():
+                if str(key) == axis and value:
+                    return int(value)
+            return None
+
+        def _channel_name(self, event: MDAEvent | None) -> str | None:
+            if event is None or event.channel is None:
+                return None
+            return event.channel.config or None
+
+        def _event_produces_frame(self, event: MDAEvent) -> bool:
+            action = getattr(event, "action", None)
+            return action is None or isinstance(action, AcquireImage)
+
+        def _count_expected_frames(self, sequence: MDASequence) -> int:
+            return sum(
+                1
+                for event in sequence.iter_events()
+                if self._event_produces_frame(event)
+            )
+
+        def _format_status(
+            self,
+            *,
+            step: str,
+            event: MDAEvent | None = None,
+            next_seconds: float | None = None,
+        ) -> str:
+            current_event = event or self._last_event
+            parts = [f"Frame {self._frame_index}/{self._frame_total}"]
+
+            for axis, label in (("p", "Pos"), ("t", "T"), ("z", "Z")):
+                value = self._axis_value(current_event, axis)
+                total = self._axis_total(axis)
+                if value is not None and total:
+                    parts.append(f"{label} {value}/{total}")
+
+            if channel := self._channel_name(current_event):
+                parts.append(f"Channel {channel}")
+
+            parts.append(f"Step: {step}")
+
+            if next_seconds is not None:
+                parts.append(f"Next: {next_seconds:.1f} s")
+
+            return " | ".join(parts)
+
+        def _set_status(
+            self,
+            *,
+            step: str,
+            event: MDAEvent | None = None,
+            next_seconds: float | None = None,
+        ) -> None:
+            self.statusRequested.emit(
+                self._format_status(step=step, event=event, next_seconds=next_seconds)
+            )
+
+        def _on_sequence_started(
+            self, sequence: MDASequence, meta: Mapping[str, object]
+        ) -> None:
+            self._active_sequence = sequence
+            self._frame_total = self._count_expected_frames(sequence)
+            self._frame_index = 0
+            self._last_event = None
+            self._is_paused = False
+            self._was_canceled = False
+            self._set_status(step="Preparing")
+
+        def _on_event_started(self, event: MDAEvent) -> None:
+            self._last_event = event
+            action = getattr(event, "action", None)
+            if isinstance(action, HardwareAutofocus):
+                self._set_status(step="Autofocus", event=event)
+            else:
+                self._set_status(step="Acquiring", event=event)
+
+        def _on_frame_ready(
+            self, image: object, event: MDAEvent, meta: Mapping[str, object]
+        ) -> None:
+            self._last_event = event
+            self._frame_index += 1
+            self._set_status(step="Acquiring", event=event)
+
+        def _on_awaiting_event(self, event: MDAEvent, remaining_sec: float) -> None:
+            self._last_event = event
+            step = "Paused" if self._is_paused else "Waiting next frame"
+            self._set_status(step=step, event=event, next_seconds=remaining_sec)
+
+        def _on_pause_toggled(self, paused: bool) -> None:
+            self._is_paused = paused
+            step = "Paused" if paused else "Waiting next frame"
+            self._set_status(step=step)
+
+        def _on_sequence_canceled(self, sequence: MDASequence) -> None:
+            self._was_canceled = True
+            self._set_status(step="Canceled")
+
+        def _on_sequence_finished(self, sequence: MDASequence) -> None:
+            finish_reason = getattr(self._mmc.mda.status, "finish_reason", None)
+            if finish_reason is not None and str(finish_reason) == "errored":
+                step = "Error"
+            else:
+                step = "Canceled" if self._was_canceled else "Finished"
+            self._set_status(step=step)
+
     return MDAWidget(parent=parent, mmcore=_get_core(parent))
 
 

tests/test_actions.py

@@ -1,9 +1,18 @@
+from __future__ import annotations
+
+from collections.abc import Callable
+from datetime import timedelta
+from typing import Any, TypeAlias, cast
+
 import pytest
+import useq
 
 from pymmcore_gui import MicroManagerGUI
 from pymmcore_gui._qt.QtWidgets import QMenu, QWidget
 from pymmcore_gui.actions import ActionInfo, CoreAction, WidgetAction, WidgetActionInfo
 
+StatusTrigger: TypeAlias = Callable[[], None]
+
 
 def test_action_registry() -> None:
     info = ActionInfo.for_key(CoreAction.SNAP)
@@ -38,3 +47,56 @@ def test_actions_in_menus() -> None:
     )
     assert isinstance(window_menu, QMenu)
     assert any(a.text() == text for a in window_menu.actions())
+
+
+def test_mda_widget_status_line(qtbot: Any) -> None:
+    win = MicroManagerGUI()
+    qtbot.addWidget(win)
+    mda = cast("Any", win.get_widget(WidgetAction.MDA_WIDGET))
+
+    def emit_status(trigger: StatusTrigger) -> str:
+        with qtbot.waitSignal(mda.statusRequested) as blocker:
+            trigger()
+        args = cast("tuple[object, ...] | None", blocker.args)
+        assert args is not None
+        return str(args[0])
+
+    sequence = useq.MDASequence(
+        stage_positions=(
+            useq.Position(x=0.0, y=0.0, name="P1"),
+            useq.Position(x=1.0, y=1.0, name="P2"),
+        ),
+        channels=(useq.Channel(config="DAPI", exposure=1.0),),
+        time_plan=useq.TIntervalLoops(interval=timedelta(seconds=0.1), loops=2),
+    )
+    event = next(sequence.iter_events())
+
+    assert mda._status_label.text() == "Idle"
+    status = emit_status(lambda: mda._on_sequence_started(sequence, {}))
+    assert mda._frame_total == 4
+    assert status == "Frame 0/4 | Step: Preparing"
+
+    status = emit_status(lambda: mda._on_event_started(event))
+    assert status.endswith("Step: Acquiring")
+    assert "Pos 1/2" in status
+    assert "T 1/2" in status
+    assert "Channel DAPI" in status
+
+    status = emit_status(lambda: mda._on_frame_ready(object(), event, {}))
+    assert status.startswith("Frame 1/4")
+
+    status = emit_status(lambda: mda._on_awaiting_event(event, 1.25))
+    assert "Step: Waiting next frame" in status
+    assert "Next: 1.2 s" in status
+
+    status = emit_status(lambda: mda._on_pause_toggled(True))
+    assert "Step: Paused" in status
+
+    af_event = useq.MDAEvent(action=useq.HardwareAutofocus())
+    status = emit_status(lambda: mda._on_event_started(af_event))
+    assert "Step: Autofocus" in status
+
+    status = emit_status(lambda: mda._on_sequence_canceled(sequence))
+    assert "Step: Canceled" in status
+    status = emit_status(lambda: mda._on_sequence_finished(sequence))
+    assert "Step: Canceled" in status