fix: keep dashboard logging inside curses UI

This commit is contained in:
Andrii Prokhorov
2025-11-21 18:19:33 +02:00
parent 5c0ad3fb72
commit 7c8321e3c4

View File

@@ -3,6 +3,8 @@
from __future__ import annotations
import curses
import logging
from collections import deque
from dataclasses import dataclass
from typing import Optional
@@ -35,6 +37,10 @@ class ReactorDashboard:
self.pending_command: Optional[ReactorCommand] = None
self.sim: Optional[ReactorSimulation] = None
self.quit_requested = False
self.log_buffer: deque[str] = deque(maxlen=4)
self._log_handler: Optional[logging.Handler] = None
self._previous_handlers: list[logging.Handler] = []
self._logger = logging.getLogger("reactor_sim")
self.keys = [
DashboardKey("q", "Quit & save"),
DashboardKey("space", "SCRAM"),
@@ -61,6 +67,7 @@ class ReactorDashboard:
curses.init_pair(3, curses.COLOR_GREEN, -1)
curses.init_pair(4, curses.COLOR_RED, -1)
stdscr.nodelay(True)
self._install_log_capture()
self.sim = ReactorSimulation(
self.reactor,
timestep=self.timestep,
@@ -79,6 +86,7 @@ class ReactorDashboard:
finally:
if self.save_path and self.sim and self.sim.last_state:
self.reactor.save_state(self.save_path, self.sim.last_state)
self._restore_logging()
def _handle_input(self, stdscr: "curses._CursesWindow") -> None:
while True:
@@ -265,7 +273,13 @@ class ReactorDashboard:
f"Failures: {', '.join(self.reactor.health_monitor.failure_log)}",
curses.color_pair(4) | curses.A_BOLD,
)
win.addstr(4, 1, "Press 'q' to exit and persist the current snapshot.", curses.color_pair(2))
log_y = 3
for record in list(self.log_buffer):
if log_y >= win.getmaxyx()[0] - 1:
break
win.addstr(log_y, 1, record[: win.getmaxyx()[1] - 2], curses.color_pair(2))
log_y += 1
win.addstr(log_y, 1, "Press 'q' to exit and persist the current snapshot.", curses.color_pair(2))
def _draw_section(
self,
@@ -320,3 +334,34 @@ class ReactorDashboard:
def _clamped_rod(self, delta: float) -> float:
new_fraction = self.reactor.control.rod_fraction + delta
return max(0.0, min(0.95, new_fraction))
def _install_log_capture(self) -> None:
if self._log_handler:
return
self._previous_handlers = list(self._logger.handlers)
for handler in self._previous_handlers:
self._logger.removeHandler(handler)
handler = _DashboardLogHandler(self.log_buffer)
handler.setFormatter(logging.Formatter("%(levelname)s | %(message)s"))
self._logger.addHandler(handler)
self._logger.propagate = False
self._log_handler = handler
def _restore_logging(self) -> None:
if not self._log_handler:
return
self._logger.removeHandler(self._log_handler)
for handler in self._previous_handlers:
self._logger.addHandler(handler)
self._log_handler = None
self._previous_handlers = []
class _DashboardLogHandler(logging.Handler):
def __init__(self, buffer: deque[str]) -> None:
super().__init__()
self.buffer = buffer
def emit(self, record: logging.LogRecord) -> None:
msg = self.format(record)
self.buffer.append(msg)