Refine Textual dashboard to mirror curses layout
This commit is contained in:
Codex Agent
@@ -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()
|
||||
|
||||
|
||||
Reference in New Issue
Block a user