refactor: Refactor out logic linking core events to viewer events fro…

…m the MainWindow (#297)

* refactor: coreviewerlink

* fix pre-commit

* fix timer event

* lint

src/napari_micromanager/_core_link.py

@@ -0,0 +1,95 @@
+from __future__ import annotations
+
+import contextlib
+from typing import TYPE_CHECKING, Callable
+
+import napari
+import napari.layers
+from pymmcore_plus import CMMCorePlus
+from qtpy.QtCore import QObject, Qt, QTimerEvent
+from superqt.utils import ensure_main_thread
+
+from ._mda_handler import _NapariMDAHandler
+
+if TYPE_CHECKING:
+    import napari.viewer
+    import numpy as np
+    from pymmcore_plus.core.events._protocol import PSignalInstance
+
+
+class CoreViewerLink(QObject):
+    """QObject linking events in a napari viewer to events in a CMMCorePlus instance."""
+
+    def __init__(
+        self,
+        viewer: napari.viewer.Viewer,
+        core: CMMCorePlus | None = None,
+        parent: QObject | None = None,
+    ) -> None:
+        super().__init__(parent)
+        self._mmc = core or CMMCorePlus.instance()
+        self.viewer = viewer
+        self._mda_handler = _NapariMDAHandler(self._mmc, viewer)
+        self._live_timer_id: int | None = None
+
+        # Add all core connections to this list.  This makes it easy to disconnect
+        # from core when this widget is closed.
+        self._connections: list[tuple[PSignalInstance, Callable]] = [
+            (self._mmc.events.imageSnapped, self._update_viewer),
+            (self._mmc.events.imageSnapped, self._stop_live),
+            (self._mmc.events.continuousSequenceAcquisitionStarted, self._start_live),
+            (self._mmc.events.sequenceAcquisitionStopped, self._stop_live),
+            (self._mmc.events.exposureChanged, self._restart_live),
+        ]
+        for signal, slot in self._connections:
+            signal.connect(slot)
+
+    def cleanup(self) -> None:
+        for signal, slot in self._connections:
+            with contextlib.suppress(TypeError, RuntimeError):
+                signal.disconnect(slot)
+        # Clean up temporary files we opened.
+        self._mda_handler._cleanup()
+
+    def timerEvent(self, a0: QTimerEvent | None) -> None:
+        self._update_viewer()
+
+    def _start_live(self) -> None:
+        interval = int(self._mmc.getExposure())
+        self._live_timer_id = self.startTimer(interval, Qt.TimerType.PreciseTimer)
+
+    def _stop_live(self) -> None:
+        if self._live_timer_id is not None:
+            self.killTimer(self._live_timer_id)
+            self._live_timer_id = None
+
+    def _restart_live(self, camera: str, exposure: float) -> None:
+        if self._live_timer_id:
+            self._mmc.stopSequenceAcquisition()
+            self._mmc.startContinuousSequenceAcquisition()
+
+    @ensure_main_thread  # type: ignore [misc]
+    def _update_viewer(self, data: np.ndarray | None = None) -> None:
+        """Update viewer with the latest image from the circular buffer."""
+        if data is None:
+            try:
+                data = self._mmc.getLastImage()
+            except (RuntimeError, IndexError):
+                # circular buffer empty
+                return
+        try:
+            preview_layer = self.viewer.layers["preview"]
+            preview_layer.data = data
+        except KeyError:
+            preview_layer = self.viewer.add_image(data, name="preview")
+
+        preview_layer.metadata["mode"] = "preview"
+
+        if (pix_size := self._mmc.getPixelSizeUm()) != 0:
+            preview_layer.scale = (pix_size, pix_size)
+        else:
+            # return to default
+            preview_layer.scale = [1.0, 1.0]
+
+        if self._live_timer_id is None:
+            self.viewer.reset_view()

src/napari_micromanager/main_window.py

@@ -2,52 +2,38 @@
 
 import atexit
 import contextlib
+import logging
 from typing import TYPE_CHECKING, Any, Callable
 
 import napari
 import napari.layers
 import napari.viewer
 from pymmcore_plus import CMMCorePlus
-from pymmcore_plus._util import find_micromanager
-from qtpy.QtCore import QTimer
-from superqt.utils import create_worker, ensure_main_thread
 
+from ._core_link import CoreViewerLink
 from ._gui_objects._toolbar import MicroManagerToolbar
-from ._mda_handler import _NapariMDAHandler
 
 if TYPE_CHECKING:
-    import numpy as np
     from pymmcore_plus.core.events._protocol import PSignalInstance
 
 
+# this is very verbose
+logging.getLogger("napari.loader").setLevel(logging.WARNING)
+
+
 class MainWindow(MicroManagerToolbar):
     """The main napari-micromanager widget that gets added to napari."""
 
     def __init__(self, viewer: napari.viewer.Viewer) -> None:
-        adapter_path = find_micromanager()
-        if not adapter_path:
-            raise RuntimeError(
-                "Could not find micromanager adapters. Please run "
-                "`mmcore install` or install manually and set "
-                "MICROMANAGER_PATH."
-            )
-
         super().__init__(viewer)
 
         # get global CMMCorePlus instance
         self._mmc = CMMCorePlus.instance()
+        # this object mediates the connection between the viewer and core events
+        self._core_link = CoreViewerLink(viewer, self._mmc, self)
 
-        self._mda_handler = _NapariMDAHandler(self._mmc, viewer)
-        self.streaming_timer: QTimer | None = None
-
-        # Add all core connections to this list.  This makes it easy to disconnect
-        # from core when this widget is closed.
+        # some remaining connections related to widgets ... TODO: unify with superclass
         self._connections: list[tuple[PSignalInstance, Callable]] = [
-            (self._mmc.events.exposureChanged, self._update_live_exp),
-            (self._mmc.events.imageSnapped, self._update_viewer),
-            (self._mmc.events.imageSnapped, self._stop_live),
-            (self._mmc.events.continuousSequenceAcquisitionStarted, self._start_live),
-            (self._mmc.events.sequenceAcquisitionStopped, self._stop_live),
             (self.viewer.layers.events, self._update_max_min),
             (self.viewer.layers.selection.events, self._update_max_min),
             (self.viewer.dims.events.current_step, self._update_max_min),
@@ -68,60 +54,11 @@ def _cleanup(self) -> None:
             with contextlib.suppress(TypeError, RuntimeError):
                 signal.disconnect(slot)
         # Clean up temporary files we opened.
-        self._mda_handler._cleanup()
+        self._core_link.cleanup()
         atexit.unregister(self._cleanup)  # doesn't raise if not connected
 
-    @ensure_main_thread  # type: ignore [misc]
-    def _update_viewer(self, data: np.ndarray | None = None) -> None:
-        """Update viewer with the latest image from the circular buffer."""
-        if data is None:
-            try:
-                data = self._mmc.getLastImage()
-            except (RuntimeError, IndexError):
-                # circular buffer empty
-                return
-        try:
-            preview_layer = self.viewer.layers["preview"]
-            preview_layer.data = data
-        except KeyError:
-            preview_layer = self.viewer.add_image(data, name="preview")
-
-        preview_layer.metadata["mode"] = "preview"
-
-        if (pix_size := self._mmc.getPixelSizeUm()) != 0:
-            preview_layer.scale = (pix_size, pix_size)
-        else:
-            # return to default
-            preview_layer.scale = [1.0, 1.0]
-
-        self._update_max_min()
-
-        if self.streaming_timer is None:
-            self.viewer.reset_view()
-
     def _update_max_min(self, *_: Any) -> None:
         visible = (x for x in self.viewer.layers.selection if x.visible)
         self.minmax.update_from_layers(
             lr for lr in visible if isinstance(lr, napari.layers.Image)
         )
-
-    def _snap(self) -> None:
-        # update in a thread so we don't freeze UI
-        create_worker(self._mmc.snap, _start_thread=True)
-
-    def _start_live(self) -> None:
-        self.streaming_timer = QTimer()
-        self.streaming_timer.timeout.connect(self._update_viewer)
-        self.streaming_timer.start(int(self._mmc.getExposure()))
-
-    def _stop_live(self) -> None:
-        if self.streaming_timer:
-            self.streaming_timer.stop()
-            self.streaming_timer.deleteLater()
-            self.streaming_timer = None
-
-    def _update_live_exp(self, camera: str, exposure: float) -> None:
-        if self.streaming_timer:
-            self.streaming_timer.setInterval(int(exposure))
-            self._mmc.stopSequenceAcquisition()
-            self._mmc.startContinuousSequenceAcquisition(exposure)

tests/test_layer_scale.py

@@ -66,12 +66,8 @@ def test_layer_scale(
 
 
 def test_preview_scale(core: CMMCorePlus, main_window: MainWindow):
-    """Basic test to check that the main window can be created.
-
-    This test should remain fast.
-    """
     img = core.snap()
-    main_window._update_viewer(img)
+    main_window._core_link._update_viewer(img)
 
     pix_size = core.getPixelSizeUm()
     assert tuple(main_window.viewer.layers["preview"].scale) == (pix_size, pix_size)
@@ -82,7 +78,7 @@ def test_preview_scale(core: CMMCorePlus, main_window: MainWindow):
     core.setPixelSizeUm("Res20x", 0)
 
     try:
-        main_window._update_viewer(img)
+        main_window._core_link._update_viewer(img)
     except Exception as e:
         # return to orig value for future tests and re-raise
         core.setPixelSizeUm(pix_size)

tests/test_main_window.py

@@ -20,12 +20,12 @@ def test_main_window(qtbot: QtBot, core: CMMCorePlus) -> None:
     qtbot.addWidget(wdg)
 
     viewer.layers.events.connect.assert_called_once_with(wdg._update_max_min)
-    wdg._snap()
+    core.snap()
 
-    wdg._update_viewer()
+    wdg._core_link._update_viewer()
     wdg._mmc.startContinuousSequenceAcquisition()
     wdg._mmc.stopSequenceAcquisition()
-    wdg._update_viewer()
+    wdg._core_link._update_viewer()
 
     wdg._cleanup()
     viewer.layers.events.disconnect.assert_called_once_with(wdg._update_max_min)