From d626007faeb3ef6688bb81067ffdf80894d234f5 Mon Sep 17 00:00:00 2001 From: Codex Agent Date: Thu, 27 Nov 2025 13:27:56 +0100 Subject: [PATCH] Add shared snapshot helper and simulation snapshot mode --- src/reactor_sim/simulation.py | 17 +++++++-- src/reactor_sim/snapshot.py | 53 ++++++++++++++++++++++++++++ src/reactor_sim/textual_dashboard.py | 27 ++------------ 3 files changed, 70 insertions(+), 27 deletions(-) create mode 100644 src/reactor_sim/snapshot.py diff --git a/src/reactor_sim/simulation.py b/src/reactor_sim/simulation.py index 9528365..3e45d6f 100644 --- a/src/reactor_sim/simulation.py +++ b/src/reactor_sim/simulation.py @@ -96,6 +96,9 @@ def main() -> None: configure_logging(log_level, log_file) realtime = os.getenv("FISSION_REALTIME", "0") == "1" alternate_dashboard = os.getenv("ALTERNATE_DASHBOARD", "0") == "1" + snapshot_at_env = os.getenv("FISSION_SNAPSHOT_AT") + snapshot_at = float(snapshot_at_env) if snapshot_at_env else None + snapshot_path = os.getenv("FISSION_SNAPSHOT_PATH", "artifacts/snapshot.txt") duration_env = os.getenv("FISSION_SIM_DURATION") if duration_env: duration = None if duration_env.lower() in {"none", "infinite"} else float(duration_env) @@ -117,7 +120,7 @@ def main() -> None: save_path = os.getenv("FISSION_SAVE_STATE") or str(state_path) if load_path: sim.start_state = reactor.load_state(load_path) - if dashboard_mode: + if dashboard_mode and snapshot_at is None: if alternate_dashboard: try: from .textual_dashboard import run_textual_dashboard @@ -143,7 +146,17 @@ def main() -> None: dashboard.run() return try: - if realtime: + if snapshot_at is not None: + sim.duration = snapshot_at + LOGGER.info("Running headless to t=%.1fs for snapshot...", snapshot_at) + for _ in sim.run(): + pass + if sim.last_state: + from .snapshot import write_snapshot + + write_snapshot(snapshot_path, reactor, sim.last_state) + LOGGER.info("Snapshot written to %s", snapshot_path) + elif realtime: LOGGER.info("Running in real-time mode (Ctrl+C to stop)...") for _ in sim.run(): pass diff --git a/src/reactor_sim/snapshot.py b/src/reactor_sim/snapshot.py new file mode 100644 index 0000000..518c737 --- /dev/null +++ b/src/reactor_sim/snapshot.py @@ -0,0 +1,53 @@ +"""Snapshot formatting helpers shared across dashboards.""" + +from __future__ import annotations + +from pathlib import Path +from typing import Iterable + +from . import constants +from .reactor import Reactor +from .state import PlantState + + +def snapshot_lines(reactor: Reactor, state: PlantState) -> list[str]: + core = state.core + prim = state.primary_loop + sec = state.secondary_loop + t0 = state.turbines[0] if state.turbines else None + lines: list[str] = [ + f"Time {state.time_elapsed:6.1f}s", + f"Core: {core.power_output_mw:6.1f}MW fuel {core.fuel_temperature:6.1f}K rods {reactor.control.rod_fraction:.3f} ({'AUTO' if not 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={state.primary_to_secondary_delta_t:4.0f}K Eff={state.heat_exchanger_efficiency*100:5.1f}%", + ] + if t0: + lines.append( + f"Turbines: h={t0.steam_enthalpy:5.0f}kJ/kg avail={_steam_available_power(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(state.turbines)])) + failures = ", ".join(reactor.health_monitor.failure_log) if reactor.health_monitor.failure_log else "None" + lines.append( + f"Status: pumps pri {[p.status for p in state.primary_pumps]} sec {[p.status for p in state.secondary_pumps]} " + f"relief pri={'OPEN' if reactor.primary_relief_open else 'CLOSED'} sec={'OPEN' if reactor.secondary_relief_open else 'CLOSED'} " + f"failures={failures}" + ) + return lines + + +def write_snapshot(path: Path | str, reactor: Reactor, state: PlantState) -> None: + p = Path(path) + p.parent.mkdir(parents=True, exist_ok=True) + p.write_text("\n".join(snapshot_lines(reactor, state))) + + +def _steam_available_power(state: PlantState) -> float: + mass_flow = state.secondary_loop.mass_flow_rate * max(0.0, state.secondary_loop.steam_quality) + if mass_flow <= 1.0: + return 0.0 + enthalpy = state.turbines[0].steam_enthalpy if state.turbines else (constants.STEAM_LATENT_HEAT / 1_000.0) + return (enthalpy * mass_flow) / 1_000.0 + diff --git a/src/reactor_sim/textual_dashboard.py b/src/reactor_sim/textual_dashboard.py index cc71815..abb829e 100644 --- a/src/reactor_sim/textual_dashboard.py +++ b/src/reactor_sim/textual_dashboard.py @@ -18,6 +18,7 @@ from .reactor import Reactor from .simulation import ReactorSimulation from .state import PlantState from .commands import ReactorCommand +from .snapshot import snapshot_lines LOGGER = logging.getLogger(__name__) @@ -431,31 +432,7 @@ class TextualDashboard(App): 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 + return snapshot_lines(self.reactor, self.state) def _save_snapshot(self, auto: bool = False) -> None: try: