diff --git a/nucon/core.py b/nucon/core.py index e7f0695..2002d3b 100644 --- a/nucon/core.py +++ b/nucon/core.py @@ -68,13 +68,13 @@ SPECIAL_VARIABLES = frozenset({ }) 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): + 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_admin = is_admin + self.is_cheat = is_cheat self.min_val = min_val self.max_val = max_val self.unit = unit @@ -121,17 +121,17 @@ class NuconParameter: 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 "" + 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, admin_mode: bool = False): + 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.admin_mode = admin_mode + self.cheat_mode = cheat_mode self._parameters = self._create_parameters() def _create_parameters(self) -> Dict[str, NuconParameter]: @@ -145,6 +145,7 @@ class Nucon: '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'), @@ -357,54 +358,48 @@ class Nucon: 'CHEMICAL_CLEANING_PUMP_OVERLOAD_STATUS': (PumpOverloadStatus, False), } - # Write-only params: normal control setpoints (no admin restriction) + # Write-only params: normal operational commands write_only_values = { - # --- MSCVs (Main Steam Control Valves) --- + # --- 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)}, - # --- Chemistry setpoints --- - 'CHEM_BORON_DOSAGE_ORDERED_RATE': (float, True, 0, 100, '%'), - 'CHEM_BORON_FILTER_ORDERED_SPEED': (float, True, 0, 100, '%'), - } + # --- 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, '%'), - # 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) --- + # --- Generic valve commands (value = valve name e.g. "M01", "M02", "M03") --- 'VALVE_OPEN': (str, True), 'VALVE_CLOSE': (str, True), 'VALVE_OFF': (str, True), - # --- Infrastructure start/stop --- + # --- 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), - # --- Fun / event triggers (game cheats) --- - 'FUN_IS_ENABLED': (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), @@ -428,8 +423,8 @@ class Nucon: } 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) + 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]: @@ -469,8 +464,8 @@ class Nucon: 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 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) @@ -604,11 +599,59 @@ class Nucon: 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_admin_mode(self, admin_mode: bool) -> None: - self.admin_mode = admin_mode + def set_cheat_mode(self, cheat_mode: bool) -> None: + self.cheat_mode = cheat_mode def __getattr__(self, name): if isinstance(name, int):