137 lines
4.7 KiB
Python
137 lines
4.7 KiB
Python
"""Wear and failure monitoring for reactor components."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass, asdict
|
|
import logging
|
|
from typing import Dict, Iterable, List
|
|
|
|
from . import constants
|
|
from .state import PlantState
|
|
|
|
LOGGER = logging.getLogger(__name__)
|
|
|
|
|
|
@dataclass
|
|
class ComponentHealth:
|
|
name: str
|
|
integrity: float = 1.0
|
|
failed: bool = False
|
|
|
|
def degrade(self, amount: float) -> None:
|
|
if self.failed:
|
|
return
|
|
self.integrity = max(0.0, self.integrity - amount)
|
|
if self.integrity <= 0.0:
|
|
self.fail()
|
|
|
|
def fail(self) -> None:
|
|
if not self.failed:
|
|
self.failed = True
|
|
LOGGER.error("Component %s has failed", self.name)
|
|
|
|
def restore(self, amount: float) -> None:
|
|
if amount <= 0.0:
|
|
return
|
|
previous = self.integrity
|
|
self.integrity = min(1.0, self.integrity + amount)
|
|
if self.integrity > 0.05:
|
|
self.failed = False
|
|
LOGGER.info("Maintenance on %s: %.3f -> %.3f", self.name, previous, self.integrity)
|
|
|
|
def snapshot(self) -> dict:
|
|
return asdict(self)
|
|
|
|
@classmethod
|
|
def from_snapshot(cls, data: dict) -> "ComponentHealth":
|
|
return cls(**data)
|
|
|
|
|
|
class HealthMonitor:
|
|
"""Tracks component wear and signals failures."""
|
|
|
|
def __init__(self) -> None:
|
|
self.components: Dict[str, ComponentHealth] = {
|
|
"core": ComponentHealth("core"),
|
|
"primary_pump": ComponentHealth("primary_pump"),
|
|
"secondary_pump": ComponentHealth("secondary_pump"),
|
|
}
|
|
for idx in range(2):
|
|
name = f"turbine_{idx + 1}"
|
|
self.components[name] = ComponentHealth(name)
|
|
self.failure_log: list[str] = []
|
|
|
|
def component(self, name: str) -> ComponentHealth:
|
|
return self.components[name]
|
|
|
|
def evaluate(
|
|
self,
|
|
state: PlantState,
|
|
primary_active: bool,
|
|
secondary_active: bool,
|
|
turbine_active: Iterable[bool],
|
|
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))
|
|
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)
|
|
else:
|
|
self.component("primary_pump").degrade(0.0005 * dt)
|
|
|
|
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)
|
|
else:
|
|
self.component("secondary_pump").degrade(0.0005 * dt)
|
|
|
|
turbines = state.turbines if hasattr(state, "turbines") else []
|
|
for idx, active in enumerate(turbine_flags):
|
|
name = f"turbine_{idx + 1}"
|
|
component = self.component(name)
|
|
if active and idx < len(turbines):
|
|
turbine_state = turbines[idx]
|
|
demand = turbine_state.load_demand_mw
|
|
supplied = turbine_state.load_supplied_mw
|
|
if demand <= 0:
|
|
load_ratio = 0.0
|
|
else:
|
|
load_ratio = min(1.0, supplied / max(1e-6, demand))
|
|
mismatch = abs(1 - load_ratio)
|
|
stress = 0.00005 + mismatch * 0.0006
|
|
else:
|
|
stress = 0.00002
|
|
component.degrade(stress * dt)
|
|
|
|
for name, component in self.components.items():
|
|
if component.failed and name not in self.failure_log:
|
|
events.append(name)
|
|
self.failure_log.append(name)
|
|
return events
|
|
|
|
def snapshot(self) -> dict:
|
|
return {name: comp.snapshot() for name, comp in self.components.items()}
|
|
|
|
def load_snapshot(self, data: dict) -> None:
|
|
for name, comp_data in data.items():
|
|
mapped = "turbine_1" if name == "turbine" else name
|
|
if mapped in self.components:
|
|
self.components[mapped] = ComponentHealth.from_snapshot(comp_data)
|
|
|
|
def maintain(self, component: str, amount: float = 0.05) -> bool:
|
|
comp = self.components.get(component)
|
|
if not comp:
|
|
LOGGER.warning("Maintenance requested for unknown component %s", component)
|
|
return False
|
|
comp.restore(amount)
|
|
return True
|