Dominik Roth
2ec68ff2e5
Parameters like CORE_SCRAM_BUTTON, CORE_EMERGENCY_STOP, bay hatch/fuel loading, VALVE_OPEN/CLOSE/OFF, STEAM_TURBINE_TRIP, and all FUN_* event triggers are now marked is_admin=True. Writing to them is blocked unless the Nucon instance has admin_mode=True or force=True is used. Normal control setpoints (MSCV_*, STEAM_TURBINE_*_BYPASS_ORDERED, CHEM_BORON_*) remain write-only but are not admin-gated. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
628 lines
29 KiB
Python
628 lines
29 KiB
Python
from enum import Enum
|
|
from typing import Union, Dict, Type, List, Optional, Any, Iterator, Tuple
|
|
import requests
|
|
import random
|
|
|
|
class ParameterEnum(Enum):
|
|
@classmethod
|
|
def _missing_(cls, value: Any) -> Union['ParameterEnum', None]:
|
|
if isinstance(value, str):
|
|
if value.lower() == 'true':
|
|
return cls(True)
|
|
elif value.lower() == 'false':
|
|
return cls(False)
|
|
try:
|
|
return cls(int(value))
|
|
except ValueError:
|
|
pass
|
|
return None
|
|
|
|
class PumpStatus(ParameterEnum):
|
|
INACTIVE = 0
|
|
ACTIVE_NO_SPEED_REACHED = 1
|
|
ACTIVE_SPEED_REACHED = 2
|
|
REQUIRES_MAINTENANCE = 3
|
|
NOT_INSTALLED = 4
|
|
INSUFFICIENT_ENERGY = 5
|
|
|
|
def __bool__(self):
|
|
return self.value in (1, 2)
|
|
|
|
class PumpDryStatus(ParameterEnum):
|
|
ACTIVE_WITHOUT_FLUID = 1
|
|
INACTIVE_OR_ACTIVE_WITH_FLUID = 4
|
|
|
|
def __bool__(self):
|
|
return self.value == 1
|
|
|
|
class PumpOverloadStatus(ParameterEnum):
|
|
ACTIVE_AND_OVERLOAD = 1
|
|
INACTIVE_OR_ACTIVE_NO_OVERLOAD = 4
|
|
|
|
def __bool__(self):
|
|
return self.value == 1
|
|
|
|
class BreakerStatus(ParameterEnum):
|
|
OPEN = True
|
|
CLOSED = False
|
|
|
|
def __bool__(self):
|
|
return self.value
|
|
|
|
CoreState = str
|
|
CoolantCoreState = str
|
|
RodsState = str
|
|
|
|
# Known game endpoints that are not regular parameters (JSON/HTML data blobs, meta-commands)
|
|
SPECIAL_VARIABLES = frozenset({
|
|
'WEBSERVER_BATCH_GET',
|
|
'WEBSERVER_LIST_VARIABLES',
|
|
'WEBSERVER_LIST_VARIABLES_JSON',
|
|
'WEBSERVER_VIEW_VARIABLES',
|
|
'VALVE_PANEL_JSON',
|
|
'RESISTOR_BANKS_JSON',
|
|
'INSTALLED_LOOPS_JSON',
|
|
'INVENTORY_HTML',
|
|
'MAINTENANCE_REPORT_HTML',
|
|
'WEATHER_FORECAST_JSON',
|
|
})
|
|
|
|
class NuconParameter:
|
|
def __init__(self, nucon: 'Nucon', id: str, param_type: Type, is_writable: bool, min_val: Optional[Union[int, float]] = None, max_val: Optional[Union[int, float]] = None, unit: Optional[str] = None, is_readable: bool = True, is_admin: bool = False):
|
|
self.nucon = nucon
|
|
self.id = id
|
|
self.param_type = param_type
|
|
self.is_writable = is_writable
|
|
self.is_readable = is_readable
|
|
self.is_admin = is_admin
|
|
self.min_val = min_val
|
|
self.max_val = max_val
|
|
self.unit = unit
|
|
|
|
@property
|
|
def enum_type(self) -> Type[Enum]:
|
|
return self.param_type if issubclass(self.param_type, Enum) else None
|
|
|
|
def check_in_range(self, value: Union[int, float, Enum], raise_on_oob: bool = False) -> bool:
|
|
if self.enum_type:
|
|
if not isinstance(value, self.enum_type):
|
|
if raise_on_oob:
|
|
raise ValueError(f"Value {value} is not a valid {self.enum_type.__name__}")
|
|
return False
|
|
return True
|
|
|
|
if self.min_val is not None and value < self.min_val:
|
|
if raise_on_oob:
|
|
raise ValueError(f"Value {value} is below the minimum allowed value {self.min_val}")
|
|
return False
|
|
if self.max_val is not None and value > self.max_val:
|
|
if raise_on_oob:
|
|
raise ValueError(f"Value {value} is above the maximum allowed value {self.max_val}")
|
|
return False
|
|
return True
|
|
|
|
@property
|
|
def value(self):
|
|
if not self.is_readable:
|
|
raise AttributeError(f"Parameter {self.id} is write-only and cannot be read")
|
|
return self.nucon.get(self)
|
|
|
|
@value.setter
|
|
def value(self, new_value):
|
|
self.nucon.set(self, new_value)
|
|
|
|
def read(self):
|
|
return self.value
|
|
|
|
def write(self, new_value, force=False):
|
|
self.nucon.set(self, new_value, force)
|
|
|
|
def __repr__(self):
|
|
unit_str = f", unit='{self.unit}'" if self.unit else ""
|
|
value_str = f", value={self.value}" if self.is_readable else ""
|
|
rw_str = "write-only" if not self.is_readable else f"is_writable={self.is_writable}"
|
|
admin_str = ", is_admin=True" if self.is_admin else ""
|
|
return f"NuconParameter(id='{self.id}'{value_str}, param_type={self.param_type.__name__}, {rw_str}{admin_str}{unit_str})"
|
|
|
|
def __str__(self):
|
|
return self.id
|
|
|
|
class Nucon:
|
|
def __init__(self, host: str = 'localhost', port: int = 8785, admin_mode: bool = False):
|
|
self.base_url = f'http://{host}:{port}/'
|
|
self.dummy_mode = False
|
|
self.admin_mode = admin_mode
|
|
self._parameters = self._create_parameters()
|
|
|
|
def _create_parameters(self) -> Dict[str, NuconParameter]:
|
|
param_values = {
|
|
# --- Game metadata ---
|
|
'GAME_VERSION': (str, False),
|
|
'GAME_DIFFICULTY': (int, False),
|
|
'TIME': (str, False),
|
|
'TIME_STAMP': (str, False),
|
|
'TIME_DAY': (int, False),
|
|
'ALARMS_ACTIVE': (str, False),
|
|
'GAME_SIM_SPEED': (float, False),
|
|
'AMBIENT_TEMPERATURE': (float, False, None, None, '°C'),
|
|
|
|
# --- Core thermal/pressure ---
|
|
'CORE_TEMP': (float, False, 0, 1000, '°C'),
|
|
'CORE_TEMP_OPERATIVE': (float, False, None, None, '°C'),
|
|
'CORE_TEMP_MAX': (float, False, None, None, '°C'),
|
|
'CORE_TEMP_MIN': (float, False, None, None, '°C'),
|
|
'CORE_TEMP_RESIDUAL': (bool, False),
|
|
'CORE_PRESSURE': (float, False, None, None, 'bar'),
|
|
'CORE_PRESSURE_MAX': (float, False, None, None, 'bar'),
|
|
'CORE_PRESSURE_OPERATIVE': (float, False, None, None, 'bar'),
|
|
'CORE_INTEGRITY': (float, False, 0, 100, '%'),
|
|
'CORE_WEAR': (float, False, 0, 100, '%'),
|
|
'CORE_STATE': (CoreState, False),
|
|
'CORE_STATE_CRITICALITY': (float, False, 0, 1),
|
|
'CORE_CRITICAL_MASS_REACHED': (bool, False),
|
|
'CORE_CRITICAL_MASS_REACHED_COUNTER': (int, False),
|
|
'CORE_IMMINENT_FUSION': (bool, False),
|
|
'CORE_READY_FOR_START': (bool, False),
|
|
'CORE_STEAM_PRESENT': (bool, False),
|
|
'CORE_HIGH_STEAM_PRESENT': (bool, False),
|
|
|
|
# --- Core physics ---
|
|
'CORE_FACTOR': (float, False),
|
|
'CORE_FACTOR_CHANGE': (float, False),
|
|
'CORE_OPERATION_MODE': (str, True),
|
|
'CORE_IODINE_GENERATION': (float, False),
|
|
'CORE_IODINE_CUMULATIVE': (float, False),
|
|
'CORE_XENON_GENERATION': (float, False),
|
|
'CORE_XENON_CUMULATIVE': (float, False),
|
|
|
|
# --- Core fuel / bays (9 bays) ---
|
|
'CORE_FUEL_AVG_FISSIONABLE': (float, False, 0, 100, '%'),
|
|
'CORE_FUEL_AVG_TEMPERATURE': (float, False, None, None, '°C'),
|
|
'CORE_FUEL_AVG_POWER_FACTOR': (float, False, 0, 1),
|
|
**{f'CORE_FUEL_{i}_TEMPERATURE': (float, False, None, None, '°C') for i in range(1, 10)},
|
|
**{f'CORE_FUEL_{i}_FISSIONABLE': (float, False, 0, 100, '%') for i in range(1, 10)},
|
|
**{f'CORE_FUEL_{i}_POWER_FACTOR': (float, False, 0, 1) for i in range(1, 10)},
|
|
**{f'CORE_BAY_{i}_STATE': (str, False) for i in range(1, 10)},
|
|
**{f'CORE_BAY_{i}_HATCH_OPEN': (bool, False) for i in range(1, 10)},
|
|
|
|
# --- Core pool ---
|
|
'CORE_POOL_PUMP': (int, False), # command-style (LOAD/OFF/REMOVE), not freely settable
|
|
'CORE_POOL_COOLANT_TANK_VOLUME': (float, False),
|
|
'CORE_PRIMARY_CIRCUIT_COOLING_TANK_VOLUME': (float, False),
|
|
'CORE_EXTERNAL_COOLANT_RESERVOIR_VOLUME': (float, False),
|
|
|
|
# --- Primary coolant (core loop) ---
|
|
'COOLANT_CORE_STATE': (CoolantCoreState, False),
|
|
'COOLANT_CORE_PRESSURE': (float, False, None, None, 'bar'),
|
|
'COOLANT_CORE_MAX_PRESSURE': (float, False, None, None, 'bar'),
|
|
'COOLANT_CORE_VESSEL_TEMPERATURE': (float, False, None, None, '°C'),
|
|
'COOLANT_CORE_QUANTITY_IN_VESSEL': (float, False),
|
|
'COOLANT_CORE_PRIMARY_LOOP_LEVEL': (float, False, 0, 100, '%'),
|
|
'COOLANT_CORE_FLOW_IN': (float, False),
|
|
'COOLANT_CORE_FLOW_OUT': (float, False),
|
|
'COOLANT_CORE_FLOW_SPEED': (float, False),
|
|
'COOLANT_CORE_FLOW_ORDERED_SPEED': (float, False),
|
|
'COOLANT_CORE_FLOW_REACHED_SPEED': (bool, False),
|
|
'COOLANT_CORE_QUANTITY_CIRCULATION_PUMPS_PRESENT': (int, False),
|
|
'COOLANT_CORE_QUANTITY_FREIGHT_PUMPS_PRESENT': (int, False),
|
|
**{f'COOLANT_CORE_CIRCULATION_PUMP_{i}_STATUS': (PumpStatus, False) for i in range(3)},
|
|
**{f'COOLANT_CORE_CIRCULATION_PUMP_{i}_DRY_STATUS': (PumpDryStatus, False) for i in range(3)},
|
|
**{f'COOLANT_CORE_CIRCULATION_PUMP_{i}_OVERLOAD_STATUS': (PumpOverloadStatus, False) for i in range(3)},
|
|
**{f'COOLANT_CORE_CIRCULATION_PUMP_{i}_ORDERED_SPEED': (float, True, 0, 100) for i in range(3)},
|
|
**{f'COOLANT_CORE_CIRCULATION_PUMP_{i}_SPEED': (float, False, 0, 100) for i in range(3)},
|
|
**{f'COOLANT_CORE_CIRCULATION_PUMP_{i}_CAPACITY': (float, False) for i in range(3)},
|
|
|
|
# --- Secondary coolant loops ---
|
|
**{f'COOLANT_SEC_CIRCULATION_PUMP_{i}_STATUS': (PumpStatus, False) for i in range(3)},
|
|
**{f'COOLANT_SEC_CIRCULATION_PUMP_{i}_DRY_STATUS': (PumpDryStatus, False) for i in range(3)},
|
|
**{f'COOLANT_SEC_CIRCULATION_PUMP_{i}_OVERLOAD_STATUS': (PumpOverloadStatus, False) for i in range(3)},
|
|
**{f'COOLANT_SEC_CIRCULATION_PUMP_{i}_ORDERED_SPEED': (float, True, 0, 100) for i in range(3)},
|
|
**{f'COOLANT_SEC_CIRCULATION_PUMP_{i}_SPEED': (float, False, 0, 100) for i in range(3)},
|
|
**{f'COOLANT_SEC_CIRCULATION_PUMP_{i}_CAPACITY': (float, False) for i in range(3)},
|
|
**{f'COOLANT_SEC_{i}_VOLUME': (float, False) for i in range(3)},
|
|
**{f'COOLANT_SEC_{i}_LIQUID_VOLUME': (float, False) for i in range(3)},
|
|
**{f'COOLANT_SEC_{i}_PRESSURE': (float, False, None, None, 'bar') for i in range(3)},
|
|
**{f'COOLANT_SEC_{i}_TEMPERATURE': (float, False, None, None, '°C') for i in range(3)},
|
|
|
|
# --- Control rods ---
|
|
'RODS_STATUS': (RodsState, False),
|
|
'RODS_MOVEMENT_SPEED': (float, False),
|
|
'RODS_MOVEMENT_SPEED_DECREASED_HIGH_TEMPERATURE': (bool, False),
|
|
'RODS_DEFORMED': (bool, False),
|
|
'RODS_TEMPERATURE': (float, False, None, None, '°C'),
|
|
'RODS_MAX_TEMPERATURE': (float, False, None, None, '°C'),
|
|
'RODS_POS_ORDERED': (float, False, 0, 100, '%'),
|
|
'RODS_POS_ACTUAL': (float, False, 0, 100, '%'),
|
|
'RODS_POS_REACHED': (bool, False),
|
|
'RODS_QUANTITY': (int, False),
|
|
'RODS_ALIGNED': (int, False),
|
|
**{f'ROD_BANK_POS_{i}_ORDERED': (float, True, 0, 100, '%') for i in range(9)},
|
|
**{f'ROD_BANK_POS_{i}_ACTUAL': (float, False, 0, 100, '%') for i in range(9)},
|
|
|
|
# --- Steam generators ---
|
|
**{f'STEAM_GEN_{i}_STATUS': (PumpStatus, False) for i in range(3)},
|
|
**{f'STEAM_GEN_{i}_OUTLET': (float, False) for i in range(3)},
|
|
**{f'STEAM_GEN_{i}_EVAPORATED': (float, False) for i in range(3)},
|
|
**{f'STEAM_GEN_{i}_BOILING_POINT': (float, False, None, None, '°C') for i in range(3)},
|
|
**{f'STEAM_GEN_{i}_INLET': (float, False) for i in range(3)},
|
|
**{f'STEAM_GEN_{i}_RETURN_FLOW_PLUS_CONDENSED': (float, False) for i in range(3)},
|
|
**{f'STEAM_GEN_{i}_VENT_SWITCH': (bool, True) for i in range(3)},
|
|
|
|
# --- Steam turbines ---
|
|
**{f'STEAM_TURBINE_{i}_RPM': (float, False) for i in range(3)},
|
|
**{f'STEAM_TURBINE_{i}_TEMPERATURE': (float, False, None, None, '°C') for i in range(3)},
|
|
**{f'STEAM_TURBINE_{i}_PRESSURE': (float, False, None, None, 'bar') for i in range(3)},
|
|
**{f'STEAM_TURBINE_{i}_TORQUE': (float, False) for i in range(3)},
|
|
**{f'STEAM_TURBINE_{i}_INSTALLED': (bool, False) for i in range(3)},
|
|
**{f'STEAM_TURBINE_{i}_BYPASS_ACTUAL': (float, False, 0, 100, '%') for i in range(3)},
|
|
|
|
# --- MSCV valves ---
|
|
**{f'MSCV_{i}_OPENING_ACTUAL': (float, False, 0, 100, '%') for i in range(3)},
|
|
|
|
# --- Main steam valves ---
|
|
**{f'VALVE_M0{i}_OPEN': (float, False, 0, 100, '%') for i in range(1, 4)},
|
|
|
|
# --- Condenser ---
|
|
'CONDENSER_VACUUM': (float, False, 0, 100, '%'),
|
|
'CONDENSER_VACUUM_RELIEF_VALVE_OPENING': (float, False, 0, 100, '%'),
|
|
'CONDENSER_VACUUM_PUMP_ACTIVE': (bool, False),
|
|
'CONDENSER_VACUUM_PUMP_MODE': (str, True),
|
|
'CONDENSER_VACUUM_PUMP_POWER': (float, False, 0, 100, '%'),
|
|
'CONDENSER_TEMPERATURE': (float, False, None, None, '°C'),
|
|
'CONDENSER_VOLUME': (float, False),
|
|
'CONDENSER_VAPOR_VOLUME': (float, False),
|
|
'CONDENSER_CONDENSATE_FLOW_RATE': (float, False),
|
|
'CONDENSER_EXTRACTION_FLOW_RATE': (float, False),
|
|
'CONDENSER_COOLANT_EVAPORATED': (float, False),
|
|
'CONDENSER_PRESSURE': (float, False, None, None, 'bar'),
|
|
'CONDENSER_CIRCULATION_PUMP_ACTIVE': (bool, False),
|
|
'CONDENSER_CIRCULATION_PUMP_OVERLOAD_STATUS': (bool, False),
|
|
'CONDENSER_CIRCULATION_PUMP_SPEED': (float, False, 0, 100),
|
|
'CONDENSER_CIRCULATION_PUMP_ORDERED_SPEED': (float, True, 0, 100),
|
|
'CONDENSER_CIRCULATION_PUMP_SWITCH': (bool, True),
|
|
|
|
# --- Vacuum retention / steam ejector ---
|
|
'VACUUM_RETENTION_TANK_VOLUME': (float, False),
|
|
'VACUUM_RETENTION_TANK_PRESSURE': (float, False, None, None, 'bar'),
|
|
'STEAM_EJECTOR_MOTIVE': (float, False),
|
|
'STEAM_EJECTOR_STARTUP_MOTIVE_VALVE_ORDERED': (float, False, 0, 100, '%'),
|
|
'STEAM_EJECTOR_STARTUP_MOTIVE_VALVE_ACTUAL': (float, False, 0, 100, '%'),
|
|
'STEAM_EJECTOR_OPERATIONAL_MOTIVE_VALVE_ORDERED': (float, False, 0, 100, '%'),
|
|
'STEAM_EJECTOR_OPERATIONAL_MOTIVE_VALVE_ACTUAL': (float, False, 0, 100, '%'),
|
|
'STEAM_EJECTOR_CONDENSER_RETURN_VALVE_ORDERED': (float, False, 0, 100, '%'),
|
|
'STEAM_EJECTOR_CONDENSER_RETURN_VALVE_ACTUAL': (float, False, 0, 100, '%'),
|
|
|
|
# --- Freight pumps ---
|
|
'FREIGHT_PUMP_CONDENSER_ACTIVE': (bool, False),
|
|
'FREIGHT_PUMP_INTERNAL_ACTIVE': (bool, False),
|
|
'FREIGHT_PUMP_EXTERNAL_ACTIVE': (bool, False),
|
|
'FREIGHT_PUMP_FEEDWATER_ACTIVE': (bool, False),
|
|
'FREIGHT_PUMP_CONDENSER_SWITCH': (bool, True),
|
|
'FREIGHT_PUMP_INTERNAL_SWITCH': (bool, True),
|
|
'FREIGHT_PUMP_EXTERNAL_SWITCH': (bool, True),
|
|
'FREIGHT_PUMP_FEEDWATER_SWITCH': (bool, True),
|
|
|
|
# --- Generators ---
|
|
**{f'GENERATOR_{i}_KW': (float, False, None, None, 'kW') for i in range(3)},
|
|
**{f'GENERATOR_{i}_V': (float, False, None, None, 'V') for i in range(3)},
|
|
**{f'GENERATOR_{i}_A': (float, False, None, None, 'A') for i in range(3)},
|
|
**{f'GENERATOR_{i}_HERTZ': (float, False, None, None, 'Hz') for i in range(3)},
|
|
**{f'GENERATOR_{i}_BREAKER': (BreakerStatus, False) for i in range(3)},
|
|
|
|
# --- Resistor banks ---
|
|
'RES_DIVERT_SURPLUS_FROM_MW': (float, False, None, None, 'MW'),
|
|
'RES_EFFECTIVELY_DERIVED_ENERGY_MW': (float, False, None, None, 'MW'),
|
|
'RES_ABSORPTION_CAPACITY_MW': (float, False, None, None, 'MW'),
|
|
**{f'RESISTOR_BANK_0{i}_SWITCH': (bool, True) for i in range(1, 5)},
|
|
'RESISTOR_BANKS_MAIN_SWITCH': (bool, True),
|
|
|
|
# --- Emergency generators ---
|
|
'EMERGENCY_GENERATOR_1_MODE': (str, True),
|
|
'EMERGENCY_GENERATOR_1_STATUS': (str, False),
|
|
'EMERGENCY_GENERATOR_1_PRESSURIZER': (str, False),
|
|
'EMERGENCY_GENERATOR_1_FUEL': (float, False),
|
|
'EMERGENCY_GENERATOR_1_MAINTENANCE_NEEDED': (bool, False),
|
|
'EMERGENCY_GENERATOR_2_MODE': (str, True),
|
|
'EMERGENCY_GENERATOR_2_STATUS': (str, False),
|
|
'EMERGENCY_GENERATOR_2_PRESSURIZER': (str, False),
|
|
'EMERGENCY_GENERATOR_2_FUEL': (float, False),
|
|
'EMERGENCY_GENERATOR_2_MAINTENANCE_NEEDED': (bool, False),
|
|
|
|
# --- Power ---
|
|
'POWER_FROM_TURBINE_KW': (float, False, None, None, 'kW'),
|
|
'POWER_FROM_EXTERNAL_KW': (float, False, None, None, 'kW'),
|
|
'EMERGENCY_GENERATOR_POWER_OUTPUT_KW': (float, False, None, None, 'kW'),
|
|
'EMERGENCY_BATTERIES_POWER_OUTPUT_KW': (float, False, None, None, 'kW'),
|
|
'EMERGENCY_BATTERIES_MODE': (int, True),
|
|
'POWER_DEMAND_MW': (float, False, None, None, 'MW'),
|
|
'POWER_MAX_THEORETICAL_FINAL_PLANT_OUTPUT_MW': (float, False, None, None, 'MW'),
|
|
'POWER_MAX_THEORETICAL_PLANT_OUTPUT_MW': (float, False, None, None, 'MW'),
|
|
|
|
# --- Chemistry ---
|
|
'CHEM_TRUCK_IN_ZONE': (bool, False),
|
|
'CHEM_TRUCK_CONNECTED': (bool, False),
|
|
'CHEM_BORON_DOSAGE_ORDERED': (float, False),
|
|
'CHEM_BORON_DOSAGE_ACTUAL': (float, False),
|
|
'CHEM_BORON_FILTER_ORDERED': (float, False),
|
|
'CHEM_BORON_FILTER_ACTUAL': (float, False),
|
|
'CHEM_BORON_PPM': (float, False, None, None, 'ppm'),
|
|
'CHEMICAL_DOSING_PUMP_STATUS': (PumpStatus, False),
|
|
'CHEMICAL_DOSING_PUMP_DRY_STATUS': (PumpDryStatus, False),
|
|
'CHEMICAL_DOSING_PUMP_OVERLOAD_STATUS': (PumpOverloadStatus, False),
|
|
'CHEMICAL_FILTER_PUMP_STATUS': (PumpStatus, False),
|
|
'CHEMICAL_FILTER_PUMP_DRY_STATUS': (PumpDryStatus, False),
|
|
'CHEMICAL_FILTER_PUMP_OVERLOAD_STATUS': (PumpOverloadStatus, False),
|
|
'CHEMICAL_CLEANING_PUMP_STATUS': (PumpStatus, False),
|
|
'CHEMICAL_CLEANING_PUMP_DRY_STATUS': (PumpDryStatus, False),
|
|
'CHEMICAL_CLEANING_PUMP_OVERLOAD_STATUS': (PumpOverloadStatus, False),
|
|
}
|
|
|
|
# Write-only params: normal control setpoints (no admin restriction)
|
|
write_only_values = {
|
|
# --- MSCVs (Main Steam Control Valves) ---
|
|
**{f'MSCV_{i}_OPENING_ORDERED': (float, True, 0, 100, '%') for i in range(3)},
|
|
|
|
# --- Steam turbine bypass setpoints ---
|
|
**{f'STEAM_TURBINE_{i}_BYPASS_ORDERED': (float, True, 0, 100, '%') for i in range(3)},
|
|
|
|
# --- Chemistry setpoints ---
|
|
'CHEM_BORON_DOSAGE_ORDERED_RATE': (float, True, 0, 100, '%'),
|
|
'CHEM_BORON_FILTER_ORDERED_SPEED': (float, True, 0, 100, '%'),
|
|
}
|
|
|
|
# Write-only admin params: destructive/irreversible operations, blocked unless admin_mode=True
|
|
write_only_admin_values = {
|
|
# --- Core safety actions ---
|
|
'CORE_SCRAM_BUTTON': (bool, True),
|
|
'CORE_EMERGENCY_STOP': (bool, True),
|
|
'CORE_END_EMERGENCY_STOP': (bool, True),
|
|
'RESET_AO': (bool, True),
|
|
|
|
# --- Core bay physical operations ---
|
|
**{f'CORE_BAY_{i}_HATCH': (bool, True) for i in range(1, 10)},
|
|
**{f'CORE_BAY_{i}_FUEL_LOADING': (int, True) for i in range(1, 10)},
|
|
|
|
# --- Bulk rod override ---
|
|
'RODS_ALL_POS_ORDERED': (float, True, 0, 100, '%'),
|
|
|
|
# --- Steam turbine trip ---
|
|
'STEAM_TURBINE_TRIP': (bool, True),
|
|
|
|
# --- Steam ejector valves ---
|
|
'STEAM_EJECTOR_CONDENSER_RETURN_VALVE': (bool, True),
|
|
'STEAM_EJECTOR_OPERATIONAL_MOTIVE_VALVE': (bool, True),
|
|
'STEAM_EJECTOR_STARTUP_MOTIVE_VALVE': (bool, True),
|
|
|
|
# --- Generic valve commands (take valve name as value) ---
|
|
'VALVE_OPEN': (str, True),
|
|
'VALVE_CLOSE': (str, True),
|
|
'VALVE_OFF': (str, True),
|
|
|
|
# --- Infrastructure start/stop ---
|
|
'CONDENSER_VACUUM_PUMP_START_STOP': (bool, True),
|
|
'EMERGENCY_GENERATOR_1_START_STOP': (bool, True),
|
|
'EMERGENCY_GENERATOR_2_START_STOP': (bool, True),
|
|
|
|
# --- Fun / event triggers (game cheats) ---
|
|
'FUN_IS_ENABLED': (bool, True),
|
|
'FUN_REQUEST_ENABLE': (bool, True),
|
|
'FUN_AO_SABOTAGE_ONCE': (bool, True),
|
|
'FUN_AO_SABOTAGE_TIME': (float, True),
|
|
'FUN_BANK_ROBBERY': (bool, True),
|
|
'FUN_BREAKER_TRIP': (bool, True),
|
|
'FUN_DECREASE_INTEGRITY': (float, True),
|
|
'FUN_FIRE_DRILL': (bool, True),
|
|
'FUN_IODINE_SPILL': (bool, True),
|
|
'FUN_OIL_SPILL': (bool, True),
|
|
'FUN_PUMP_JAM': (bool, True),
|
|
'FUN_SHOW_MESSAGE': (str, True),
|
|
'FUN_TOGGLE_RANDOM_SWITCH': (bool, True),
|
|
'FUN_TRIGGER_AUDIT': (bool, True),
|
|
'FUN_WEATHER_CONTROL': (str, True),
|
|
'FUN_XENON_SPILL': (bool, True),
|
|
}
|
|
|
|
params = {
|
|
name: NuconParameter(self, name, *values)
|
|
for name, values in param_values.items()
|
|
}
|
|
for name, values in write_only_values.items():
|
|
params[name] = NuconParameter(self, name, *values, is_readable=False)
|
|
for name, values in write_only_admin_values.items():
|
|
params[name] = NuconParameter(self, name, *values, is_readable=False, is_admin=True)
|
|
return params
|
|
|
|
def _parse_value(self, parameter: NuconParameter, value: str) -> Union[float, int, bool, str, Enum, None]:
|
|
if value == '' or value is None or (isinstance(value, str) and value.lower() == 'null'):
|
|
return None
|
|
if parameter.enum_type:
|
|
try:
|
|
return parameter.enum_type(value)
|
|
except ValueError as e:
|
|
raise ValueError(f"Failed to convert {value} to {parameter.enum_type.__name__} for parameter {parameter.id}: {e}")
|
|
elif parameter.param_type == bool:
|
|
if isinstance(value, str):
|
|
if value.lower() not in ('true', 'false'):
|
|
raise ValueError(f"Invalid boolean value for parameter {parameter.id}: {value}")
|
|
return value.lower() == 'true'
|
|
else:
|
|
raise ValueError(f"Expected string for boolean parameter {parameter.id}, got {type(value)}")
|
|
else:
|
|
try:
|
|
return parameter.param_type(value)
|
|
except ValueError as e:
|
|
raise ValueError(f"Failed to convert {value} to {parameter.param_type.__name__} for parameter {parameter.id}: {e}")
|
|
|
|
def get(self, parameter: Union[str, NuconParameter]) -> Union[float, int, bool, str, Enum]:
|
|
if isinstance(parameter, str):
|
|
parameter = self._parameters[parameter]
|
|
|
|
if self.dummy_mode:
|
|
return self._get_dummy_value(parameter)
|
|
|
|
value = self._query(parameter)
|
|
return self._parse_value(parameter, value)
|
|
|
|
def set(self, parameter: Union[str, NuconParameter], value: Union[float, int, bool, str, Enum], force: bool = False) -> None:
|
|
if isinstance(parameter, str):
|
|
parameter = self._parameters[parameter]
|
|
|
|
if not force and not parameter.is_writable:
|
|
raise ValueError(f"Parameter {parameter} is not writable")
|
|
if not force and parameter.is_admin and not self.admin_mode:
|
|
raise ValueError(f"Parameter {parameter} is an admin parameter. Enable admin_mode on the Nucon instance or use force=True")
|
|
|
|
if not force:
|
|
parameter.check_in_range(value, raise_on_oob=True)
|
|
|
|
if parameter.enum_type and isinstance(value, parameter.enum_type):
|
|
value = value.value
|
|
|
|
if self.dummy_mode:
|
|
print(f"Dummy mode: {'Force ' if force else ''}Setting {parameter} to {value}")
|
|
return
|
|
|
|
self._set_value(parameter, str(value))
|
|
|
|
def get_type(self, parameter: Union[str, NuconParameter]) -> Type:
|
|
if isinstance(parameter, str):
|
|
parameter = self._parameters[parameter]
|
|
return parameter.param_type
|
|
|
|
def _query(self, parameter: NuconParameter) -> str:
|
|
response = requests.get(self.base_url, params={"variable": parameter.id})
|
|
|
|
if response.status_code != 200:
|
|
raise Exception(f"Failed to query parameter {parameter.id}. Status code: {response.status_code}")
|
|
|
|
if response.text.strip() == 'NOT FOUND':
|
|
raise Exception(f"Failed to query parameter {parameter.id}. Returned 'NOT FOUND'")
|
|
|
|
return response.text.strip()
|
|
|
|
def _batch_query(self, param_names: List[str]) -> Dict[str, str]:
|
|
value = ','.join(param_names)
|
|
response = requests.get(self.base_url, params={"variable": "WEBSERVER_BATCH_GET", "value": value})
|
|
|
|
if response.status_code != 200:
|
|
raise Exception(f"Batch query failed. Status code: {response.status_code}")
|
|
|
|
data = response.json()
|
|
# Game returns {"values": {...}, "errors": {...}} wrapper
|
|
if isinstance(data, dict) and 'values' in data:
|
|
data = data['values']
|
|
return {k.upper(): ('' if v is None else str(v)) for k, v in data.items()}
|
|
|
|
def get_game_variable_names(self) -> List[str]:
|
|
response = requests.get(self.base_url, params={"variable": "WEBSERVER_LIST_VARIABLES"})
|
|
|
|
if response.status_code != 200:
|
|
raise Exception(f"Failed to get variable list. Status code: {response.status_code}")
|
|
|
|
names = []
|
|
for line in response.text.strip().splitlines():
|
|
line = line.strip()
|
|
if not line:
|
|
continue
|
|
# Format is "GET:var1,var2,..." or "POST:var1,var2,..."
|
|
if ':' in line:
|
|
line = line.split(':', 1)[1]
|
|
names.extend(v.strip() for v in line.split(',') if v.strip())
|
|
# Return unique names, excluding known special (non-parameter) endpoints
|
|
seen = set()
|
|
result = []
|
|
for name in names:
|
|
upper = name.upper()
|
|
if upper not in seen and upper not in SPECIAL_VARIABLES:
|
|
seen.add(upper)
|
|
result.append(name)
|
|
return result
|
|
|
|
def _set_value(self, parameter: NuconParameter, value: str) -> None:
|
|
response = requests.post(self.base_url, params={"variable": parameter.id, "value": value})
|
|
|
|
if response.status_code != 200:
|
|
raise Exception(f"Failed to set parameter {parameter.id}. Status code: {response.status_code}")
|
|
|
|
def _get_dummy_value(self, parameter: NuconParameter) -> Union[float, int, bool, str, Enum]:
|
|
if parameter.enum_type:
|
|
return next(iter(parameter.enum_type))
|
|
elif parameter.param_type == float:
|
|
if parameter.max_val is not None and parameter.min_val is not None:
|
|
return (parameter.max_val - parameter.min_val) / 2 + parameter.min_val
|
|
else:
|
|
return 3.14
|
|
elif parameter.param_type == int:
|
|
if parameter.max_val is not None and parameter.min_val is not None:
|
|
return (parameter.max_val - parameter.min_val) // 2 + parameter.min_val
|
|
else:
|
|
return 42
|
|
elif parameter.param_type == bool:
|
|
return random.choice([True, False])
|
|
else:
|
|
return "dummy"
|
|
|
|
def get_multiple_iter(self, parameters: List[Union[str, NuconParameter]]) -> Iterator[Tuple[str, Union[float, int, bool, str, Enum]]]:
|
|
for param in parameters:
|
|
if isinstance(param, str):
|
|
param_name = param
|
|
param_obj = self._parameters[param]
|
|
else:
|
|
param_name = next(name for name, p in self._parameters.items() if p is param)
|
|
param_obj = param
|
|
yield param_name, self.get(param_obj)
|
|
|
|
def get_multiple(self, parameters: List[Union[str, NuconParameter]]) -> Dict[str, Union[float, int, bool, str, Enum]]:
|
|
return dict(self.get_multiple_iter(parameters))
|
|
|
|
def get_all_iter(self) -> Iterator[Tuple[str, Union[float, int, bool, str, Enum]]]:
|
|
if self.dummy_mode:
|
|
yield from self.get_multiple_iter(self._parameters.keys())
|
|
return
|
|
|
|
try:
|
|
raw = self._batch_query(list(self._parameters.keys()))
|
|
except Exception:
|
|
raw = {}
|
|
|
|
for name, param in self._parameters.items():
|
|
if not param.is_readable:
|
|
continue
|
|
raw_value = raw.get(name)
|
|
# Batch query returns int codes for str-typed params; use individual query for those
|
|
if raw_value is not None and param.param_type != str:
|
|
yield name, self._parse_value(param, raw_value)
|
|
else:
|
|
yield name, self.get(param)
|
|
|
|
def get_all(self) -> Dict[str, Union[float, int, bool, str, Enum]]:
|
|
return dict(self.get_all_iter())
|
|
|
|
def get_all_readable(self) -> List[NuconParameter]:
|
|
return {name: param for name, param in self._parameters.items() if param.is_readable}
|
|
|
|
def get_all_writable(self) -> List[NuconParameter]:
|
|
return {name: param for name, param in self._parameters.items() if param.is_writable}
|
|
|
|
def set_dummy_mode(self, dummy_mode: bool) -> None:
|
|
self.dummy_mode = dummy_mode
|
|
|
|
def set_admin_mode(self, admin_mode: bool) -> None:
|
|
self.admin_mode = admin_mode
|
|
|
|
def __getattr__(self, name):
|
|
if isinstance(name, int):
|
|
return self.__getattr__(list(self._parameters.keys())[name])
|
|
if name in self._parameters:
|
|
return self._parameters[name]
|
|
raise AttributeError(f"'{self.__class__.__name__}' object has no attribute '{name}'")
|
|
|
|
def __getitem__(self, key):
|
|
return self.__getattr__(key)
|
|
|
|
def __dir__(self):
|
|
return list(super().__dir__()) + list(self._parameters.keys())
|
|
|
|
def __len__(self):
|
|
return sum(1 for p in self._parameters.values() if p.is_readable)
|