Add optional rich-based alternate dashboard

src/reactor_sim/rich_dashboard.py

@@ -0,0 +1,131 @@
+"""Alternate dashboard using rich Live rendering (non-interactive)."""
+
+from __future__ import annotations
+
+import logging
+from typing import Optional
+
+from rich.console import Group
+from rich.live import Live
+from rich.panel import Panel
+from rich.table import Table
+from rich.layout import Layout
+from rich.text import Text
+
+from .reactor import Reactor
+from .simulation import ReactorSimulation
+from .state import PlantState
+from . import constants
+
+LOGGER = logging.getLogger(__name__)
+
+
+def _table(title: str) -> Table:
+    return Table.grid(expand=True, padding=(0, 1), title=title, title_style="bold cyan")
+
+
+class RichDashboard:
+    """Read-only dashboard that refreshes plant metrics using rich."""
+
+    def __init__(
+        self,
+        reactor: Reactor,
+        start_state: Optional[PlantState],
+        timestep: float = 1.0,
+        save_path: Optional[str] = None,
+    ) -> None:
+        self.reactor = reactor
+        self.start_state = start_state
+        self.timestep = timestep
+        self.save_path = save_path
+        self.sim: Optional[ReactorSimulation] = None
+
+    def _render(self, state: PlantState) -> Layout:
+        layout = Layout()
+        layout.split_column(Layout(name="upper"), Layout(name="lower"))
+        layout["upper"].split_row(Layout(name="core"), Layout(name="loops"))
+        layout["lower"].split_row(Layout(name="turbine"), Layout(name="misc"))
+
+        core = _table("Core")
+        core.add_row(f"Power {state.core.power_output_mw:6.1f} MW", f"Fuel {state.core.fuel_temperature:6.1f} K")
+        core.add_row(f"Rods {self.reactor.control.rod_fraction:.3f}", f"Mode {'AUTO' if not self.reactor.control.manual_control else 'MAN'}")
+        core.add_row(f"Setpoint {self.reactor.control.setpoint_mw:5.0f} MW", f"Reactivity {state.core.reactivity_margin:+.4f}")
+        layout["upper"]["core"].update(Panel(core, padding=0))
+
+        loops = _table("Loops")
+        loops.add_row(
+            f"P pri {state.primary_loop.pressure:4.1f}/{constants.MAX_PRESSURE:4.1f} MPa",
+            f"Tin {state.primary_loop.temperature_in:5.1f}K",
+            f"Tout {state.primary_loop.temperature_out:5.1f}K",
+            f"Flow {state.primary_loop.mass_flow_rate:6.0f} kg/s",
+        )
+        loops.add_row(
+            f"P sec {state.secondary_loop.pressure:4.1f}/{constants.MAX_PRESSURE:4.1f} MPa",
+            f"Tin {state.secondary_loop.temperature_in:5.1f}K",
+            f"Tout {state.secondary_loop.temperature_out:5.1f}K",
+            f"q {state.secondary_loop.steam_quality:4.2f}",
+        )
+        loops.add_row(
+            f"HX ΔT {state.primary_to_secondary_delta_t:4.0f}K",
+            f"Eff {state.heat_exchanger_efficiency*100:5.1f}%",
+            f"Relief pri {'OPEN' if self.reactor.primary_relief_open else 'CLOSED'}",
+            f"Relief sec {'OPEN' if self.reactor.secondary_relief_open else 'CLOSED'}",
+        )
+        layout["upper"]["loops"].update(Panel(loops, padding=0))
+
+        turbine = _table("Turbines / Grid")
+        avail = getattr(self, "_steam_available_power", lambda s: 0.0)(state)  # type: ignore[arg-type]
+        steam_h = state.turbines[0].steam_enthalpy if state.turbines else 0.0
+        turbine.add_row(
+            f"Steam avail {avail:5.1f} MW",
+            f"h {steam_h:5.0f} kJ/kg",
+            f"Cond P {state.turbines[0].condenser_pressure:4.2f} MPa" if state.turbines else "Cond P n/a",
+        )
+        turbine.add_row(
+            f"Unit1 {state.turbines[0].electrical_output_mw:5.1f} MW" if state.turbines else "Unit1 n/a",
+            f"Unit2 {state.turbines[1].electrical_output_mw:5.1f} MW" if len(state.turbines) > 1 else "Unit2 n/a",
+            f"Unit3 {state.turbines[2].electrical_output_mw:5.1f} MW" if len(state.turbines) > 2 else "Unit3 n/a",
+        )
+        turbine.add_row(
+            f"Electrical {state.total_electrical_output():5.1f} MW",
+            f"Demand {self._total_load_demand(state):5.1f} MW",
+            f"Supplied {self._total_load_supplied(state):5.1f} MW",
+        )
+        layout["lower"]["turbine"].update(Panel(turbine, padding=0))
+
+        misc_group = []
+        misc_group.append(Text(f"Time {state.time_elapsed:6.1f}s", style="cyan"))
+        misc_group.append(Text(f"Primary pumps: {[p.status for p in state.primary_pumps] if state.primary_pumps else []}"))
+        misc_group.append(Text(f"Secondary pumps: {[p.status for p in state.secondary_pumps] if state.secondary_pumps else []}"))
+        misc_group.append(Text(f"Pressurizer level {self.reactor.pressurizer_level*100:5.1f}% @ {constants.PRIMARY_PRESSURIZER_SETPOINT_MPA:4.1f} MPa"))
+        misc_group.append(Text(f"Reliefs: pri={'OPEN' if self.reactor.primary_relief_open else 'CLOSED'} sec={'OPEN' if self.reactor.secondary_relief_open else 'CLOSED'}"))
+        if self.reactor.health_monitor.failure_log:
+            misc_group.append(Text(f"Failures: {', '.join(self.reactor.health_monitor.failure_log)}", style="bold red"))
+        layout["lower"]["misc"].update(Panel(Group(*misc_group), padding=0))
+        return layout
+
+    def _total_load_supplied(self, state: PlantState) -> float:
+        return sum(t.load_supplied_mw for t in state.turbines)
+
+    def _total_load_demand(self, state: PlantState) -> float:
+        return sum(t.load_demand_mw for t in state.turbines)
+
+    def run(self) -> None:
+        self.sim = ReactorSimulation(
+            self.reactor,
+            timestep=self.timestep,
+            duration=None,
+            realtime=True,
+        )
+        self.sim.start_state = self.start_state
+        try:
+            with Live(console=None, refresh_per_second=4) as live:
+                for state in self.sim.run():
+                    live.update(self._render(state))
+        except KeyboardInterrupt:
+            if self.sim:
+                self.sim.stop()
+        finally:
+            if self.save_path and self.sim and self.sim.last_state:
+                self.reactor.save_state(self.save_path, self.sim.last_state)
+