feat: Core log widget (#429)

Author: gselzer

docs/widget_list.json

@@ -39,6 +39,7 @@
     "ChannelWidget",
     "ImagePreview",
     "SnapButton",
-    "LiveButton"
+    "LiveButton",
+    "CoreLogWidget"
   ]
 }

examples/core_log_widget.py

@@ -0,0 +1,8 @@
+from qtpy.QtWidgets import QApplication
+
+from pymmcore_widgets import CoreLogWidget
+
+app = QApplication([])
+wdg = CoreLogWidget()
+wdg.show()
+app.exec()

src/pymmcore_widgets/__init__.py

@@ -16,6 +16,7 @@
     "ChannelWidget",
     "ConfigWizard",
     "ConfigurationWidget",
+    "CoreLogWidget",
     "DefaultCameraExposureWidget",
     "DeviceWidget",
     "ExposureWidget",
@@ -45,6 +46,7 @@
 ]
 
 from ._install_widget import InstallWidget
+from ._log import CoreLogWidget
 from .config_presets import (
     GroupPresetTableWidget,
     ObjectivesPixelConfigurationWidget,

src/pymmcore_widgets/_log.py

@@ -0,0 +1,179 @@
+from __future__ import annotations
+
+import os
+from collections import deque
+from typing import TYPE_CHECKING
+
+from pymmcore_plus import CMMCorePlus
+from qtpy.QtCore import QFileSystemWatcher, QObject, QTimer, QUrl, Signal
+from qtpy.QtGui import QCloseEvent, QDesktopServices, QFontDatabase, QPalette
+from qtpy.QtWidgets import (
+    QApplication,
+    QHBoxLayout,
+    QPlainTextEdit,
+    QPushButton,
+    QSizePolicy,
+    QVBoxLayout,
+    QWidget,
+)
+from superqt import QElidingLabel, QIconifyIcon
+
+if TYPE_CHECKING:
+    from io import TextIOWrapper
+
+
+class _LogReader(QObject):
+    """Watches a log file and emits new lines as they arrive."""
+
+    new_lines: Signal = Signal(str)
+    finished: Signal = Signal()
+
+    def __init__(
+        self,
+        path: str,
+        interval: int = 200,
+        parent: QObject | None = None,
+    ) -> None:
+        super().__init__(parent)
+        self._path = path
+        self._interval = interval
+        self._file: TextIOWrapper | None = None
+
+        # Unfortunately, on Windows, QFileSystemWatcher does not detect file changes
+        # unless the file is flushed from cache to disk. This does NOT happen
+        # when CMMCorePlus.logMessage() is called. So we need to poll the file for
+        # Windows' sake.
+        self._timer = QTimer(self)
+        self._timer.setInterval(self._interval)
+        self._timer.timeout.connect(self._read_new)
+
+        # Watcher for rotation/truncate events
+        self._watcher = QFileSystemWatcher(self)
+        self._watcher.addPath(self._path)
+        self._watcher.fileChanged.connect(self._on_file_changed)
+
+    def start(self) -> None:
+        """Open the file and start polling."""
+        self._file = open(self._path, encoding="utf-8", errors="replace")
+        self._file.seek(0, os.SEEK_END)
+        self._timer.start()
+
+    def _stop(self) -> None:
+        """Stop polling and close the file."""
+        self._timer.stop()
+        if self._file:
+            self._file.close()
+        self.finished.emit()
+
+    def _on_file_changed(self, path: str) -> None:
+        """Handle log rotation or truncation."""
+        try:
+            real_size = os.path.getsize(path)
+            current_pos = self._file.tell() if self._file else 0
+            if real_size < current_pos:
+                # rotated or truncated
+                if self._file:
+                    self._file.close()
+                self._file = open(self._path, encoding="utf-8", errors="replace")
+            self._read_new()
+        except Exception:
+            pass
+
+    def _read_new(self) -> None:
+        """Read and emit any new lines."""
+        if not self._file:
+            return
+        for line in self._file:
+            self.new_lines.emit(line.rstrip("\n"))
+
+
+class CoreLogWidget(QWidget):
+    """High-performance log console with pause, follow-tail, clear, and initial load."""
+
+    def __init__(
+        self,
+        path: str | None = None,
+        max_lines: int = 5_000,
+        parent: QWidget | None = None,
+        mmcore: CMMCorePlus | None = None,
+    ) -> None:
+        super().__init__(parent)
+        self._mmcore = mmcore or CMMCorePlus().instance()
+        self.setWindowTitle("Log Console")
+
+        # --- Log path ---
+        self._log_path = QElidingLabel()
+        self._log_path.setSizePolicy(
+            QSizePolicy.Policy.Expanding, QSizePolicy.Policy.Minimum
+        )
+
+        self._clear_btn = QPushButton("Clear Display")
+        self._clear_btn.setToolTip(
+            "Clears this view. Does not delete lines from the log file."
+        )
+
+        self._log_btn = QPushButton()
+        color = QApplication.palette().color(QPalette.ColorRole.WindowText).name()
+        self._log_btn.setIcon(QIconifyIcon("majesticons:open", color=color))
+        self._log_btn.setToolTip("Open log file in native editor")
+
+        # --- Log view ---
+        self._log_view = QPlainTextEdit(self)
+        self._log_view.setReadOnly(True)
+        self._log_view.setMaximumBlockCount(max_lines)
+        self._log_view.setLineWrapMode(QPlainTextEdit.LineWrapMode.NoWrap)
+        # Monospaced font
+        fixed_font = QFontDatabase.systemFont(QFontDatabase.SystemFont.FixedFont)
+        fixed_font.setPixelSize(12)
+        self._log_view.setFont(fixed_font)
+
+        path = path or self._mmcore.getPrimaryLogFile()
+        self._log_path.setText(path)
+        # Load the last `max_lines` from file
+        try:
+            with open(path, encoding="utf-8", errors="replace") as f:
+                for line in deque(f, maxlen=max_lines):
+                    self._log_view.appendPlainText(line.rstrip("\n"))
+        except Exception:
+            pass
+
+        # --- Reader thread setup ---
+        self._reader = _LogReader(path)
+
+        # --- Layout ---
+        file_layout = QHBoxLayout()
+        file_layout.setContentsMargins(5, 5, 5, 0)
+        file_layout.addWidget(self._log_path)
+        file_layout.addWidget(self._clear_btn)
+        file_layout.addWidget(self._log_btn)
+
+        layout = QVBoxLayout(self)
+        layout.setContentsMargins(0, 0, 0, 0)
+        layout.addLayout(file_layout)
+        layout.addWidget(self._log_view)
+        self.setLayout(layout)
+
+        # --- Connections ---
+        self._reader.new_lines.connect(self._append_line)
+        self._clear_btn.clicked.connect(self._log_view.clear)
+        self._log_btn.clicked.connect(self._open_native)
+        self._reader.start()
+
+    def __del__(self) -> None:
+        """Stop reader before deletion."""
+        self._reader._stop()
+
+    def _append_line(self, line: str) -> None:
+        """Append a line, respecting pause/follow settings."""
+        self._log_view.appendPlainText(line)
+
+    def closeEvent(self, event: QCloseEvent | None) -> None:
+        """Clean up thread on close."""
+        self._reader._stop()
+        # self._thread.quit()
+        # self._thread.wait()
+        super().closeEvent(event)
+
+    def _open_native(self) -> None:
+        """Open the log file in the system's default text editor."""
+        QDesktopServices.openUrl(QUrl.fromLocalFile(self._mmcore.getPrimaryLogFile()))

tests/test_core_log_widget.py

@@ -0,0 +1,98 @@
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
+from qtpy.QtWidgets import QApplication
+
+from pymmcore_widgets import CoreLogWidget
+
+if TYPE_CHECKING:
+    from pymmcore_plus import CMMCorePlus
+    from pytestqt.qtbot import QtBot
+
+
+def test_core_log_widget_init(qtbot: QtBot, global_mmcore: CMMCorePlus) -> None:
+    """Asserts that the CoreLogWidget initializes with the entire log to this point."""
+    wdg = CoreLogWidget()
+    qtbot.addWidget(wdg)
+
+    # Assert log path is in the widget LineEdit
+    log_path = global_mmcore.getPrimaryLogFile()
+    assert log_path == wdg._log_path.text()
+
+    # Assert log content is in the widget TextEdit
+    # This is a bit tricky because more can be appended to the log file.
+    with open(log_path) as f:
+        log_content = [s.strip() for s in f.readlines()]
+        # Trim down to the final 5000 lines if necessary
+        # (this is all that will fit in the Log Widget)
+        max_lines = wdg._log_view.maximumBlockCount()
+        if len(log_content) > max_lines:
+            log_content = log_content[-max_lines:]
+    edit_content = [s.strip() for s in wdg._log_view.toPlainText().splitlines()]
+    min_length = min(len(log_content), len(edit_content))
+    for i in range(min_length):
+        assert log_content[i] == edit_content[i]
+
+
+def test_core_log_widget_update(qtbot: QtBot, global_mmcore: CMMCorePlus) -> None:
+    wdg = CoreLogWidget()
+    qtbot.addWidget(wdg)
+    # Remove some lines for faster checking later
+    wdg._log_view.clear()
+
+    # Log something new
+    new_message = "Test message"
+    global_mmcore.logMessage(new_message)
+
+    def wait_for_update() -> None:
+        # Sometimes, our new message will be flushed before other initialization
+        # completes. Thus we need to check all lines after what is currently written to
+        # the TextEdit.
+        all_lines = wdg._log_view.toPlainText().splitlines()
+        for line in reversed(all_lines):
+            if f"[IFO,App] {new_message}" in line:
+                return
+        raise AssertionError("New message not found in CoreLogWidget.")
+
+    qtbot.waitUntil(wait_for_update, timeout=1000)
+
+
+def test_core_log_widget_clear(qtbot: QtBot, global_mmcore: CMMCorePlus) -> None:
+    wdg = CoreLogWidget()
+    qtbot.addWidget(wdg)
+
+    assert wdg._log_view.toPlainText() != ""
+    wdg._clear_btn.click()
+    assert wdg._log_view.toPlainText() == ""
+
+
+def test_core_log_widget_autoscroll(qtbot: QtBot, global_mmcore: CMMCorePlus) -> None:
+    wdg = CoreLogWidget()
+    qtbot.addWidget(wdg)
+    # Note that we must show the widget for the scrollbar maximum to be computed
+    wdg.show()
+    sb = wdg._log_view.verticalScrollBar()
+    assert sb is not None
+
+    def add_new_line() -> None:
+        wdg._append_line("Test message")
+        QApplication.processEvents()
+
+    # Make sure we have a scrollbar with nonzero size to test with
+    # But we don't want it full yet
+    wdg._log_view.clear()
+    while sb.maximum() == 0:
+        add_new_line()
+
+    # Assert that adding a new line does not scroll if not at the bottom
+    sb.setValue(sb.minimum())
+    add_new_line()
+    assert sb.value() == sb.minimum()
+
+    # Assert that adding a new line does scroll if at the bottom
+    old_max = sb.maximum()
+    sb.setValue(old_max)
+    add_new_line()
+    assert sb.maximum() == old_max + 1
+    assert sb.value() == sb.maximum()