prox/Reactor-Sim
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add schematic dashboard page and F1/F2 navigation

Browse Source
This commit is contained in:
Codex Agent
2025-11-26 22:33:55 +01:00
parent fae85404a7
commit abc1cb79e1
2 changed files with 90 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files

89
src/reactor_sim/dashboard.py
View File

@@ -13,7 +13,7 @@ from . import constants
from .commands import ReactorCommand
from .reactor import Reactor
from .simulation import ReactorSimulation
from .state import PlantState
from .state import PlantState, PumpState
LOGGER = logging.getLogger(__name__)
@@ -76,6 +76,7 @@ class ReactorDashboard:
self.sim: Optional[ReactorSimulation] = None
self.quit_requested = False
self.reset_requested = False
self.page = 1 # 1=metrics, 2=schematic
self._last_state: Optional[PlantState] = None
self._trend_history: deque[tuple[float, float, float]] = deque(maxlen=120)
self.log_buffer: deque[str] = deque(maxlen=8)
@@ -90,6 +91,7 @@ class ReactorDashboard:
DashboardKey("space", "SCRAM"),
DashboardKey("r", "Reset & clear state"),
DashboardKey("a", "Toggle auto rod control"),
DashboardKey("F1/F2", "Metrics / schematic views"),
DashboardKey("+/-", "Withdraw/insert rods"),
DashboardKey("1-9 / Numpad", "Set rods to 0.1 … 0.9 (manual)"),
DashboardKey("[/]", "Adjust consumer demand −/+50 MW"),
@@ -213,6 +215,10 @@ class ReactorDashboard:
self._queue_command(ReactorCommand(generator_auto=not self.reactor.generator_auto))
elif ch in (ord("t"), ord("T")):
self._queue_command(ReactorCommand(turbine_on=not self.reactor.turbine_active))
elif ch == curses.KEY_F1:
self.page = 1
elif ch == curses.KEY_F2:
self.page = 2
elif keyname and keyname.decode(errors="ignore") in ("!", "@", "#", '"'):
name = keyname.decode(errors="ignore")
turbine_hotkeys = {"!": 0, "@": 1, "#": 2, '"': 1}
@@ -228,7 +234,7 @@ class ReactorDashboard:
self._queue_command(ReactorCommand(rod_position=target, rod_manual=True))
elif ch in _NUMPAD_ROD_KEYS:
self._queue_command(ReactorCommand(rod_position=_NUMPAD_ROD_KEYS[ch], rod_manual=True))
elif curses.KEY_F1 <= ch <= curses.KEY_F9:
elif curses.KEY_F3 <= ch <= curses.KEY_F9:
target = (ch - curses.KEY_F1 + 1) / 10.0
self._queue_command(ReactorCommand(rod_position=target, rod_manual=True))
elif ch in (ord("+"), ord("=")):
@@ -379,7 +385,10 @@ class ReactorDashboard:
help_win = stdscr.derwin(data_height, right_width, 0, left_width + gap)
status_win = stdscr.derwin(status_height, width, data_height, 0)
self._draw_data_panel(data_win, state)
if self.page == 1:
self._draw_data_panel(data_win, state)
else:
self._draw_schematic_panel(data_win, state)
self._draw_help_panel(help_win)
self._draw_status_panel(status_win, state)
stdscr.refresh()
@@ -563,7 +572,7 @@ class ReactorDashboard:
f"Time {state.time_elapsed:7.1f}s | Rods {self.reactor.control.rod_fraction:.3f} | "
f"Primary {'ON' if self.reactor.primary_pump_active else 'OFF'} | "
f"Secondary {'ON' if self.reactor.secondary_pump_active else 'OFF'} | "
f"Turbines {turbine_text}"
f"Turbines {turbine_text} | Page {'Metrics' if self.page == 1 else 'Schematic'}"
)
win.addstr(1, 1, msg, curses.color_pair(3))
if self.reactor.health_monitor.failure_log:
@@ -608,6 +617,78 @@ class ReactorDashboard:
row += 1
return row + 1
def _flow_arrow(self, flow: float) -> str:
if flow > 15000:
return "====>"
if flow > 5000:
return "===>"
if flow > 500:
return "->"
return "--"
def _pump_glyph(self, pump_state: PumpState | None) -> str:
if pump_state is None:
return "·"
status = getattr(pump_state, "status", "OFF")
if status == "RUN":
return "▶"
if status == "CAV":
return "!"
if status == "STARTING":
return ">"
if status == "STOPPING":
return "-"
return "·"
def _draw_schematic_panel(self, win: "curses._CursesWindow", state: PlantState) -> None:
win.erase()
win.box()
try:
win.addstr(0, 2, " Plant Schematic ", curses.color_pair(1) | curses.A_BOLD)
except curses.error:
pass
height, width = win.getmaxyx()
prim = state.primary_loop
sec = state.secondary_loop
p_pumps = state.primary_pumps if state.primary_pumps else []
s_pumps = state.secondary_pumps if state.secondary_pumps else []
p1 = p_pumps[0] if len(p_pumps) > 0 else None
p2 = p_pumps[1] if len(p_pumps) > 1 else None
s1 = s_pumps[0] if len(s_pumps) > 0 else None
s2 = s_pumps[1] if len(s_pumps) > 1 else None
steam_avail = self._steam_available_power(state)
enthalpy = state.turbines[0].steam_enthalpy if state.turbines else 0.0
lines = [
f"CORE {state.core.power_output_mw:5.0f}MW {state.core.fuel_temperature:5.0f}K | Rods {self.reactor.control.rod_fraction:.2f} ({'AUTO' if not self.reactor.control.manual_control else 'MAN'})",
f"Primary Flow: {self._flow_arrow(prim.mass_flow_rate)} {prim.mass_flow_rate:7.0f} kg/s | ΔT hx={state.primary_to_secondary_delta_t:4.0f}K eff={state.heat_exchanger_efficiency*100:5.1f}%",
f"Pumps P1[{self._pump_glyph(p1)}]{(p1.flow_rate if p1 else 0):6.0f}kg/s P2[{self._pump_glyph(p2)}]{(p2.flow_rate if p2 else 0):6.0f}kg/s Relief:{'OPEN' if self.reactor.primary_relief_open else 'CLOSED'}",
f"Ppri={prim.pressure:4.1f}MPa | Tin={prim.temperature_in:6.1f}K Tout={prim.temperature_out:6.1f}K",
"-" * max(10, min(width - 4, 70)),
f"Secondary Flow: {self._flow_arrow(sec.mass_flow_rate)} {sec.mass_flow_rate:7.0f} kg/s | Steam q={sec.steam_quality:4.2f} h={enthalpy:5.0f} kJ/kg avail={steam_avail:5.1f}MW",
f"Pumps S1[{self._pump_glyph(s1)}]{(s1.flow_rate if s1 else 0):6.0f}kg/s S2[{self._pump_glyph(s2)}]{(s2.flow_rate if s2 else 0):6.0f}kg/s Relief:{'OPEN' if self.reactor.secondary_relief_open else 'CLOSED'}",
f"Psec={sec.pressure:4.1f}MPa | Tin={sec.temperature_in:6.1f}K Tout={sec.temperature_out:6.1f}K",
f"Drum Level={sec.level*100:5.1f}% Energy={sec.energy_j/1e6:7.0f} MJ",
]
turbine_bits = []
for idx, t_state in enumerate(state.turbines):
turbine_bits.append(f"T{idx+1}:{t_state.electrical_output_mw:4.0f}MW")
lines.append("Turbines " + " ".join(turbine_bits) if turbine_bits else "Turbines n/a")
consumer_status = "OFF"
demand = 0.0
if self.reactor.consumer:
consumer_status = "ONLINE" if self.reactor.consumer.online else "OFF"
demand = self.reactor.consumer.demand_mw
lines.append(f"Consumer {consumer_status} demand={demand:5.0f}MW supplied={state.total_electrical_output():5.0f}MW")
for row, text in enumerate(lines, start=1):
if row >= height - 1:
break
try:
win.addstr(row, 2, text[: max(1, width - 3)])
except curses.error:
continue
def _turbine_status_lines(self) -> list[str]:
if not self.reactor.turbine_unit_active:
return ["n/a"]