Add logs and richer panels to Textual dashboard

This commit is contained in:
Codex Agent
2025-11-27 13:24:48 +01:00
parent 5a22f7471f
commit 00434e3a88

View File

@@ -79,6 +79,9 @@ class TextualDashboard(App):
self._pending: deque[ReactorCommand] = deque()
self._timer: Optional[Timer] = None
self._trend_history: deque[tuple[float, float, float]] = deque(maxlen=120)
self.log_buffer: deque[str] = deque(maxlen=8)
self._log_handler: Optional[logging.Handler] = None
self._previous_handlers: list[logging.Handler] = []
snap_at_env = os.getenv("FISSION_SNAPSHOT_AT")
self.snapshot_at = float(snap_at_env) if snap_at_env else None
self.snapshot_done = False
@@ -98,6 +101,7 @@ class TextualDashboard(App):
self.maintenance_panel = Static(classes="panel")
self.health_panel = Static(classes="panel")
self.help_panel = Static(classes="panel")
self.log_panel = Static(classes="panel")
self.status_panel = Static(classes="panel")
def compose(self) -> ComposeResult:
@@ -117,11 +121,13 @@ class TextualDashboard(App):
yield self.protection_panel
yield self.maintenance_panel
yield self.health_panel
yield self.log_panel
yield self.help_panel
yield self.status_panel
yield Footer()
def on_mount(self) -> None:
self._install_log_capture()
self._timer = self.set_interval(self.timestep, self._tick, pause=False)
self._render_panels()
@@ -130,6 +136,7 @@ class TextualDashboard(App):
self._timer.pause()
if self.save_path and self.state:
self.reactor.save_state(self.save_path, self.state)
self._restore_logging()
self.exit()
# Command helpers
@@ -260,6 +267,7 @@ class TextualDashboard(App):
self.protection_panel.update(self._protection_text())
self.maintenance_panel.update(self._maintenance_text())
self.health_panel.update(self._health_text())
self.log_panel.update(self._log_text())
self.help_panel.update(self._help_text())
failures = ", ".join(self.reactor.health_monitor.failure_log) if self.reactor.health_monitor.failure_log else "None"
self.status_panel.update(
@@ -388,6 +396,14 @@ class TextualDashboard(App):
]
return "[bold cyan]Controls & Tips[/bold cyan]\\n" + "\\n".join(f"- {t}" for t in tips)
def _log_text(self) -> str:
lines = ["[bold cyan]Logs[/bold cyan]"]
if not self.log_buffer:
lines.append("No recent logs.")
else:
lines.extend(list(self.log_buffer))
return "\\n".join(lines)
def _turbine_status_lines(self) -> list[str]:
if not self.reactor.turbine_unit_active:
return []
@@ -450,8 +466,29 @@ class TextualDashboard(App):
except Exception as exc: # pragma: no cover
LOGGER.error("Failed to save snapshot: %s", exc)
def _install_log_capture(self) -> None:
logger = logging.getLogger("reactor_sim")
self._previous_handlers = list(logger.handlers)
handler = logging.StreamHandler()
handler.setLevel(logging.INFO)
def emit(record: logging.LogRecord) -> None:
msg = handler.format(record)
self.log_buffer.append(msg)
handler.emit = emit # type: ignore[assignment]
logger.handlers = [handler]
logger.setLevel(logging.INFO)
self._log_handler = handler
def _restore_logging(self) -> None:
logger = logging.getLogger("reactor_sim")
if self._previous_handlers:
logger.handlers = self._previous_handlers
if self._log_handler and self._log_handler in logger.handlers:
logger.removeHandler(self._log_handler)
def run_textual_dashboard(reactor: Reactor, start_state: Optional[PlantState], timestep: float, save_path: Optional[str]) -> None:
app = TextualDashboard(reactor, start_state, timestep=timestep, save_path=save_path)
app.run()