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_cheat: 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_cheat = is_cheat 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_cheat=True" if self.is_cheat 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, cheat_mode: bool = False): self.base_url = f'http://{host}:{port}/' self.dummy_mode = False self.cheat_mode = cheat_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'), 'FUN_IS_ENABLED': (bool, False), # --- 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 operational commands write_only_values = { # --- MSCVs (Main Steam Control Valves) setpoints --- **{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)}, # --- Steam ejector valve setpoints (0-100 position, not bool) --- 'STEAM_EJECTOR_STARTUP_MOTIVE_VALVE': (int, True, 0, 100, '%'), 'STEAM_EJECTOR_OPERATIONAL_MOTIVE_VALVE': (int, True, 0, 100, '%'), 'STEAM_EJECTOR_CONDENSER_RETURN_VALVE': (int, True, 0, 100, '%'), # --- Generic valve commands (value = valve name e.g. "M01", "M02", "M03") --- 'VALVE_OPEN': (str, True), 'VALVE_CLOSE': (str, True), 'VALVE_OFF': (str, True), # --- Pump / generator start/stop --- 'CONDENSER_VACUUM_PUMP_START_STOP': (bool, True), 'EMERGENCY_GENERATOR_1_START_STOP': (bool, True), 'EMERGENCY_GENERATOR_2_START_STOP': (bool, True), # --- Chemistry setpoints --- 'CHEM_BORON_DOSAGE_ORDERED_RATE': (float, True, 0, 100, '%'), 'CHEM_BORON_FILTER_ORDERED_SPEED': (float, True, 0, 100, '%'), # --- Core safety / operational actions --- 'CORE_SCRAM_BUTTON': (bool, True), 'CORE_EMERGENCY_STOP': (bool, True), 'CORE_END_EMERGENCY_STOP': (bool, True), 'RESET_AO': (bool, True), 'STEAM_TURBINE_TRIP': (bool, True), 'RODS_ALL_POS_ORDERED': (float, True, 0, 100, '%'), # --- 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)}, } # Write-only cheat params: game event triggers, blocked unless cheat_mode=True write_only_cheat_values = { '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_cheat_values.items(): params[name] = NuconParameter(self, name, *values, is_readable=False, is_cheat=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_cheat and not self.cheat_mode: raise ValueError(f"Parameter {parameter} is a cheat parameter. Enable cheat_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} # --- Valve API --- # Valves have a motorized actuator. OPEN/CLOSE power the motor toward that end-state; # OFF cuts power and holds the current position. Normal resting state is OFF. # The Value field (0-100) is the actual live position during travel. def _post_valve_command(self, command: str, valve_name: str) -> None: response = requests.post(self.base_url, params={"variable": command, "value": valve_name}) if response.status_code != 200: raise Exception(f"Valve command {command} on '{valve_name}' failed. Status: {response.status_code}") def get_valve(self, valve_name: str) -> Dict[str, Any]: """Return current state dict for a single valve (from VALVE_PANEL_JSON).""" valves = self.get_valves() if valve_name not in valves: raise KeyError(f"Valve '{valve_name}' not found") return valves[valve_name] def get_valves(self) -> Dict[str, Any]: """Return state dict for all valves, keyed by valve name.""" response = requests.get(self.base_url, params={"variable": "VALVE_PANEL_JSON"}) if response.status_code != 200: raise Exception(f"Failed to get valve panel. Status: {response.status_code}") return response.json().get("valves", {}) def open_valve(self, valve_name: str) -> None: """Power actuator toward open state. Send off_valve() once target is reached.""" self._post_valve_command("VALVE_OPEN", valve_name) def close_valve(self, valve_name: str) -> None: """Power actuator toward closed state. Send off_valve() once target is reached.""" self._post_valve_command("VALVE_CLOSE", valve_name) def off_valve(self, valve_name: str) -> None: """Cut actuator power, hold current position. Normal resting state.""" self._post_valve_command("VALVE_OFF", valve_name) def open_valves(self, valve_names: List[str]) -> None: for name in valve_names: self.open_valve(name) def close_valves(self, valve_names: List[str]) -> None: for name in valve_names: self.close_valve(name) def off_valves(self, valve_names: List[str]) -> None: for name in valve_names: self.off_valve(name) def set_dummy_mode(self, dummy_mode: bool) -> None: self.dummy_mode = dummy_mode def set_cheat_mode(self, cheat_mode: bool) -> None: self.cheat_mode = cheat_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)