diff --git a/src/reactor_sim/textual_dashboard.py b/src/reactor_sim/textual_dashboard.py index 4722451..0f2ef4d 100644 --- a/src/reactor_sim/textual_dashboard.py +++ b/src/reactor_sim/textual_dashboard.py @@ -1,15 +1,16 @@ -"""Interactive Textual-based dashboard (optional).""" +"""Interactive Textual-based dashboard mirroring the curses layout.""" from __future__ import annotations import logging +import os from collections import deque +from pathlib import Path from typing import Optional from textual.app import App, ComposeResult +from textual.containers import Horizontal, VerticalScroll from textual.widgets import Static, Header, Footer -from textual.containers import Horizontal, Vertical -from textual.reactive import reactive from textual.timer import Timer from . import constants @@ -21,13 +22,18 @@ from .commands import ReactorCommand LOGGER = logging.getLogger(__name__) +def _bar(label: str, value: float, width: int = 24) -> str: + filled = int(max(0.0, min(1.0, value)) * width) + return f"{label:<14} [{'#'*filled}{'-'*(width-filled)}] {value*100:5.1f}%" + + class TextualDashboard(App): - """A lightweight Textual UI with keybindings mapped to reactor commands.""" + """Textual dashboard with controls and sections similar to the curses view.""" CSS = """ Screen { layout: vertical; } - .row { width: 1fr; } - .panel { padding: 0 1; } + .col { width: 1fr; } + .panel { padding: 0 1; border: solid gray; } """ BINDINGS = [ @@ -53,6 +59,7 @@ class TextualDashboard(App): (";", "toggle_secondary_relief", "Relief sec"), ("c", "toggle_consumer", "Consumer"), ("a", "toggle_auto_rods", "Auto rods"), + ("f12", "snapshot", "Snapshot"), ] timestep: float = 1.0 @@ -69,23 +76,48 @@ class TextualDashboard(App): self.state = start_state or self.reactor.initial_state() self.timestep = timestep self.save_path = save_path - self._sim: Optional[ReactorSimulation] = None self._pending: deque[ReactorCommand] = deque() self._timer: Optional[Timer] = None + self._trend_history: deque[tuple[float, float, float]] = deque(maxlen=120) + 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 + self.snapshot_path = Path(os.getenv("FISSION_SNAPSHOT_PATH", "artifacts/textual_snapshot.txt")) + # Panels self.core_panel = Static(classes="panel") - self.loop_panel = Static(classes="panel") + self.trend_panel = Static(classes="panel") + self.poison_panel = Static(classes="panel") + self.primary_panel = Static(classes="panel") + self.secondary_panel = Static(classes="panel") self.turbine_panel = Static(classes="panel") + self.generator_panel = Static(classes="panel") + self.power_panel = Static(classes="panel") + self.hx_panel = Static(classes="panel") + self.protection_panel = Static(classes="panel") + self.maintenance_panel = Static(classes="panel") + self.health_panel = Static(classes="panel") + self.help_panel = Static(classes="panel") self.status_panel = Static(classes="panel") def compose(self) -> ComposeResult: yield Header() - with Vertical(): - with Horizontal(classes="row"): + with Horizontal(): + with VerticalScroll(classes="col"): yield self.core_panel - yield self.loop_panel - with Horizontal(classes="row"): + yield self.trend_panel + yield self.poison_panel + yield self.primary_panel + yield self.secondary_panel + with VerticalScroll(classes="col"): yield self.turbine_panel + yield self.generator_panel + yield self.power_panel + yield self.hx_panel + yield self.protection_panel + yield self.maintenance_panel + yield self.health_panel + yield self.help_panel yield self.status_panel yield Footer() @@ -166,10 +198,12 @@ class TextualDashboard(App): def action_toggle_auto_rods(self) -> None: self._enqueue(ReactorCommand(rod_manual=not self.reactor.control.manual_control)) + def action_snapshot(self) -> None: + self._save_snapshot(auto=False) + def _merge_commands(self) -> Optional[ReactorCommand]: if not self._pending: return None - # Take the last command to approximate latest intent. cmd = ReactorCommand() while self._pending: nxt = self._pending.popleft() @@ -184,46 +218,189 @@ class TextualDashboard(App): cmd = self._merge_commands() self.reactor.step(self.state, self.timestep, cmd) self._render_panels() + if self.snapshot_at is not None and not self.snapshot_done and self.state.time_elapsed >= self.snapshot_at: + self._save_snapshot(auto=True) def _render_panels(self) -> None: - core = self.state.core - prim = self.state.primary_loop - sec = self.state.secondary_loop - t0 = self.state.turbines[0] if self.state.turbines else None + self._trend_history.append((self.state.time_elapsed, self.state.core.fuel_temperature, self.state.core.power_output_mw)) + self.core_panel.update( f"[bold cyan]Core[/bold cyan]\\n" - f"Power {core.power_output_mw:6.1f} MW Fuel {core.fuel_temperature:6.1f} K\\n" - f"Rods {self.reactor.control.rod_fraction:.3f} ({'AUTO' if not self.reactor.control.manual_control else 'MAN'})\\n" - f"Setpoint {self.reactor.control.setpoint_mw:5.0f} MW Reactivity {core.reactivity_margin:+.4f}\\n" - f"DNB {core.dnb_margin:4.2f} Subcool {core.subcooling_margin:4.1f}K" + f"Power {self.state.core.power_output_mw:6.1f} MW (Nom {constants.NORMAL_CORE_POWER_MW:4.0f}/Max {constants.TEST_MAX_POWER_MW:4.0f})\\n" + f"Fuel {self.state.core.fuel_temperature:6.1f} K Rods {self.reactor.control.rod_fraction:.3f} ({'AUTO' if not self.reactor.control.manual_control else 'MAN'})\\n" + f"Setpoint {self.reactor.control.setpoint_mw:5.0f} MW Reactivity {self.state.core.reactivity_margin:+.4f}\\n" + f"DNB {self.state.core.dnb_margin:4.2f} Subcool {self.state.core.subcooling_margin:4.1f}K" ) - self.loop_panel.update( - f"[bold cyan]Loops[/bold cyan]\\n" - f"Primary P {prim.pressure:4.1f}/{constants.MAX_PRESSURE:4.1f} MPa Tin {prim.temperature_in:6.1f}K Tout {prim.temperature_out:6.1f}K\\n" - f"Flow {prim.mass_flow_rate:6.0f} kg/s Pressurizer {self.reactor.pressurizer_level*100:5.1f}%\\n" - f"Secondary P {sec.pressure:4.1f}/{constants.MAX_PRESSURE:4.1f} MPa Tin {sec.temperature_in:6.1f}K Tout {sec.temperature_out:6.1f}K q {sec.steam_quality:4.2f}\\n" - f"HX ΔT {self.state.primary_to_secondary_delta_t:4.0f}K Eff {self.state.heat_exchanger_efficiency*100:5.1f}%" + self.trend_panel.update(self._trend_text()) + self.poison_panel.update(self._poison_text()) + self.primary_panel.update( + f"[bold cyan]Primary Loop[/bold cyan]\\n" + f"Flow {self.state.primary_loop.mass_flow_rate:7.0f}/{self.reactor.primary_pump.nominal_flow * len(self.reactor.primary_pump_units):.0f} kg/s\\n" + f"Level {self.state.primary_loop.level*100:6.1f}%\\n" + f"Tin {self.state.primary_loop.temperature_in:7.1f} K Tout {self.state.primary_loop.temperature_out:7.1f} K (Target {constants.PRIMARY_OUTLET_TARGET_K:4.0f})\\n" + f"P {self.state.primary_loop.pressure:5.2f}/{constants.MAX_PRESSURE:4.1f} MPa Pressurizer {self.reactor.pressurizer_level*100:6.1f}% @ {constants.PRIMARY_PRESSURIZER_SETPOINT_MPA:4.1f} MPa\\n" + f"Relief {'OPEN' if self.reactor.primary_relief_open else 'CLOSED'} Pumps {[p.status for p in self.state.primary_pumps]}" ) - if t0: - condenser = f"P={t0.condenser_pressure:4.2f}/{constants.CONDENSER_MAX_PRESSURE_MPA:4.2f}MPa T={t0.condenser_temperature:6.1f}K" - else: - condenser = "n/a" - turbine_lines = [ - f"[bold cyan]Turbines/Grid[/bold cyan]", - f"Steam avail {getattr(self, '_steam_available_power', lambda s:0.0)(self.state):5.1f} MW h={t0.steam_enthalpy if t0 else 0:5.0f} kJ/kg Cond {condenser}", - " ".join([f"T{idx+1}:{t.electrical_output_mw:5.1f}MW" for idx, t in enumerate(self.state.turbines)]) if self.state.turbines else "Turbines n/a", - f"Electrical {self.state.total_electrical_output():5.1f} MW Load {self._total_load_supplied(self.state):5.1f}/{self._total_load_demand(self.state):5.1f} MW", - ] - self.turbine_panel.update("\\n".join(turbine_lines)) + self.secondary_panel.update( + f"[bold cyan]Secondary Loop[/bold cyan]\\n" + f"Flow {self.state.secondary_loop.mass_flow_rate:7.0f}/{self.reactor.secondary_pump.nominal_flow * len(self.reactor.secondary_pump_units):.0f} kg/s\\n" + f"Level {self.state.secondary_loop.level*100:6.1f}%\\n" + f"Tin {self.state.secondary_loop.temperature_in:7.1f} K Tout {self.state.secondary_loop.temperature_out:7.1f} K (Target {constants.SECONDARY_OUTLET_TARGET_K:4.0f})\\n" + f"P {self.state.secondary_loop.pressure:5.2f}/{constants.MAX_PRESSURE:4.1f} MPa q {self.state.secondary_loop.steam_quality:5.2f}/1.00\\n" + f"Relief {'OPEN' if self.reactor.secondary_relief_open else 'CLOSED'} Pumps {[p.status for p in self.state.secondary_pumps]}" + ) + self.turbine_panel.update(self._turbine_text()) + self.generator_panel.update(self._generator_text()) + self.power_panel.update(self._power_text()) + self.hx_panel.update( + f"[bold cyan]Heat Exchanger[/bold cyan]\\n" + f"ΔT (pri-sec) {self.state.primary_to_secondary_delta_t:4.0f} K\\n" + f"Efficiency {self.state.heat_exchanger_efficiency*100:5.1f}%" + ) + self.protection_panel.update(self._protection_text()) + self.maintenance_panel.update(self._maintenance_text()) + self.health_panel.update(self._health_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( f"[bold cyan]Status[/bold cyan]\\n" f"Time {self.state.time_elapsed:6.1f}s\\n" - f"Pumps pri {[p.status for p in self.state.primary_pumps]} sec {[p.status for p in self.state.secondary_pumps]}\\n" - f"Reliefs pri={'OPEN' if self.reactor.primary_relief_open else 'CLOSED'} sec={'OPEN' if self.reactor.secondary_relief_open else 'CLOSED'}\\n" + f"Consumer {'ON' if (self.reactor.consumer and self.reactor.consumer.online) else 'OFF'} Demand {self.reactor.consumer.demand_mw if self.reactor.consumer else 0.0:5.1f} MW\\n" f"Failures: {failures}" ) + def _trend_text(self) -> str: + if len(self._trend_history) < 2: + return "[bold cyan]Trends[/bold cyan]\\nFuel Temp Δ n/a\\nCore Power Δ n/a" + start_t, start_temp, start_power = self._trend_history[0] + end_t, end_temp, end_power = self._trend_history[-1] + duration = max(1.0, end_t - start_t) + temp_delta = end_temp - start_temp + power_delta = end_power - start_power + temp_rate = temp_delta / duration + power_rate = power_delta / duration + return ( + "[bold cyan]Trends[/bold cyan]\\n" + f"Fuel Temp Δ {end_temp:7.1f} K (Δ{temp_delta:+6.1f} / {duration:4.0f}s, {temp_rate:+5.2f}/s)\\n" + f"Core Power Δ {end_power:7.1f} MW (Δ{power_delta:+6.1f} / {duration:4.0f}s, {power_rate:+5.2f}/s)" + ) + + def _poison_text(self) -> str: + inventory = self.state.core.fission_product_inventory or {} + particles = self.state.core.emitted_particles or {} + xe = getattr(self.state.core, "xenon_inventory", 0.0) + sm = inventory.get("Sm", 0.0) + iodine = inventory.get("I", 0.0) + xe_drho = getattr(self.state.core, "reactivity_margin", 0.0) + return ( + "[bold cyan]Key Poisons / Emitters[/bold cyan]\\n" + f"Xe (xenon): {xe:9.2e}\\n" + f"Sm (samarium): {sm:9.2e}\\n" + f"I (iodine): {iodine:9.2e}\\n" + f"Xe Δρ: {xe_drho:+.4f}\\n" + f"Neutrons (src): {particles.get('neutrons', 0.0):.2e}" + ) + + def _turbine_text(self) -> str: + steam_avail = self._steam_available_power(self.state) + enthalpy = self.state.turbines[0].steam_enthalpy if self.state.turbines else 0.0 + cond = "" + if self.state.turbines: + cond = ( + f"Cond P {self.state.turbines[0].condenser_pressure:4.2f}/{constants.CONDENSER_MAX_PRESSURE_MPA:4.2f} MPa " + f"T {self.state.turbines[0].condenser_temperature:6.1f}K" + ) + lines = [ + "[bold cyan]Turbine / Grid[/bold cyan]", + f"Turbines {' '.join(self._turbine_status_lines()) if self._turbine_status_lines() else 'n/a'}", + f"Rated Elec {len(self.reactor.turbines)*self.reactor.turbines[0].rated_output_mw:7.1f} MW" if self.reactor.turbines else "Rated Elec n/a", + f"Steam Avail {steam_avail:5.1f} MW h={enthalpy:5.0f} kJ/kg {cond}", + ] + if self.state.turbines: + lines.append("Unit Elec " + " ".join([f"{t.electrical_output_mw:6.1f}MW" for t in self.state.turbines])) + lines.append(f"Throttle {self.reactor.turbines[0].throttle:5.2f}" if self.reactor.turbines else "Throttle n/a") + lines.append(f"Electrical {self.state.total_electrical_output():7.1f} MW Load {self._total_load_supplied(self.state):7.1f}/{self._total_load_demand(self.state):7.1f} MW") + if self.reactor.consumer: + lines.append(f"Consumer {'ONLINE' if self.reactor.consumer.online else 'OFF'} Demand {self.reactor.consumer.demand_mw:7.1f} MW") + return "\\n".join(lines) + + def _generator_text(self) -> str: + lines = ["[bold cyan]Generators[/bold cyan]"] + for idx, g in enumerate(self.state.generators): + status = "RUN" if g.running else "OFF" + if g.starting: + status = "START" + lines.append(f"Gen{idx+1}: {status:5} {g.power_output_mw:5.1f} MW batt {g.battery_charge*100:5.1f}%") + return "\\n".join(lines) + + def _power_text(self) -> str: + draws = getattr(self.state, "aux_draws", {}) or {} + base = draws.get("base", 0.0) + prim = draws.get("primary_pumps", 0.0) + sec = draws.get("secondary_pumps", 0.0) + demand = draws.get("total_demand", 0.0) + supplied = draws.get("supplied", 0.0) + gen_out = draws.get("generator_output", 0.0) + turb_out = draws.get("turbine_output", 0.0) + return ( + "[bold cyan]Power Stats[/bold cyan]\\n" + f"Base Aux {base:5.1f} MW Prim Aux {prim:5.1f} MW Sec Aux {sec:5.1f} MW\\n" + f"Aux demand {demand:5.1f} MW supplied {supplied:5.1f} MW\\n" + f"Gen out {gen_out:5.1f} MW Turbine out {turb_out:5.1f} MW" + ) + + def _protection_text(self) -> str: + lines = ["[bold cyan]Protections / Warnings[/bold cyan]"] + lines.append(f"SCRAM {'ACTIVE' if self.reactor.shutdown else 'CLEAR'}") + if self.reactor.meltdown: + lines.append("Meltdown IN PROGRESS") + sec_flow_low = self.state.secondary_loop.mass_flow_rate <= 1.0 or not self.reactor.secondary_pump_active + heat_sink_risk = sec_flow_low and self.state.core.power_output_mw > 50.0 + if heat_sink_risk: + heat_text = "TRIPPED low secondary flow >50 MW" + elif sec_flow_low: + heat_text = "ARMED (secondary off/low flow)" + else: + heat_text = "OK" + lines.append(f"Heat sink {heat_text}") + lines.append(f"DNB margin {self.state.core.dnb_margin:4.2f}") + lines.append(f"Subcooling {self.state.core.subcooling_margin:5.1f} K") + lines.append(f"Reliefs pri={'OPEN' if self.reactor.primary_relief_open else 'CLOSED'} sec={'OPEN' if self.reactor.secondary_relief_open else 'CLOSED'}") + return "\\n".join(lines) + + def _maintenance_text(self) -> str: + active = list(self.reactor.maintenance_active) + return "[bold cyan]Maintenance[/bold cyan]\\nActive: " + (", ".join(active) if active else "None") + + def _health_text(self) -> str: + lines = ["[bold cyan]Component Health[/bold cyan]"] + for name, comp in self.reactor.health_monitor.components.items(): + lines.append(_bar(name, comp.integrity)) + return "\\n".join(lines) + + def _help_text(self) -> str: + tips = [ + "Start pumps before withdrawing rods.", + "Bring turbine/consumer online after thermal stabilization.", + "Toggle turbine units 1/2/3 individually.", + "Use m/n/,/. in curses; mapped to j/k etc here.", + "F12 saves a snapshot; set FISSION_SNAPSHOT_AT for auto.", + ] + return "[bold cyan]Controls & Tips[/bold cyan]\\n" + "\\n".join(f"- {t}" for t in tips) + + def _turbine_status_lines(self) -> list[str]: + if not self.reactor.turbine_unit_active: + return [] + lines: list[str] = [] + for idx, active in enumerate(self.reactor.turbine_unit_active): + label = f"{idx + 1}:" + status = "ON" if active else "OFF" + if idx < len(getattr(self.state, "turbines", [])): + t_state = self.state.turbines[idx] + status = getattr(t_state, "status", status) + lines.append(f"{label}{status}") + return lines + def _total_load_supplied(self, state: PlantState) -> float: return sum(t.load_supplied_mw for t in state.turbines) @@ -237,7 +414,44 @@ class TextualDashboard(App): enthalpy = state.turbines[0].steam_enthalpy if state.turbines else (constants.STEAM_LATENT_HEAT / 1_000.0) return (enthalpy * mass_flow) / 1_000.0 + def _snapshot_lines(self) -> list[str]: + core = self.state.core + prim = self.state.primary_loop + sec = self.state.secondary_loop + t0 = self.state.turbines[0] if self.state.turbines else None + lines = [ + f"Time {self.state.time_elapsed:6.1f}s", + f"Core: {core.power_output_mw:6.1f}MW fuel {core.fuel_temperature:6.1f}K rods {self.reactor.control.rod_fraction:.3f} ({'AUTO' if not self.reactor.control.manual_control else 'MAN'})", + f"Primary: P={prim.pressure:4.1f}/{constants.MAX_PRESSURE:4.1f}MPa Tin={prim.temperature_in:6.1f}K Tout={prim.temperature_out:6.1f}K Flow={prim.mass_flow_rate:6.0f}kg/s", + f"Secondary: P={sec.pressure:4.1f}/{constants.MAX_PRESSURE:4.1f}MPa Tin={sec.temperature_in:6.1f}K Tout={sec.temperature_out:6.1f}K q={sec.steam_quality:4.2f} Flow={sec.mass_flow_rate:6.0f}kg/s", + f"HX ΔT={self.state.primary_to_secondary_delta_t:4.0f}K Eff={self.state.heat_exchanger_efficiency*100:5.1f}%", + ] + if t0: + lines.append( + f"Turbines: h={t0.steam_enthalpy:5.0f}kJ/kg avail={self._steam_available_power(self.state):5.1f}MW " + f"CondP={t0.condenser_pressure:4.2f}/{constants.CONDENSER_MAX_PRESSURE_MPA:4.2f}MPa " + f"CondT={t0.condenser_temperature:6.1f}K" + ) + lines.append("Outputs: " + " ".join([f"T{idx+1}:{t.electrical_output_mw:5.1f}MW" for idx, t in enumerate(self.state.turbines)])) + failures = ", ".join(self.reactor.health_monitor.failure_log) if self.reactor.health_monitor.failure_log else "None" + lines.append( + f"Status: pumps pri {[p.status for p in self.state.primary_pumps]} sec {[p.status for p in self.state.secondary_pumps]} " + f"relief pri={'OPEN' if self.reactor.primary_relief_open else 'CLOSED'} sec={'OPEN' if self.reactor.secondary_relief_open else 'CLOSED'} " + f"failures={failures}" + ) + return lines + + def _save_snapshot(self, auto: bool = False) -> None: + try: + self.snapshot_path.parent.mkdir(parents=True, exist_ok=True) + self.snapshot_path.write_text("\\n".join(self._snapshot_lines())) + self.snapshot_done = True + LOGGER.info("Saved dashboard snapshot to %s%s", self.snapshot_path, " (auto)" if auto else "") + except Exception as exc: # pragma: no cover + LOGGER.error("Failed to save snapshot: %s", exc) + 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() +