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

Improve persistence and reactor dynamics

Browse Source
This commit is contained in:
Andrii Prokhorov
2025-11-21 18:59:11 +02:00
parent 7c8321e3c4
commit d37620ccc1
11 changed files with 340 additions and 97 deletions
Download Patch File Download Diff File Expand all files Collapse all files

166
src/reactor_sim/reactor.py
View File

@@ -30,15 +30,23 @@ class Reactor:
secondary_pump: Pump
thermal: ThermalSolver
steam_generator: SteamGenerator
turbine: Turbine
turbines: list[Turbine]
atomic_model: AtomicPhysics
consumer: ElectricalConsumer | None = None
health_monitor: HealthMonitor = field(default_factory=HealthMonitor)
primary_pump_active: bool = True
secondary_pump_active: bool = True
turbine_active: bool = True
turbine_unit_active: list[bool] = field(default_factory=lambda: [True, True, True])
shutdown: bool = False
def __post_init__(self) -> None:
if not self.turbines:
self.turbines = [Turbine()]
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)
@classmethod
def default(cls) -> "Reactor":
atomic_model = AtomicPhysics()
@@ -50,7 +58,7 @@ class Reactor:
secondary_pump=Pump(nominal_flow=16_000.0, efficiency=0.85),
thermal=ThermalSolver(),
steam_generator=SteamGenerator(),
turbine=Turbine(),
turbines=[Turbine() for _ in range(3)],
atomic_model=atomic_model,
consumer=ElectricalConsumer(name="Grid", demand_mw=800.0, online=False),
health_monitor=HealthMonitor(),
@@ -79,15 +87,18 @@ class Reactor:
mass_flow_rate=0.0,
steam_quality=0.0,
)
turbine = TurbineState(
steam_enthalpy=2_000.0,
shaft_power_mw=0.0,
electrical_output_mw=0.0,
condenser_temperature=ambient,
load_demand_mw=0.0,
load_supplied_mw=0.0,
)
return PlantState(core=core, primary_loop=primary, secondary_loop=secondary, turbine=turbine)
turbine_states = [
TurbineState(
steam_enthalpy=2_000.0,
shaft_power_mw=0.0,
electrical_output_mw=0.0,
condenser_temperature=ambient,
load_demand_mw=0.0,
load_supplied_mw=0.0,
)
for _ in self.turbines
]
return PlantState(core=core, primary_loop=primary, secondary_loop=secondary, turbines=turbine_states)
def step(self, state: PlantState, dt: float, command: ReactorCommand | None = None) -> None:
if self.shutdown:
@@ -126,19 +137,13 @@ class Reactor:
transferred = heat_transfer(state.primary_loop, state.secondary_loop, total_power)
self.thermal.step_secondary(state.secondary_loop, transferred)
if self.turbine_active:
self.turbine.step(state.secondary_loop, state.turbine, self.consumer, steam_power_mw=transferred)
else:
state.turbine.shaft_power_mw = 0.0
state.turbine.electrical_output_mw = 0.0
if self.consumer:
self.consumer.update_power_received(0.0)
self._step_turbine_bank(state, transferred)
failures = self.health_monitor.evaluate(
state,
self.primary_pump_active,
self.secondary_pump_active,
self.turbine_active,
self.turbine_unit_active,
dt,
)
for failure in failures:
@@ -157,11 +162,54 @@ class Reactor:
prompt_power,
fission_rate,
state.primary_loop.temperature_out,
state.turbine.electrical_output_mw,
state.turbine.load_supplied_mw,
state.turbine.load_demand_mw,
state.total_electrical_output(),
sum(t.load_supplied_mw for t in state.turbines),
sum(t.load_demand_mw for t in state.turbines),
)
def _step_turbine_bank(self, state: PlantState, steam_power_mw: float) -> None:
if not state.turbines:
return
active_indices = [
idx for idx, active in enumerate(self.turbine_unit_active) if active and idx < len(state.turbines)
]
power_per_unit = steam_power_mw / len(active_indices) if active_indices else 0.0
for idx, turbine in enumerate(self.turbines):
if idx >= len(state.turbines):
break
turbine_state = state.turbines[idx]
if idx in active_indices:
turbine.step(state.secondary_loop, turbine_state, steam_power_mw=power_per_unit)
else:
self._reset_turbine_state(turbine_state)
self._dispatch_consumer_load(state, active_indices)
def _reset_turbine_state(self, turbine_state: TurbineState) -> None:
turbine_state.shaft_power_mw = 0.0
turbine_state.electrical_output_mw = 0.0
turbine_state.load_demand_mw = 0.0
turbine_state.load_supplied_mw = 0.0
def _dispatch_consumer_load(self, state: PlantState, active_indices: list[int]) -> None:
total_electrical = sum(state.turbines[idx].electrical_output_mw for idx in active_indices)
if self.consumer:
demand = self.consumer.request_power()
supplied = min(total_electrical, demand)
self.consumer.update_power_received(supplied)
else:
demand = 0.0
supplied = 0.0
demand_per_unit = demand / len(active_indices) if active_indices else 0.0
total_for_share = total_electrical if total_electrical > 0 else 1.0
for idx, turbine_state in enumerate(state.turbines):
if idx not in active_indices:
turbine_state.load_demand_mw = 0.0
turbine_state.load_supplied_mw = 0.0
continue
share = 0.0 if total_electrical <= 0 else supplied * (turbine_state.electrical_output_mw / total_for_share)
turbine_state.load_demand_mw = demand_per_unit
turbine_state.load_supplied_mw = share if demand_per_unit <= 0 else min(share, demand_per_unit)
def _handle_failure(self, component: str) -> None:
if component == "core":
LOGGER.critical("Core failure detected. Initiating SCRAM.")
@@ -171,8 +219,9 @@ class Reactor:
self._set_primary_pump(False)
elif component == "secondary_pump":
self._set_secondary_pump(False)
elif component == "turbine":
self._set_turbine_state(False)
elif component.startswith("turbine"):
idx = self._component_index(component)
self._set_turbine_state(False, index=idx)
def _apply_command(self, command: ReactorCommand, state: PlantState) -> dict[str, float]:
overrides: dict[str, float] = {}
@@ -196,12 +245,18 @@ class Reactor:
self._set_secondary_pump(command.secondary_pump_on)
if command.turbine_on is not None:
self._set_turbine_state(command.turbine_on)
if command.turbine_units:
for key, state_flag in command.turbine_units.items():
idx = key - 1
self._set_turbine_state(state_flag, index=idx)
if command.consumer_online is not None and self.consumer:
self.consumer.set_online(command.consumer_online)
if command.consumer_demand is not None and self.consumer:
self.consumer.set_demand(command.consumer_demand)
if command.coolant_demand is not None:
overrides["coolant_demand"] = max(0.0, min(1.0, command.coolant_demand))
for component in command.maintenance_components:
self._perform_maintenance(component)
return overrides
def _set_primary_pump(self, active: bool) -> None:
@@ -214,10 +269,46 @@ class Reactor:
self.secondary_pump_active = active
LOGGER.info("Secondary pump %s", "enabled" if active else "stopped")
def _set_turbine_state(self, active: bool) -> None:
if self.turbine_active != active:
self.turbine_active = active
LOGGER.info("Turbine %s", "started" if active else "stopped")
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)):
self._set_turbine_state(active, index=idx)
return
if index < 0 or index >= len(self.turbine_unit_active):
LOGGER.warning("Ignoring turbine index %s", index)
return
if self.turbine_unit_active[index] != active:
self.turbine_unit_active[index] = active
LOGGER.info("Turbine %d %s", index + 1, "started" if active else "stopped")
self.turbine_active = any(self.turbine_unit_active)
def _component_index(self, name: str) -> int:
if name == "turbine":
return 0
try:
return int(name.split("_")[1]) - 1
except (IndexError, ValueError):
return -1
def _perform_maintenance(self, component: str) -> None:
if component == "core" and not self.shutdown:
LOGGER.warning("Cannot maintain core while reactor is running")
return
if component == "primary_pump" and self.primary_pump_active:
LOGGER.warning("Stop primary pump before maintenance")
return
if component == "secondary_pump" and self.secondary_pump_active:
LOGGER.warning("Stop secondary pump before maintenance")
return
if component.startswith("turbine"):
idx = self._component_index(component)
if idx < 0 or idx >= len(self.turbine_unit_active):
LOGGER.warning("Unknown turbine maintenance target %s", component)
return
if self.turbine_unit_active[idx]:
LOGGER.warning("Stop turbine %d before maintenance", idx + 1)
return
self.health_monitor.maintain(component)
def attach_consumer(self, consumer: ElectricalConsumer) -> None:
self.consumer = consumer
@@ -233,6 +324,7 @@ class Reactor:
"primary_pump_active": self.primary_pump_active,
"secondary_pump_active": self.secondary_pump_active,
"turbine_active": self.turbine_active,
"turbine_units": self.turbine_unit_active,
"shutdown": self.shutdown,
"consumer": {
"online": self.consumer.online if self.consumer else False,
@@ -246,7 +338,10 @@ class Reactor:
plant, metadata, health = self.control.load_state(filepath)
self.primary_pump_active = metadata.get("primary_pump_active", self.primary_pump_active)
self.secondary_pump_active = metadata.get("secondary_pump_active", self.secondary_pump_active)
self.turbine_active = metadata.get("turbine_active", self.turbine_active)
unit_states = metadata.get("turbine_units")
if unit_states:
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)
consumer_cfg = metadata.get("consumer")
if consumer_cfg:
@@ -262,4 +357,17 @@ class Reactor:
if health:
self.health_monitor.load_snapshot(health)
LOGGER.info("Reactor state restored from %s", filepath)
if len(plant.turbines) < len(self.turbines):
ambient = constants.ENVIRONMENT_TEMPERATURE
while len(plant.turbines) < len(self.turbines):
plant.turbines.append(
TurbineState(
steam_enthalpy=2_000.0,
shaft_power_mw=0.0,
electrical_output_mw=0.0,
condenser_temperature=ambient,
load_demand_mw=0.0,
load_supplied_mw=0.0,
)
)
return plant