Compare commits

..

4 Commits

Author SHA1 Message Date
b0a2ac7574 Fix port in README 2024-10-02 22:36:54 +02:00
f6598c908c Define url 2024-10-02 22:36:43 +02:00
9a690f42dc Fix port and allow silence float could be int 2024-10-02 22:36:20 +02:00
0d99378f7d Bug fixes and additions 2024-10-02 22:36:06 +02:00
4 changed files with 108 additions and 42 deletions

View File

@ -28,7 +28,7 @@ Here's a basic example of how to use NuCon:
from nucon import Nucon, BreakerStatus
# Set the base URL for the game's API (if different from default)
Nucon.set_base_url("http://localhost:8080/")
Nucon.set_base_url("http://localhost:8785/")
# Enable dummy mode for testing (optional)
Nucon.set_dummy_mode(True)
@ -134,7 +134,7 @@ NuCon includes a test suite to verify its functionality and compatibility with t
To run the tests:
1. Ensure the Nucleares game is running and accessible at http://localhost:8080/ (or update the URL in the test setup).
1. Ensure the Nucleares game is running and accessible at http://localhost:8785/ (or update the URL in the test setup).
2. Install pytest: `pip install pytest`
3. Run the tests: `pytest test/test.py`

View File

@ -1,12 +1,31 @@
from enum import Enum, IntEnum
import requests
from typing import Union, Dict, Type, List
from typing import Union, Dict, Type, List, Optional, Any
class NuconConfig:
base_url = "http://localhost:8080/"
base_url = "http://localhost:8785/"
dummy_mode = False
class PumpStatus(IntEnum):
class ParameterEnum(Enum):
@classmethod
def _missing_(cls, value: Any) -> Union['ParameterEnum', None]:
if isinstance(value, str):
# Handle boolean-like strings
if value.lower() in ('true'):
return cls(True)
elif value.lower() in ('false'):
return cls(False)
# Try to convert to int for int-based enums
try:
return cls(int(value))
except ValueError:
pass
# If we can't handle the value, let the default Enum behavior take over
return None
class PumpStatus(ParameterEnum):
INACTIVE = 0
ACTIVE_NO_SPEED_REACHED = 1
ACTIVE_SPEED_REACHED = 2
@ -17,39 +36,57 @@ class PumpStatus(IntEnum):
def __bool__(self):
return self.value in (1, 2)
class PumpDryStatus(IntEnum):
class PumpDryStatus(ParameterEnum):
ACTIVE_WITHOUT_FLUID = 1
INACTIVE_OR_ACTIVE_WITH_FLUID = 4
def __bool__(self):
return self.value == 4
class PumpOverloadStatus(IntEnum):
class PumpOverloadStatus(ParameterEnum):
ACTIVE_AND_OVERLOAD = 1
INACTIVE_OR_ACTIVE_NO_OVERLOAD = 4
def __bool__(self):
return self.value == 4
class BreakerStatus(Enum):
class BreakerStatus(ParameterEnum):
OPEN = True
CLOSED = False
def __bool__(self):
return self.value
CoreState = str
CoolantCoreState = str
RodsState = str
#class CoreState(ParameterEnum):
# REACTIVE = 'REACTIVO'
#
# def __bool__(self):
# return self.value == 'REACTIVO'
#class CoolantCoreState(ParameterEnum):
# CIRCULATING = 'CIRCULANDO'
#
# def __bool__(self):
# return self.value == 'CIRCULANDO'
class Nucon(Enum):
# Core
CORE_TEMP = ("CORE_TEMP", float, False)
CORE_TEMP_OPERATIVE = ("CORE_TEMP_OPERATIVE", float, False)
CORE_TEMP_MAX = ("CORE_TEMP_MAX", float, False)
CORE_TEMP_MIN = ("CORE_TEMP_MIN", float, False)
CORE_TEMP_RESIDUAL = ("CORE_TEMP_RESIDUAL", float, False)
CORE_TEMP_RESIDUAL = ("CORE_TEMP_RESIDUAL", bool, False)
CORE_PRESSURE = ("CORE_PRESSURE", float, False)
CORE_PRESSURE_MAX = ("CORE_PRESSURE_MAX", float, False)
CORE_PRESSURE_OPERATIVE = ("CORE_PRESSURE_OPERATIVE", float, False)
CORE_INTEGRITY = ("CORE_INTEGRITY", float, False)
CORE_WEAR = ("CORE_WEAR", float, False)
CORE_STATE = ("CORE_STATE", int, False)
CORE_STATE = ("CORE_STATE", CoreState, False)
CORE_STATE_CRITICALITY = ("CORE_STATE_CRITICALITY", float, False)
CORE_CRITICAL_MASS_REACHED = ("CORE_CRITICAL_MASS_REACHED", bool, False)
CORE_CRITICAL_MASS_REACHED_COUNTER = ("CORE_CRITICAL_MASS_REACHED_COUNTER", int, False)
@ -58,41 +95,42 @@ class Nucon(Enum):
CORE_STEAM_PRESENT = ("CORE_STEAM_PRESENT", bool, False)
CORE_HIGH_STEAM_PRESENT = ("CORE_HIGH_STEAM_PRESENT", bool, False)
TIME = ("TIME", float, False)
# Time
TIME = ("TIME", str, False)
TIME_STAMP = ("TIME_STAMP", str, False)
COOLANT_CORE_STATE = ("COOLANT_CORE_STATE", int, False)
# Coolant Core
COOLANT_CORE_STATE = ("COOLANT_CORE_STATE", CoolantCoreState, False)
COOLANT_CORE_PRESSURE = ("COOLANT_CORE_PRESSURE", float, False)
COOLANT_CORE_MAX_PRESSURE = ("COOLANT_CORE_MAX_PRESSURE", float, False)
COOLANT_CORE_VESSEL_TEMPERATURE = ("COOLANT_CORE_VESSEL_TEMPERATURE", float, False)
COOLANT_CORE_QUANTITY_IN_VESSEL = ("COOLANT_CORE_QUANTITY_IN_VESSEL", float, False)
COOLANT_CORE_PRIMARY_LOOP_LEVEL = ("COOLANT_CORE_PRIMARY_LOOP_LEVEL", float, False)
COOLANT_CORE_FLOW_SPEED = ("COOLANT_CORE_FLOW_SPEED", float, False)
COOLANT_CORE_FLOW_ORDERED_SPEED = ("COOLANT_CORE_FLOW_ORDERED_SPEED", float, True)
COOLANT_CORE_FLOW_REACHED_SPEED = ("COOLANT_CORE_FLOW_REACHED_SPEED", float, False)
COOLANT_CORE_FLOW_ORDERED_SPEED = ("COOLANT_CORE_FLOW_ORDERED_SPEED", float, False)
COOLANT_CORE_FLOW_REACHED_SPEED = ("COOLANT_CORE_FLOW_REACHED_SPEED", bool, False)
COOLANT_CORE_QUANTITY_CIRCULATION_PUMPS_PRESENT = ("COOLANT_CORE_QUANTITY_CIRCULATION_PUMPS_PRESENT", int, False)
COOLANT_CORE_QUANTITY_FREIGHT_PUMPS_PRESENT = ("COOLANT_CORE_QUANTITY_FREIGHT_PUMPS_PRESENT", int, False)
# Circulation Pumps
COOLANT_CORE_CIRCULATION_PUMP_0_STATUS = ("COOLANT_CORE_CIRCULATION_PUMP_0_STATUS", PumpStatus, False)
COOLANT_CORE_CIRCULATION_PUMP_1_STATUS = ("COOLANT_CORE_CIRCULATION_PUMP_1_STATUS", PumpStatus, False)
COOLANT_CORE_CIRCULATION_PUMP_2_STATUS = ("COOLANT_CORE_CIRCULATION_PUMP_2_STATUS", PumpStatus, False)
COOLANT_CORE_CIRCULATION_PUMP_0_DRY_STATUS = ("COOLANT_CORE_CIRCULATION_PUMP_0_DRY_STATUS", PumpDryStatus, False)
COOLANT_CORE_CIRCULATION_PUMP_1_DRY_STATUS = ("COOLANT_CORE_CIRCULATION_PUMP_1_DRY_STATUS", PumpDryStatus, False)
COOLANT_CORE_CIRCULATION_PUMP_2_DRY_STATUS = ("COOLANT_CORE_CIRCULATION_PUMP_2_DRY_STATUS", PumpDryStatus, False)
COOLANT_CORE_CIRCULATION_PUMP_0_OVERLOAD_STATUS = ("COOLANT_CORE_CIRCULATION_PUMP_0_OVERLOAD_STATUS", PumpOverloadStatus, False)
COOLANT_CORE_CIRCULATION_PUMP_1_OVERLOAD_STATUS = ("COOLANT_CORE_CIRCULATION_PUMP_1_OVERLOAD_STATUS", PumpOverloadStatus, False)
COOLANT_CORE_CIRCULATION_PUMP_2_OVERLOAD_STATUS = ("COOLANT_CORE_CIRCULATION_PUMP_2_OVERLOAD_STATUS", PumpOverloadStatus, False)
COOLANT_CORE_CIRCULATION_PUMP_0_ORDERED_SPEED = ("COOLANT_CORE_CIRCULATION_PUMP_0_ORDERED_SPEED", float, True)
COOLANT_CORE_CIRCULATION_PUMP_1_ORDERED_SPEED = ("COOLANT_CORE_CIRCULATION_PUMP_1_ORDERED_SPEED", float, True)
COOLANT_CORE_CIRCULATION_PUMP_2_ORDERED_SPEED = ("COOLANT_CORE_CIRCULATION_PUMP_2_ORDERED_SPEED", float, True)
COOLANT_CORE_CIRCULATION_PUMP_0_ORDERED_SPEED = ("COOLANT_CORE_CIRCULATION_PUMP_0_ORDERED_SPEED", float, False)
COOLANT_CORE_CIRCULATION_PUMP_1_ORDERED_SPEED = ("COOLANT_CORE_CIRCULATION_PUMP_1_ORDERED_SPEED", float, False)
COOLANT_CORE_CIRCULATION_PUMP_2_ORDERED_SPEED = ("COOLANT_CORE_CIRCULATION_PUMP_2_ORDERED_SPEED", float, False)
COOLANT_CORE_CIRCULATION_PUMP_0_SPEED = ("COOLANT_CORE_CIRCULATION_PUMP_0_SPEED", float, False)
COOLANT_CORE_CIRCULATION_PUMP_1_SPEED = ("COOLANT_CORE_CIRCULATION_PUMP_1_SPEED", float, False)
COOLANT_CORE_CIRCULATION_PUMP_2_SPEED = ("COOLANT_CORE_CIRCULATION_PUMP_2_SPEED", float, False)
RODS_STATUS = ("RODS_STATUS", int, False)
# Rods
RODS_STATUS = ("RODS_STATUS", RodsState, False)
RODS_MOVEMENT_SPEED = ("RODS_MOVEMENT_SPEED", float, False)
RODS_MOVEMENT_SPEED_DECREASED_HIGH_TEMPERATURE = ("RODS_MOVEMENT_SPEED_DECREASED_HIGH_TEMPERATURE", bool, False)
RODS_DEFORMED = ("RODS_DEFORMED", bool, False)
@ -104,6 +142,7 @@ class Nucon(Enum):
RODS_QUANTITY = ("RODS_QUANTITY", int, False)
RODS_ALIGNED = ("RODS_ALIGNED", bool, False)
# Generators
GENERATOR_0_KW = ("GENERATOR_0_KW", float, False)
GENERATOR_1_KW = ("GENERATOR_1_KW", float, False)
GENERATOR_2_KW = ("GENERATOR_2_KW", float, False)
@ -116,10 +155,11 @@ class Nucon(Enum):
GENERATOR_0_HERTZ = ("GENERATOR_0_HERTZ", float, False)
GENERATOR_1_HERTZ = ("GENERATOR_1_HERTZ", float, False)
GENERATOR_2_HERTZ = ("GENERATOR_2_HERTZ", float, False)
GENERATOR_0_BREAKER = ("GENERATOR_0_BREAKER", BreakerStatus, True)
GENERATOR_1_BREAKER = ("GENERATOR_1_BREAKER", BreakerStatus, True)
GENERATOR_2_BREAKER = ("GENERATOR_2_BREAKER", BreakerStatus, True)
GENERATOR_0_BREAKER = ("GENERATOR_0_BREAKER", BreakerStatus, False)
GENERATOR_1_BREAKER = ("GENERATOR_1_BREAKER", BreakerStatus, False)
GENERATOR_2_BREAKER = ("GENERATOR_2_BREAKER", BreakerStatus, False)
# Steam Turbines
STEAM_TURBINE_0_RPM = ("STEAM_TURBINE_0_RPM", float, False)
STEAM_TURBINE_1_RPM = ("STEAM_TURBINE_1_RPM", float, False)
STEAM_TURBINE_2_RPM = ("STEAM_TURBINE_2_RPM", float, False)
@ -137,6 +177,13 @@ class Nucon(Enum):
self.min_val = min_val
self.max_val = max_val
def __repr__(self) -> str:
type_repr = repr(self.param_type)
return (f"<{self.__class__.__name__}.{self.name}: "
f"{{'value': {repr(self.value)}, 'type': {type_repr}, "
f"'writable': {self.is_writable}}}>"
)
def __str__(self):
return self.id
@ -186,41 +233,58 @@ class Nucon(Enum):
return [param for param in cls if param.is_writable]
@classmethod
def get(cls, parameter: Union['Nucon', str]) -> Union[float, int, bool, str, Enum]:
def get(cls, parameter: Union['Nucon', str]) -> Union[float, int, bool, str, ParameterEnum]:
if isinstance(parameter, str):
parameter = cls[parameter]
if NuconConfig.dummy_mode:
return cls._get_dummy_value(parameter)
response = requests.get(NuconConfig.base_url, params={"variable": parameter.name})
if response.status_code != 200:
raise Exception(f"Failed to query parameter {parameter.name}. Status code: {response.status_code}")
value = cls._query(parameter.name)
value = response.text.strip()
if parameter.enum_type:
return parameter.enum_type(int(value))
elif parameter.param_type in (float, int):
return parameter.param_type(value)
if parameter.enum_type and issubclass(parameter.enum_type, ParameterEnum):
return parameter.enum_type(value)
elif parameter.param_type == float:
return float(value)
elif parameter.param_type == int:
return int(value)
elif parameter.param_type == bool:
return value.lower() == "true"
if value.lower() in ('true'):
return True
elif value.lower() in ('false'):
return False
else:
raise ValueError(f"Invalid boolean value: {value}. Expected 'TRUE' or 'FALSE' (case insensitive).")
else:
return value
@classmethod
def _query(cls, parameter_name: str) -> str:
response = requests.get(NuconConfig.base_url, params={"variable": parameter_name})
if response.status_code != 200:
raise Exception(f"Failed to query parameter {parameter_name}. Status code: {response.status_code}")
return response.text.strip()
@classmethod
def _get_dummy_value(cls, parameter: 'Nucon') -> Union[float, int, bool, str, Enum]:
if parameter.enum_type:
return next(iter(parameter.enum_type))
elif parameter.param_type == float:
return 0.0
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:
return 0
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 False
else:
return ""
return "dummy"
@classmethod
def get_multiple(cls, parameters: List['Nucon']) -> Dict['Nucon', Union[float, int, bool, str, Enum]]:

View File

@ -25,7 +25,7 @@ dependencies = [
]
[project.urls]
Homepage = ""
Homepage = "https://git.dominik-roth.eu/dodox/nucon"
[project.optional-dependencies]
dev = ["pytest"]

View File

@ -2,10 +2,12 @@ import pytest
import warnings
from nucon import Nucon, NuconConfig, PumpStatus, PumpDryStatus, PumpOverloadStatus, BreakerStatus
WARN_FLOAT_COULD_BE_INT = False
@pytest.fixture(scope="module")
def nucon_setup():
Nucon.set_dummy_mode(False) # Assume the game is running
Nucon.set_base_url("http://localhost:8080/")
Nucon.set_base_url("http://localhost:8785/")
yield
Nucon.set_dummy_mode(True)
@ -14,7 +16,7 @@ def test_read_all_parameters(nucon_setup):
assert len(all_params) == len(Nucon)
for param, value in all_params.items():
assert isinstance(value, param.param_type), f"Parameter {param.name} has incorrect type. Expected {param.param_type}, got {type(value)}"
if param.param_type == float and value.is_integer():
if param.param_type == float and value.is_integer() and WARN_FLOAT_COULD_BE_INT:
warnings.warn(f"Parameter {param.name} is a float but has an integer value: {value}")
def test_write_writable_parameters(nucon_setup):