dodox/NuCon
1
1
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

Add valve API, cheat_mode, and write-only param fixes

Browse Source 
- Rename is_admin/admin_mode -> is_cheat/cheat_mode (only FUN_* event
  triggers are cheat params, not operational commands like SCRAM)
- Fix steam ejector valve write commands: int 0-100, not bool
- Move SCRAM, EMERGENCY_STOP, bay hatches, turbine trip etc. to normal
  write-only (not cheat-gated)
- Add FUN_IS_ENABLED to readable params (it appears in GET list)
- Add get_valve/get_valves, open/close/off_valve(s) methods with correct
  actuator semantics: OPEN/CLOSE powers motor, OFF holds position

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
Dominik Moritz Roth 2026-03-12 17:16:08 +01:00
parent 2ec68ff2e5
commit c78106dffc
1 changed files with 87 additions and 44 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

131
nucon/core.py
View File

@ -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):