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

Add generator power model and meltdown handling

Browse Source
This commit is contained in:
Codex Agent
2025-11-22 20:04:04 +01:00
parent e7d8a34818
commit b03e80da9f
8 changed files with 396 additions and 66 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
src/reactor_sim/commands.py
View File

@@ -23,6 +23,7 @@ class ReactorCommand:
turbine_units: dict[int, bool] | None = None
primary_pumps: dict[int, bool] | None = None
secondary_pumps: dict[int, bool] | None = None
generator_units: dict[int, bool] | None = None
maintenance_components: tuple[str, ...] = tuple()
@classmethod

7
src/reactor_sim/constants.py
View File

@@ -8,7 +8,8 @@ NEUTRON_LIFETIME = 0.1 # seconds, prompt neutron lifetime surrogate
FUEL_ENERGY_DENSITY = 200.0 * MEGAWATT # J/kg released as heat
COOLANT_HEAT_CAPACITY = 4_200.0 # J/(kg*K) for water/steam
COOLANT_DENSITY = 700.0 # kg/m^3 averaged between phases
MAX_CORE_TEMPERATURE = 1_800.0 # K
CORE_MELTDOWN_TEMPERATURE = 2_873.0 # K (approx 2600C) threshold for irreversible meltdown
MAX_CORE_TEMPERATURE = CORE_MELTDOWN_TEMPERATURE # Allow simulation to approach meltdown temperature
MAX_PRESSURE = 15.0 # MPa typical PWR primary loop limit
CONTROL_ROD_SPEED = 0.03 # fraction insertion per second
CONTROL_ROD_WORTH = 0.042 # delta rho contribution when fully withdrawn
@@ -20,6 +21,10 @@ MEV_TO_J = 1.602_176_634e-13
ELECTRON_FISSION_CROSS_SECTION = 5e-16 # cm^2, tuned for simulation scale
PUMP_SPOOL_TIME = 5.0 # seconds to reach commanded flow
TURBINE_SPOOL_TIME = 12.0 # seconds to reach steady output
GENERATOR_SPOOL_TIME = 10.0 # seconds to reach full output
# Auxiliary power assumptions
PUMP_POWER_MW = 12.0 # MW draw per pump unit
BASE_AUX_LOAD_MW = 5.0 # control, instrumentation, misc.
# Threshold inventories (event counts) for flagging common poisons in diagnostics.
KEY_POISON_THRESHOLDS = {
"Xe": 1e20, # xenon

58
src/reactor_sim/dashboard.py
View File

@@ -42,6 +42,7 @@ class ReactorDashboard:
self.sim: Optional[ReactorSimulation] = None
self.quit_requested = False
self.reset_requested = False
self._last_state: Optional[PlantState] = None
self.log_buffer: deque[str] = deque(maxlen=4)
self._log_handler: Optional[logging.Handler] = None
self._previous_handlers: list[logging.Handler] = []
@@ -49,19 +50,20 @@ class ReactorDashboard:
self.keys = [
DashboardKey("q", "Quit & save"),
DashboardKey("space", "SCRAM"),
DashboardKey("p", "Toggle primary pump"),
DashboardKey("o", "Toggle secondary pump"),
DashboardKey("g", "Toggle primary pump 1"),
DashboardKey("h", "Toggle primary pump 2"),
DashboardKey("j", "Toggle secondary pump 1"),
DashboardKey("k", "Toggle secondary pump 2"),
DashboardKey("b", "Toggle generator 1"),
DashboardKey("v", "Toggle generator 2"),
DashboardKey("t", "Toggle turbine"),
DashboardKey("1/2/3", "Toggle turbine units 1-3"),
DashboardKey("y/u/i", "Maintain turbine 1/2/3"),
DashboardKey("c", "Toggle consumer"),
DashboardKey("r", "Reset & clear state"),
DashboardKey("m", "Maintain primary pump"),
DashboardKey("n", "Maintain secondary pump"),
DashboardKey("m/n", "Maintain primary pumps 1/2"),
DashboardKey(",/.", "Maintain secondary pumps 1/2"),
DashboardKey("B/V", "Maintain generator 1/2"),
DashboardKey("k", "Maintain core (requires shutdown)"),
DashboardKey("+/-", "Withdraw/insert rods"),
DashboardKey("[/]", "Adjust consumer demand −/+50 MW"),
@@ -95,6 +97,7 @@ class ReactorDashboard:
self.sim.start_state = self.start_state
try:
for state in self.sim.run():
self._last_state = state
self._draw(stdscr, state)
self._handle_input(stdscr)
if self.quit_requested or self.reset_requested:
@@ -127,9 +130,10 @@ class ReactorDashboard:
if ch == ord(" "):
self._queue_command(ReactorCommand.scram_all())
elif ch in (ord("p"), ord("P")):
self._queue_command(ReactorCommand(primary_pump_on=not self.reactor.primary_pump_active))
# Deprecated master toggles ignored.
continue
elif ch in (ord("o"), ord("O")):
self._queue_command(ReactorCommand(secondary_pump_on=not self.reactor.secondary_pump_active))
continue
elif ch in (ord("g"), ord("G")):
self._toggle_primary_pump_unit(0)
elif ch in (ord("h"), ord("H")):
@@ -138,6 +142,10 @@ class ReactorDashboard:
self._toggle_secondary_pump_unit(0)
elif ch in (ord("k"), ord("K")):
self._toggle_secondary_pump_unit(1)
elif ch in (ord("b"), ord("B")):
self._toggle_generator_unit(0)
elif ch in (ord("v"), ord("V")):
self._toggle_generator_unit(1)
elif ch in (ord("t"), ord("T")):
self._queue_command(ReactorCommand(turbine_on=not self.reactor.turbine_active))
elif ord("1") <= ch <= ord("9"):
@@ -167,11 +175,19 @@ class ReactorDashboard:
elif ch in (ord("a"), ord("A")):
self._queue_command(ReactorCommand(rod_manual=not self.reactor.control.manual_control))
elif ch in (ord("m"), ord("M")):
self._queue_command(ReactorCommand.maintain("primary_pump"))
self._queue_command(ReactorCommand.maintain("primary_pump_1"))
elif ch in (ord("n"), ord("N")):
self._queue_command(ReactorCommand.maintain("secondary_pump"))
self._queue_command(ReactorCommand.maintain("primary_pump_2"))
elif ch in (ord("k"), ord("K")):
self._queue_command(ReactorCommand.maintain("core"))
elif ch == ord(","):
self._queue_command(ReactorCommand.maintain("secondary_pump_1"))
elif ch == ord("."):
self._queue_command(ReactorCommand.maintain("secondary_pump_2"))
elif ch in (ord("B"),):
self._queue_command(ReactorCommand.maintain("generator_1"))
elif ch in (ord("V"),):
self._queue_command(ReactorCommand.maintain("generator_2"))
elif ch in (ord("y"), ord("Y")):
self._queue_command(ReactorCommand.maintain("turbine_1"))
elif ch in (ord("u"), ord("U")):
@@ -216,6 +232,13 @@ class ReactorDashboard:
current = self.reactor.secondary_pump_units[index]
self._queue_command(ReactorCommand(secondary_pumps={index + 1: not current}))
def _toggle_generator_unit(self, index: int) -> None:
current = False
if self._last_state and index < len(self._last_state.generators):
gen = self._last_state.generators[index]
current = gen.running or gen.starting
self._queue_command(ReactorCommand(generator_units={index + 1: not current}))
def _request_reset(self) -> None:
self.reset_requested = True
if self.sim:
@@ -326,6 +349,12 @@ class ReactorDashboard:
("Steam Quality", f"{state.secondary_loop.steam_quality:5.2f}"),
],
)
y = self._draw_section(
win,
y,
"Generators",
self._generator_lines(state),
)
consumer_status = "n/a"
consumer_demand = 0.0
if self.reactor.consumer:
@@ -368,7 +397,7 @@ class ReactorDashboard:
"Start pumps before withdrawing rods.",
"Bring turbine and consumer online after thermal stabilization.",
"Toggle turbine units (1/2/3) for staggered maintenance.",
"Use m/n/k/y/u/i to request maintenance (stop equipment first).",
"Use m/n/,/. for pump maintenance; B/V for generators.",
"Press 'r' to reset/clear state if you want a cold start.",
"Watch component health to avoid automatic trips.",
]
@@ -471,6 +500,17 @@ class ReactorDashboard:
return [("Active", "None")]
return [(comp, "IN PROGRESS") for comp in sorted(self.reactor.maintenance_active)]
def _generator_lines(self, state: PlantState) -> list[tuple[str, str]]:
if not state.generators:
return [("Status", "n/a")]
lines: list[tuple[str, str]] = []
for idx, gen in enumerate(state.generators):
status = "RUN" if gen.running else "START" if gen.starting else "OFF"
spool = f" spool {gen.spool_remaining:4.1f}s" if gen.starting else ""
lines.append((f"Gen{idx + 1}", f"{status} {gen.power_output_mw:6.1f} MW{spool}"))
lines.append((f" Battery", f"{gen.battery_charge*100:5.1f}%"))
return lines
def _draw_health_bars(self, win: "curses._CursesWindow", start_y: int) -> int:
height, width = win.getmaxyx()
inner_width = width - 4

55
src/reactor_sim/failures.py
View File

@@ -53,8 +53,12 @@ class HealthMonitor:
def __init__(self) -> None:
self.components: Dict[str, ComponentHealth] = {
"core": ComponentHealth("core"),
"primary_pump": ComponentHealth("primary_pump"),
"secondary_pump": ComponentHealth("secondary_pump"),
"primary_pump_1": ComponentHealth("primary_pump_1"),
"primary_pump_2": ComponentHealth("primary_pump_2"),
"secondary_pump_1": ComponentHealth("secondary_pump_1"),
"secondary_pump_2": ComponentHealth("secondary_pump_2"),
"generator_1": ComponentHealth("generator_1"),
"generator_2": ComponentHealth("generator_2"),
}
for idx in range(3):
name = f"turbine_{idx + 1}"
@@ -67,32 +71,48 @@ class HealthMonitor:
def evaluate(
self,
state: PlantState,
primary_active: bool,
secondary_active: bool,
primary_units: Iterable[bool],
secondary_units: Iterable[bool],
turbine_active: Iterable[bool],
generator_states: Iterable,
dt: float,
) -> List[str]:
events: list[str] = []
turbine_flags = list(turbine_active)
core = self.component("core")
core_temp = state.core.fuel_temperature
temp_stress = max(0.0, (core_temp - 900.0) / (constants.MAX_CORE_TEMPERATURE - 900.0))
temp_stress = max(0.0, (core_temp - 900.0) / max(1e-6, (constants.MAX_CORE_TEMPERATURE - 900.0)))
base_degrade = 0.0001 * dt
core.degrade(base_degrade + temp_stress * 0.01 * dt)
if primary_active:
primary_flow = state.primary_loop.mass_flow_rate
flow_ratio = 0.0 if primary_flow <= 0 else min(1.0, primary_flow / 18_000.0)
self.component("primary_pump").degrade((0.0002 + (1 - flow_ratio) * 0.005) * dt)
prim_units = list(primary_units)
sec_units = list(secondary_units)
prim_states = state.primary_pumps or []
sec_states = state.secondary_pumps or []
for idx, active in enumerate(prim_units):
comp = self.component(f"primary_pump_{idx + 1}")
if idx < len(prim_states) and active:
flow = prim_states[idx].flow_rate
flow_ratio = 0.0 if flow <= 0 else min(1.0, flow / 9_000.0)
comp.degrade((0.0002 + (1 - flow_ratio) * 0.005) * dt)
else:
self.component("primary_pump").degrade(0.0)
comp.degrade(0.0)
for idx, active in enumerate(sec_units):
comp = self.component(f"secondary_pump_{idx + 1}")
if idx < len(sec_states) and active:
flow = sec_states[idx].flow_rate
flow_ratio = 0.0 if flow <= 0 else min(1.0, flow / 8_000.0)
comp.degrade((0.0002 + (1 - flow_ratio) * 0.004) * dt)
else:
comp.degrade(0.0)
if secondary_active:
secondary_flow = state.secondary_loop.mass_flow_rate
flow_ratio = 0.0 if secondary_flow <= 0 else min(1.0, secondary_flow / 16_000.0)
self.component("secondary_pump").degrade((0.0002 + (1 - flow_ratio) * 0.004) * dt)
for idx, gen_state in enumerate(generator_states):
comp = self.component(f"generator_{idx + 1}")
running = getattr(gen_state, "running", False) or getattr(gen_state, "starting", False)
if running:
comp.degrade(0.00015 * dt)
else:
self.component("secondary_pump").degrade(0.0)
comp.degrade(0.0)
turbines = state.turbines if hasattr(state, "turbines") else []
for idx, active in enumerate(turbine_flags):
@@ -126,6 +146,11 @@ class HealthMonitor:
mapped = "turbine_1" if name == "turbine" else name
if mapped in self.components:
self.components[mapped] = ComponentHealth.from_snapshot(comp_data)
elif mapped == "primary_pump":
self.components["primary_pump_1"] = ComponentHealth.from_snapshot(comp_data)
self.components["primary_pump_2"] = ComponentHealth.from_snapshot(comp_data)
elif mapped == "secondary_pump":
self.components["secondary_pump_1"] = ComponentHealth.from_snapshot(comp_data)
def maintain(self, component: str, amount: float = 0.05) -> bool:
comp = self.components.get(component)

68
src/reactor_sim/generator.py Normal file
View File

@@ -0,0 +1,68 @@
"""Auxiliary diesel generator model with spool dynamics."""
from __future__ import annotations
from dataclasses import dataclass
import logging
from . import constants
LOGGER = logging.getLogger(__name__)
@dataclass
class GeneratorState:
running: bool
starting: bool
spool_remaining: float
power_output_mw: float
battery_charge: float
@dataclass
class DieselGenerator:
rated_output_mw: float = 50.0
spool_time: float = constants.GENERATOR_SPOOL_TIME
def start(self, state: GeneratorState) -> None:
if state.running or state.starting:
return
if state.battery_charge <= 0.05:
LOGGER.warning("Generator start failed: insufficient battery")
return
state.starting = True
state.spool_remaining = self.spool_time
LOGGER.info("Generator starting (spool %.0fs)", self.spool_time)
def stop(self, state: GeneratorState) -> None:
if not (state.running or state.starting):
return
state.running = False
state.starting = False
state.spool_remaining = 0.0
state.power_output_mw = 0.0
LOGGER.info("Generator stopped")
def step(self, state: GeneratorState, load_demand_mw: float, dt: float) -> float:
"""Advance generator dynamics and return delivered power."""
if state.starting:
state.spool_remaining = max(0.0, state.spool_remaining - dt)
state.power_output_mw = self.rated_output_mw * (1.0 - state.spool_remaining / max(self.spool_time, 1e-6))
if state.spool_remaining <= 0.0:
state.starting = False
state.running = True
LOGGER.info("Generator online at %.1f MW", self.rated_output_mw)
elif state.running:
available = self.rated_output_mw
state.power_output_mw = min(available, load_demand_mw)
else:
state.power_output_mw = 0.0
if state.running:
state.battery_charge = min(1.0, state.battery_charge + 0.02 * dt)
elif state.starting:
state.battery_charge = max(0.0, state.battery_charge - 0.01 * dt)
else:
state.battery_charge = max(0.0, state.battery_charge - 0.001 * dt)
return state.power_output_mw

192
src/reactor_sim/reactor.py
View File

@@ -13,6 +13,7 @@ from .consumer import ElectricalConsumer
from .control import ControlSystem
from .failures import HealthMonitor
from .fuel import FuelAssembly, decay_heat_fraction
from .generator import DieselGenerator, GeneratorState
from .neutronics import NeutronDynamics
from .state import CoolantLoopState, CoreState, PlantState, PumpState, TurbineState
from .thermal import ThermalSolver, heat_transfer
@@ -31,6 +32,7 @@ class Reactor:
thermal: ThermalSolver
steam_generator: SteamGenerator
turbines: list[Turbine]
generators: list[DieselGenerator]
atomic_model: AtomicPhysics
consumer: ElectricalConsumer | None = None
health_monitor: HealthMonitor = field(default_factory=HealthMonitor)
@@ -41,6 +43,7 @@ class Reactor:
turbine_active: bool = True
turbine_unit_active: list[bool] = field(default_factory=lambda: [True, True, True])
shutdown: bool = False
meltdown: bool = False
poison_alerts: set[str] = field(default_factory=set)
maintenance_active: set[str] = field(default_factory=set)
@@ -50,6 +53,8 @@ class Reactor:
if not self.turbine_unit_active or len(self.turbine_unit_active) != len(self.turbines):
self.turbine_unit_active = [True] * len(self.turbines)
self.turbine_active = any(self.turbine_unit_active)
if not self.generators:
self.generators = [DieselGenerator() for _ in range(2)]
if not self.primary_pump_units or len(self.primary_pump_units) != 2:
self.primary_pump_units = [True, True]
if not self.secondary_pump_units or len(self.secondary_pump_units) != 2:
@@ -67,6 +72,7 @@ class Reactor:
thermal=ThermalSolver(),
steam_generator=SteamGenerator(),
turbines=[Turbine() for _ in range(3)],
generators=[DieselGenerator() for _ in range(2)],
atomic_model=atomic_model,
consumer=ElectricalConsumer(name="Grid", demand_mw=800.0, online=False),
health_monitor=HealthMonitor(),
@@ -87,6 +93,7 @@ class Reactor:
self.control.manual_control = True
self.control.rod_fraction = 0.95
self.shutdown = True
self.meltdown = False
self.primary_pump_active = False
self.secondary_pump_active = False
self.turbine_unit_active = [False] * len(self.turbines)
@@ -110,6 +117,10 @@ class Reactor:
)
primary_pumps = [PumpState(active=self.primary_pump_active, flow_rate=0.0, pressure=0.5) for _ in range(2)]
secondary_pumps = [PumpState(active=self.secondary_pump_active, flow_rate=0.0, pressure=0.5) for _ in range(2)]
generator_states = [
GeneratorState(running=False, starting=False, spool_remaining=0.0, power_output_mw=0.0, battery_charge=1.0)
for _ in self.generators
]
turbine_states = [
TurbineState(
steam_enthalpy=2_000.0,
@@ -128,6 +139,7 @@ class Reactor:
turbines=turbine_states,
primary_pumps=primary_pumps,
secondary_pumps=secondary_pumps,
generators=generator_states,
)
def step(self, state: PlantState, dt: float, command: ReactorCommand | None = None) -> None:
@@ -136,6 +148,9 @@ class Reactor:
else:
rod_fraction = self.control.update_rods(state.core, dt)
if state.core.fuel_temperature >= constants.CORE_MELTDOWN_TEMPERATURE and not self.meltdown:
self._trigger_meltdown(state)
overrides = {}
if command:
overrides = self._apply_command(command, state)
@@ -165,13 +180,35 @@ class Reactor:
self._check_poison_alerts(state)
pump_demand = overrides.get("coolant_demand", self.control.coolant_demand(state.primary_loop))
self.primary_pump_active = self.primary_pump_active and any(self.primary_pump_units)
self.secondary_pump_active = self.secondary_pump_active and any(self.secondary_pump_units)
primary_units_active = [
self.primary_pump_active and idx < len(self.primary_pump_units) and self.primary_pump_units[idx]
for idx in range(2)
]
secondary_units_active = [
self.secondary_pump_active and idx < len(self.secondary_pump_units) and self.secondary_pump_units[idx]
for idx in range(2)
]
aux_demand = constants.BASE_AUX_LOAD_MW + constants.PUMP_POWER_MW * (
sum(primary_units_active) + sum(secondary_units_active)
)
turbine_electrical = state.total_electrical_output()
generator_power = self._step_generators(state, aux_demand, turbine_electrical, dt)
aux_available = turbine_electrical + generator_power
power_ratio = 1.0 if aux_demand <= 0 else min(1.0, aux_available / aux_demand)
if aux_demand > 0 and aux_available < 0.5 * aux_demand:
LOGGER.warning("Aux power deficit: available %.1f/%.1f MW", aux_available, aux_demand)
if self.primary_pump_active:
total_flow = 0.0
target_pressure = 12.0 * pump_demand + 2.0
target_pressure = (12.0 * pump_demand + 2.0) * power_ratio
loop_pressure = 0.5
target_flow = self.primary_pump.flow_rate(pump_demand)
target_flow = self.primary_pump.flow_rate(pump_demand) * power_ratio
for idx, pump_state in enumerate(state.primary_pumps):
unit_enabled = idx < len(self.primary_pump_units) and self.primary_pump_units[idx]
unit_enabled = (
self.primary_pump_active and idx < len(self.primary_pump_units) and self.primary_pump_units[idx]
)
desired_flow = target_flow if unit_enabled else 0.0
desired_pressure = target_pressure if unit_enabled else 0.5
pump_state.flow_rate = self._ramp_value(
@@ -180,7 +217,7 @@ class Reactor:
pump_state.pressure = self._ramp_value(
pump_state.pressure, desired_pressure, dt, self.primary_pump.spool_time
)
pump_state.active = unit_enabled or pump_state.flow_rate > 1.0
pump_state.active = (unit_enabled and power_ratio > 0.05) or pump_state.flow_rate > 1.0
total_flow += pump_state.flow_rate
loop_pressure = max(loop_pressure, pump_state.pressure)
state.primary_loop.mass_flow_rate = total_flow
@@ -208,7 +245,9 @@ class Reactor:
loop_pressure = 0.5
target_flow = self.secondary_pump.flow_rate(0.75)
for idx, pump_state in enumerate(state.secondary_pumps):
unit_enabled = idx < len(self.secondary_pump_units) and self.secondary_pump_units[idx]
unit_enabled = (
self.secondary_pump_active and idx < len(self.secondary_pump_units) and self.secondary_pump_units[idx]
)
desired_flow = target_flow if unit_enabled else 0.0
desired_pressure = target_pressure if unit_enabled else 0.5
pump_state.flow_rate = self._ramp_value(
@@ -255,9 +294,10 @@ class Reactor:
failures = self.health_monitor.evaluate(
state,
self.primary_pump_active,
self.secondary_pump_active,
primary_units_active,
secondary_units_active,
self.turbine_unit_active,
state.generators,
dt,
)
for failure in failures:
@@ -346,10 +386,15 @@ class Reactor:
LOGGER.critical("Core failure detected. Initiating SCRAM.")
self.shutdown = True
self.control.scram()
elif component == "primary_pump":
self._set_primary_pump(False)
elif component == "secondary_pump":
self._set_secondary_pump(False)
elif component.startswith("primary_pump"):
idx = self._component_index(component)
self._toggle_primary_pump_unit(idx, False)
elif component.startswith("secondary_pump"):
idx = self._component_index(component)
self._toggle_secondary_pump_unit(idx, False)
elif component.startswith("generator"):
idx = self._component_index(component)
LOGGER.warning("Generator %d failed", idx + 1)
elif component.startswith("turbine"):
idx = self._component_index(component)
self._set_turbine_state(False, index=idx)
@@ -370,16 +415,15 @@ class Reactor:
self.shutdown = self.shutdown or command.rod_position >= 0.95
elif command.rod_step is not None:
overrides["rod_fraction"] = self.control.increment_rods(command.rod_step)
if command.primary_pump_on is not None:
self._set_primary_pump(command.primary_pump_on)
if command.secondary_pump_on is not None:
self._set_secondary_pump(command.secondary_pump_on)
if command.primary_pumps:
for idx, flag in command.primary_pumps.items():
self._toggle_primary_pump_unit(idx - 1, flag)
if command.secondary_pumps:
for idx, flag in command.secondary_pumps.items():
self._toggle_secondary_pump_unit(idx - 1, flag)
if command.generator_units:
for idx, flag in command.generator_units.items():
self._toggle_generator(idx - 1, flag, state)
if command.turbine_on is not None:
self._set_turbine_state(command.turbine_on)
if command.turbine_units:
@@ -418,8 +462,7 @@ class Reactor:
if index < 0 or index >= len(self.primary_pump_units):
LOGGER.warning("Ignoring primary pump index %s", index)
return
if self.primary_pump_units[index] == active:
return
if self.primary_pump_units[index] != active:
self.primary_pump_units[index] = active
LOGGER.info("Primary pump %d %s", index + 1, "enabled" if active else "stopped")
if active:
@@ -431,8 +474,7 @@ class Reactor:
if index < 0 or index >= len(self.secondary_pump_units):
LOGGER.warning("Ignoring secondary pump index %s", index)
return
if self.secondary_pump_units[index] == active:
return
if self.secondary_pump_units[index] != active:
self.secondary_pump_units[index] = active
LOGGER.info("Secondary pump %d %s", index + 1, "enabled" if active else "stopped")
if active:
@@ -440,6 +482,57 @@ class Reactor:
elif not any(self.secondary_pump_units):
self._set_secondary_pump(False)
def _toggle_generator(self, index: int, active: bool, state: PlantState) -> None:
if index < 0 or index >= len(self.generators) or index >= len(state.generators):
LOGGER.warning("Ignoring generator index %s", index)
return
gen_state = state.generators[index]
if active:
self.generators[index].start(gen_state)
else:
self.generators[index].stop(gen_state)
def _trigger_meltdown(self, state: PlantState) -> None:
LOGGER.critical("Core meltdown in progress (%.1f K)", state.core.fuel_temperature)
self.meltdown = True
self.shutdown = True
self.control.scram()
try:
self.health_monitor.component("core").fail()
except KeyError:
pass
self._set_turbine_state(False)
def _step_generators(self, state: PlantState, aux_demand: float, turbine_electric: float, dt: float) -> float:
# Ensure we have generator state objects aligned with hardware.
if not state.generators or len(state.generators) < len(self.generators):
missing = len(self.generators) - len(state.generators)
for _ in range(missing):
state.generators.append(
GeneratorState(running=False, starting=False, spool_remaining=0.0, power_output_mw=0.0, battery_charge=1.0)
)
deficit = max(0.0, aux_demand - turbine_electric)
if deficit > 0.0:
for idx, gen_state in enumerate(state.generators):
if not (gen_state.running or gen_state.starting):
self.generators[idx].start(gen_state)
deficit -= self.generators[idx].rated_output_mw
if deficit <= 0:
break
elif turbine_electric > aux_demand:
for idx, gen_state in enumerate(state.generators):
if gen_state.running and not gen_state.starting:
self.generators[idx].stop(gen_state)
total_power = 0.0
remaining = max(0.0, aux_demand - turbine_electric)
for idx, gen_state in enumerate(state.generators):
load = remaining if remaining > 0 else 0.0
delivered = self.generators[idx].step(gen_state, load, dt)
total_power += delivered
remaining = max(0.0, remaining - delivered)
return total_power
def _set_turbine_state(self, active: bool, index: int | None = None) -> None:
if index is None:
for idx in range(len(self.turbine_unit_active)):
@@ -456,9 +549,11 @@ class Reactor:
def _component_index(self, name: str) -> int:
if name == "turbine":
return 0
parts = name.split("_")
try:
return int(name.split("_")[1]) - 1
except (IndexError, ValueError):
for token in reversed(parts):
return int(token) - 1
except (ValueError, TypeError):
return -1
def _perform_maintenance(self, component: str) -> None:
@@ -495,11 +590,26 @@ class Reactor:
if component == "core" and not self.shutdown:
LOGGER.warning("Cannot maintain core while reactor is running")
return False
if component == "primary_pump" and self.primary_pump_active:
LOGGER.warning("Stop primary pump before maintenance")
if component.startswith("primary_pump_"):
idx = self._component_index(component)
if idx < 0 or idx >= len(self.primary_pump_units):
LOGGER.warning("Unknown primary pump maintenance target %s", component)
return False
if component == "secondary_pump" and self.secondary_pump_active:
LOGGER.warning("Stop secondary pump before maintenance")
if self.primary_pump_units[idx]:
LOGGER.warning("Stop primary pump %d before maintenance", idx + 1)
return False
if component.startswith("secondary_pump_"):
idx = self._component_index(component)
if idx < 0 or idx >= len(self.secondary_pump_units):
LOGGER.warning("Unknown secondary pump maintenance target %s", component)
return False
if self.secondary_pump_units[idx]:
LOGGER.warning("Stop secondary pump %d before maintenance", idx + 1)
return False
if component.startswith("generator_"):
idx = self._component_index(component)
if idx < 0 or idx >= len(self.generators):
LOGGER.warning("Unknown generator maintenance target %s", component)
return False
if component.startswith("turbine"):
idx = self._component_index(component)
@@ -529,7 +639,18 @@ class Reactor:
"turbine_active": self.turbine_active,
"turbine_units": self.turbine_unit_active,
"shutdown": self.shutdown,
"meltdown": self.meltdown,
"maintenance_active": list(self.maintenance_active),
"generators": [
{
"running": g.running,
"starting": g.starting,
"spool_remaining": g.spool_remaining,
"power_output_mw": g.power_output_mw,
"battery_charge": g.battery_charge,
}
for g in state.generators
],
"consumer": {
"online": self.consumer.online if self.consumer else False,
"demand_mw": self.consumer.demand_mw if self.consumer else 0.0,
@@ -549,6 +670,7 @@ class Reactor:
self.turbine_unit_active = list(unit_states)
self.turbine_active = metadata.get("turbine_active", any(self.turbine_unit_active))
self.shutdown = metadata.get("shutdown", self.shutdown)
self.meltdown = metadata.get("meltdown", self.meltdown)
maint = metadata.get("maintenance_active")
if maint is not None:
self.maintenance_active = set(maint)
@@ -594,6 +716,26 @@ class Reactor:
load_supplied_mw=0.0,
)
)
gen_meta = metadata.get("generators", [])
if not plant.generators or len(plant.generators) < len(self.generators):
while len(plant.generators) < len(self.generators):
plant.generators.append(
GeneratorState(
running=False,
starting=False,
spool_remaining=0.0,
power_output_mw=0.0,
battery_charge=1.0,
)
)
for idx, gen_state in enumerate(plant.generators):
if idx < len(gen_meta):
cfg = gen_meta[idx]
gen_state.running = cfg.get("running", gen_state.running)
gen_state.starting = cfg.get("starting", gen_state.starting)
gen_state.spool_remaining = cfg.get("spool_remaining", gen_state.spool_remaining)
gen_state.power_output_mw = cfg.get("power_output_mw", gen_state.power_output_mw)
gen_state.battery_charge = cfg.get("battery_charge", gen_state.battery_charge)
return plant
def _handle_heat_sink_loss(self, state: PlantState) -> None:

7
src/reactor_sim/state.py
View File

@@ -4,6 +4,8 @@ from __future__ import annotations
from dataclasses import dataclass, field, asdict
from .generator import GeneratorState
def clamp(value: float, min_value: float, max_value: float) -> float:
return max(min_value, min(max_value, value))
@@ -69,6 +71,7 @@ class PlantState:
turbines: list[TurbineState]
primary_pumps: list[PumpState] = field(default_factory=list)
secondary_pumps: list[PumpState] = field(default_factory=list)
generators: list[GeneratorState] = field(default_factory=list)
time_elapsed: float = field(default=0.0)
def snapshot(self) -> dict[str, float]:
@@ -84,6 +87,7 @@ class PlantState:
"particles": self.core.emitted_particles,
"primary_pumps": [pump.active for pump in self.primary_pumps],
"secondary_pumps": [pump.active for pump in self.secondary_pumps],
"generators": [gen.running or gen.starting for gen in self.generators],
}
def total_electrical_output(self) -> float:
@@ -105,6 +109,8 @@ class PlantState:
turbines = [TurbineState(**t) for t in turbines_blob]
prim_pumps_blob = data.get("primary_pumps", [])
sec_pumps_blob = data.get("secondary_pumps", [])
generators_blob = data.get("generators", [])
generators = [GeneratorState(**g) for g in generators_blob]
return cls(
core=CoreState(**core_blob, fission_product_inventory=inventory, emitted_particles=particles),
primary_loop=CoolantLoopState(**data["primary_loop"]),
@@ -112,5 +118,6 @@ class PlantState:
turbines=turbines,
primary_pumps=[PumpState(**p) for p in prim_pumps_blob],
secondary_pumps=[PumpState(**p) for p in sec_pumps_blob],
generators=generators,
time_elapsed=data.get("time_elapsed", 0.0),
)

58
tests/test_simulation.py
View File

@@ -42,7 +42,9 @@ def test_health_monitor_flags_core_failure():
reactor = Reactor.default()
state = reactor.initial_state()
state.core.fuel_temperature = constants.MAX_CORE_TEMPERATURE
failures = reactor.health_monitor.evaluate(state, True, True, [True, True, True], dt=200.0)
failures = reactor.health_monitor.evaluate(
state, [True, True], [True, True], [True, True, True], state.generators, dt=200.0
)
assert "core" in failures
reactor._handle_failure("core")
assert reactor.shutdown is True
@@ -50,10 +52,10 @@ def test_health_monitor_flags_core_failure():
def test_maintenance_recovers_component_health():
monitor = HealthMonitor()
pump = monitor.component("secondary_pump")
pump = monitor.component("secondary_pump_1")
pump.integrity = 0.3
pump.fail()
restored = monitor.maintain("secondary_pump", amount=0.5)
restored = monitor.maintain("secondary_pump_1", amount=0.5)
assert restored is True
assert pump.integrity == pytest.approx(0.8)
assert pump.failed is False
@@ -88,19 +90,20 @@ def test_cold_shutdown_stays_subcritical():
def test_toggle_maintenance_progresses_until_restored():
reactor = Reactor.default()
reactor.primary_pump_units = [False, False]
reactor.primary_pump_active = False
pump = reactor.health_monitor.component("primary_pump")
pump = reactor.health_monitor.component("primary_pump_1")
pump.integrity = 0.2
def provider(t: float, _state):
if t == 0:
return ReactorCommand.maintain("primary_pump")
return ReactorCommand.maintain("primary_pump_1")
return None
sim = ReactorSimulation(reactor, timestep=1.0, duration=50.0, command_provider=provider)
sim.log()
assert pump.integrity >= 0.99
assert "primary_pump" not in reactor.maintenance_active
assert "primary_pump_1" not in reactor.maintenance_active
def test_primary_pump_unit_toggle_updates_active_flag():
@@ -134,16 +137,21 @@ def test_secondary_pump_unit_toggle_can_restart_pump():
def test_primary_pumps_spool_up_over_seconds():
reactor = Reactor.default()
state = reactor.initial_state()
reactor.secondary_pump_units = [False, False]
# Enable both pumps and command full flow; spool should take multiple steps.
target_flow = reactor.primary_pump.flow_rate(1.0) * len(reactor.primary_pump_units)
reactor.step(state, dt=1.0, command=ReactorCommand(primary_pump_on=True, coolant_demand=1.0))
reactor.step(
state,
dt=1.0,
command=ReactorCommand(primary_pumps={1: True, 2: True}, generator_units={1: True}, coolant_demand=1.0),
)
first_flow = state.primary_loop.mass_flow_rate
assert 0.0 < first_flow < target_flow
for _ in range(10):
reactor.step(state, dt=1.0, command=ReactorCommand(coolant_demand=1.0))
assert state.primary_loop.mass_flow_rate == pytest.approx(target_flow, rel=0.1)
assert state.primary_loop.mass_flow_rate == pytest.approx(target_flow, rel=0.15)
def test_full_rod_withdrawal_reaches_gigawatt_power():
@@ -178,3 +186,37 @@ def test_partially_inserted_rods_hold_near_three_gw():
assert 2_000.0 < state.core.power_output_mw < 4_000.0
assert 500.0 < state.core.fuel_temperature < 800.0
def test_generator_spools_and_powers_pumps():
reactor = Reactor.default()
state = reactor.initial_state()
reactor.shutdown = False
reactor.control.manual_control = True
reactor.control.rod_fraction = 0.95 # keep power low; focus on aux power
reactor.turbine_unit_active = [False, False, False]
reactor.secondary_pump_units = [False, False]
for step in range(12):
cmd = ReactorCommand(generator_units={1: True}, primary_pumps={1: True}) if step == 0 else None
reactor.step(state, dt=1.0, command=cmd)
assert state.generators and state.generators[0].running is True
assert state.generators[0].power_output_mw > 0.0
assert state.primary_loop.mass_flow_rate > 0.0
def test_meltdown_triggers_shutdown():
reactor = Reactor.default()
state = reactor.initial_state()
reactor.shutdown = False
reactor.control.manual_control = True
reactor.control.rod_fraction = 0.0
reactor.primary_pump_active = True
reactor.secondary_pump_active = True
state.core.fuel_temperature = constants.CORE_MELTDOWN_TEMPERATURE + 50.0
reactor.step(state, dt=1.0)
assert reactor.shutdown is True
assert reactor.meltdown is True