diff --git a/src/pymmcore_gui/actions/widget_actions.py b/src/pymmcore_gui/actions/widget_actions.py index 634b5e6c..53582b98 100644 --- a/src/pymmcore_gui/actions/widget_actions.py +++ b/src/pymmcore_gui/actions/widget_actions.py @@ -2,16 +2,17 @@ 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 @@ -19,7 +20,6 @@ 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)) diff --git a/tests/test_actions.py b/tests/test_actions.py index 34b9eee6..2321af9d 100644 --- a/tests/test_actions.py +++ b/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