diff options
| author | Christian Kolset <ckolset@colostate.edu> | 2026-07-29 13:40:09 -0600 |
|---|---|---|
| committer | Christian Kolset <ckolset@colostate.edu> | 2026-07-29 13:40:09 -0600 |
| commit | 23800a32231acecbace12bfc41c03b081b1bf565 (patch) | |
| tree | 213caf20b68b95ba674cf29d4e806a9b59aadb0d /core/debug_log.py | |
| parent | 2c6b4b751ffdd1156254b31745597654b5f2e286 (diff) | |
| parent | cf33095f77e6ff902fb69932fa704eb6192b3418 (diff) | |
Merge branch 'feat/dev-debug-window'
Diffstat (limited to 'core/debug_log.py')
| -rw-r--r-- | core/debug_log.py | 65 |
1 files changed, 65 insertions, 0 deletions
diff --git a/core/debug_log.py b/core/debug_log.py new file mode 100644 index 0000000..86a8a59 --- /dev/null +++ b/core/debug_log.py @@ -0,0 +1,65 @@ +""" +core/debug_log.py + +Tees stdout/stderr into an in-memory ring buffer + Qt signal so the Debug +window can show everything the app has printed since startup — including +messages from background poll threads (e.g. "[CMLLayer] poll error: ...") — +not just whatever gets printed while the window happens to be open. + +install() should be called once, early, before anything prints. The real +streams are still written to, so running from a terminal is unaffected. +""" + +from __future__ import annotations + +import sys +from collections import deque +from typing import Optional + +from PyQt6.QtCore import QObject, pyqtSignal + + +class _Broadcaster(QObject): + line_written = pyqtSignal(str) + + +class _StreamTee: + def __init__(self, real_stream, lines: deque, broadcaster: _Broadcaster): + self._real = real_stream + self._lines = lines + self._broadcaster = broadcaster + + def write(self, text: str) -> None: + self._real.write(text) + if text: + self._lines.append(text) + self._broadcaster.line_written.emit(text) + + def flush(self) -> None: + self._real.flush() + + def isatty(self) -> bool: + return False + + +_lines: Optional[deque] = None +_broadcaster: Optional[_Broadcaster] = None + + +def install(max_lines: int = 2000) -> None: + """Redirect sys.stdout/sys.stderr through the tee. Safe to call once.""" + global _lines, _broadcaster + if _broadcaster is not None: + return + _lines = deque(maxlen=max_lines) + _broadcaster = _Broadcaster() + sys.stdout = _StreamTee(sys.stdout, _lines, _broadcaster) + sys.stderr = _StreamTee(sys.stderr, _lines, _broadcaster) + + +def get_broadcaster() -> Optional[_Broadcaster]: + return _broadcaster + + +def get_history() -> str: + return "".join(_lines) if _lines is not None else "" |
