1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
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 ""
|
