"""Bounded GUI logging, error presentation and request-scoped error capture.""" from __future__ import annotations from collections import deque from collections.abc import Callable, Iterator from contextlib import contextmanager from datetime import datetime import html import logging from pathlib import Path import threading import traceback from PyQt6.QtCore import QSignalBlocker, QTimer from PyQt6.QtGui import QTextCursor from PyQt6.QtWidgets import QComboBox, QHBoxLayout, QLabel, QMessageBox, QPushButton, QTextEdit, QVBoxLayout, QWidget from python_app.gui.settings_defaults import env_enabled from python_app.logging_setup import LOG_LEVELS, add_handler, configure_logging, get_logger, package_logger, set_log_level class _PanelLogBuffer: """Worker threads enqueue records; the GUI flushes one bounded batch per tick.""" _CAPACITY = 2000 def __init__(self) -> None: self._lock = threading.Lock() self._entries: deque[tuple[str, str, str | None, str | None]] = deque(maxlen=self._CAPACITY) self._dropped_count = 0 def append(self, level: str, text: str, details: str | None, once_key: str | None) -> None: with self._lock: if len(self._entries) == self._CAPACITY: self._dropped_count += 1 self._entries.append((level, text, details, once_key)) def drain(self) -> tuple[list[tuple[str, str, str | None, str | None]], int]: with self._lock: entries, dropped = list(self._entries), self._dropped_count self._entries.clear() self._dropped_count = 0 return entries, dropped class _PanelHandler(logging.Handler): def __init__(self, buffer: _PanelLogBuffer) -> None: super().__init__() self._buffer = buffer def emit(self, record: logging.LogRecord) -> None: try: details = getattr(record, "panel_details", None) once_key = getattr(record, "panel_once_key", None) self._buffer.append( "WARN" if record.levelname == "WARNING" else record.levelname, record.getMessage(), details if isinstance(details, str) else None, once_key if isinstance(once_key, str) else None, ) except Exception: self.handleError(record) def _entry_html(level: str, text: str, details: str | None = None) -> str: palette = {"DEBUG": ("#6c7b8d", "#52627a", "#8a97a8"), "INFO": ("#1d5fbf", "#1f2937", "#526277"), "WARN": ("#9a5b00", "#5c4300", "#7a6640"), "ERROR": ("#c43d4d", "#6b1f2a", "#8b5d66")} accent, message, detail = palette.get(level.upper(), palette["INFO"]) timestamp = datetime.now().strftime("%H:%M:%S.%f")[:-3] body = (f"{html.escape(level)}" f" {timestamp}" f" {html.escape(text).replace(chr(10), '
')}") if details: body += f"
{html.escape(details)}
" return f"
{body}
" class LogController: def __init__(self, parent: QWidget, project_root: Path, status: Callable[[str], None]) -> None: self._parent, self._status = parent, status configure_logging(log_dir=project_root / "python_app/runtime/logs") self._logger = get_logger("gui") self._buffer = _PanelLogBuffer() self._seen: set[str] = set() self._capture: list[str] | None = None self.panel = QWidget(parent) layout = QVBoxLayout(self.panel) layout.setContentsMargins(0, 0, 0, 0) layout.setSpacing(6) header = QHBoxLayout() header.setSpacing(8) title = QLabel("Runtime Log") title.setObjectName("hintLabel") header.addWidget(title) header.addStretch(1) label = QLabel("Level") label.setObjectName("hintLabel") header.addWidget(label) self.level = QComboBox() self.level.setObjectName("logLevelCombo") self.level.addItems([name.capitalize() for name in LOG_LEVELS]) self.level.setCurrentText("Info") self.level.setToolTip("Logging verbosity — applied live and saved to run_config") self.level.currentTextChanged.connect(self.set_level) header.addWidget(self.level) self._toggle = QPushButton("Collapse") self._toggle.setObjectName("sectionToggleButton") self._toggle.setToolTip("Hide runtime log entries") self._toggle.clicked.connect(self.toggle) header.addWidget(self._toggle) layout.addLayout(header) self._box = QTextEdit() self._box.setObjectName("runtimeLogBox") self._box.setReadOnly(True) self._box.setUndoRedoEnabled(False) self._box.setMinimumHeight(170) document = self._box.document() assert document is not None document.setMaximumBlockCount(1200) layout.addWidget(self._box) self._handler = _PanelHandler(self._buffer) add_handler(self._handler) self._timer = QTimer(self.panel) self._timer.setInterval(100) self._timer.timeout.connect(self.flush) self._timer.start() def set_level(self, text: str) -> None: with QSignalBlocker(self.level): self.level.setCurrentText(text.capitalize()) set_log_level(text) self.info(f"Log level set to {text.upper()}.") def toggle(self) -> None: visible = self._box.isHidden() self._box.setVisible(visible) self._toggle.setText("Collapse" if visible else "Expand") self._toggle.setToolTip("Hide runtime log entries" if visible else "Show runtime log entries") def flush(self) -> None: entries, dropped = self._buffer.drain() rendered: list[str] = [] if dropped: rendered.append(_entry_html("WARN", f"Log panel overflow: {dropped} record(s) dropped (they are still in the log file).")) for level, text, details, once_key in entries: if once_key is not None: if once_key in self._seen: continue self._seen.add(once_key) rendered.append(_entry_html(level, text, details)) if level == "ERROR": self._status("Status: error") if rendered: cursor = self._box.textCursor() cursor.movePosition(QTextCursor.MoveOperation.End) self._box.setTextCursor(cursor) self._box.insertHtml("".join(rendered)) self._box.insertPlainText("\n") self._box.ensureCursorVisible() def debug(self, text: str, *, once_key: str | None = None) -> None: self._logger.debug(text, extra={"panel_once_key": once_key}) def info(self, text: str, *, once_key: str | None = None) -> None: self._logger.info(text, extra={"panel_once_key": once_key}) def warning(self, text: str, *, details: str | None = None, once_key: str | None = None) -> None: self._logger.warning(text, extra={"panel_details": details, "panel_once_key": once_key}) def error(self, text: str, *, details: str | None = None, once_key: str | None = None) -> None: self._logger.error(text, extra={"panel_details": details, "panel_once_key": once_key}) @staticmethod def exception_summary(exc: Exception) -> str: message = str(exc).strip() return f"{type(exc).__name__}: {message}" if message else type(exc).__name__ @staticmethod def exception_details(exc: Exception) -> str: return "".join(traceback.TracebackException.from_exception(exc).format(chain=True)).strip() def exception(self, context: str, exc: Exception, *, level: str = "ERROR") -> tuple[str, str]: message, details = f"{context}: {self.exception_summary(exc)}", self.exception_details(exc) (self.warning if level.upper() in {"WARN", "WARNING"} else self.error)(message, details=details) return message, details @contextmanager def capture_errors(self) -> Iterator[list[str]]: previous, captured = self._capture, [] self._capture = captured try: yield captured finally: self._capture = previous @contextmanager def suspend_error_capture(self) -> Iterator[None]: """Keep unrelated Qt callbacks out of the active web or GPIO action.""" previous, self._capture = self._capture, None try: yield finally: self._capture = previous def show_error(self, message: str, *, details: str | None = None) -> None: self.error(message, details=details) if self._capture is not None: if not self._capture: self._capture.append(message) return if env_enabled("RADAR_SYSTEM_HEADLESS"): return dialog = QMessageBox(self._parent) dialog.setIcon(QMessageBox.Icon.Critical) dialog.setWindowTitle("Error") dialog.setText(message) if details: dialog.setDetailedText(details) dialog.exec() def show_exception(self, context: str, exc: Exception) -> None: self.show_error(f"{context}: {self.exception_summary(exc)}", details=self.exception_details(exc)) def close(self) -> None: self._timer.stop() package_logger().removeHandler(self._handler) self._handler.close()