229 lines
9.4 KiB
Python
229 lines
9.4 KiB
Python
"""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, Qt
|
|
from PyQt6.QtGui import QTextCursor
|
|
from PyQt6.QtWidgets import QComboBox, QHBoxLayout, QLabel, QMessageBox, QToolButton, 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"<span style='color:{accent}; font-weight:700;'>{html.escape(level)}</span>"
|
|
f" <span style='color:#60758d;'>{timestamp}</span>"
|
|
f" <span style='color:{message};'>{html.escape(text).replace(chr(10), '<br>')}</span>")
|
|
if details:
|
|
body += f"<pre style='margin:3px 0 0 16px; color:{detail};'>{html.escape(details)}</pre>"
|
|
return f"<div style='margin:0 0 6px 0;'>{body}</div>"
|
|
|
|
|
|
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 = QToolButton()
|
|
self._toggle.setAutoRaise(True)
|
|
self._toggle.setArrowType(Qt.ArrowType.DownArrow)
|
|
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(130)
|
|
self._box.setMaximumHeight(140)
|
|
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.setArrowType(Qt.ArrowType.DownArrow if visible else Qt.ArrowType.RightArrow)
|
|
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()
|