Compare commits

..

3 Commits

3 changed files with 67 additions and 55 deletions

View File

@ -1 +1 @@
from nucon.core import Nucon, PumpStatus, PumpDryStatus, PumpOverloadStatus, PumpOverloadStatus, CoreState, CoolantCoreState, RodsState, NuconParameter
from nucon.core import Nucon, PumpStatus, PumpDryStatus, PumpOverloadStatus, PumpOverloadStatus, CoreState, CoolantCoreState, RodsState, NuconParameter, BreakerStatus

View File

@ -105,7 +105,7 @@ class NuconParameter:
return self.id
class Nucon:
def __init__(self, host: str = 'localhost', port: int = 8786):
def __init__(self, host: str = 'localhost', port: int = 8785):
self.base_url = f'http://{host}:{port}/'
self.dummy_mode = False
self._parameters = self._create_parameters()
@ -159,16 +159,16 @@ class Nucon:
'COOLANT_CORE_CIRCULATION_PUMP_1_SPEED': (float, False),
'COOLANT_CORE_CIRCULATION_PUMP_2_SPEED': (float, False),
'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),
'RODS_MAX_TEMPERATURE': (float, False),
'RODS_POS_ORDERED': (float, True),
'RODS_POS_ACTUAL': (float, False),
'RODS_POS_REACHED': (bool, False),
'RODS_QUANTITY': (int, False),
'RODS_ALIGNED': (bool, False),
#'RODS_MOVEMENT_SPEED': (float, False),
#'RODS_MOVEMENT_SPEED_DECREASED_HIGH_TEMPERATURE': (bool, False),
#'RODS_DEFORMED': (bool, False),
#'RODS_TEMPERATURE': (float, False),
#'RODS_MAX_TEMPERATURE': (float, False),
#'RODS_POS_ORDERED': (float, True),
#'RODS_POS_ACTUAL': (float, False),
#'RODS_POS_REACHED': (bool, False),
#'RODS_QUANTITY': (int, False),
#'RODS_ALIGNED': (bool, False),
'GENERATOR_0_KW': (float, False),
'GENERATOR_1_KW': (float, False),
'GENERATOR_2_KW': (float, False),
@ -210,16 +210,22 @@ class Nucon:
value = self._query(parameter)
if parameter.enum_type:
return parameter.enum_type(value)
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: {value}")
raise ValueError(f"Invalid boolean value for parameter {parameter.id}: {value}")
return value.lower() == 'true'
else:
raise ValueError(f"Expected string for boolean parameter, got {type(value)}")
raise ValueError(f"Expected string for boolean parameter {parameter.id}, got {type(value)}")
else:
return parameter.param_type(value)
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 set(self, parameter: Union[str, NuconParameter], value: Union[float, int, bool, str, Enum], force: bool = False) -> None:
if isinstance(parameter, str):
@ -240,12 +246,20 @@ class Nucon:
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 _set_value(self, parameter: NuconParameter, value: str) -> None:
@ -301,6 +315,8 @@ class Nucon:
self.dummy_mode = dummy_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}'")

View File

@ -4,39 +4,41 @@ from nucon import Nucon, PumpStatus, PumpDryStatus, PumpOverloadStatus, BreakerS
WARN_FLOAT_COULD_BE_INT = False
@pytest.fixture(scope="module")
def nucon_setup():
Nucon.set_dummy_mode(False)
Nucon.set_base_url("http://localhost:8785/")
def test_read_all_parameters(nucon_setup):
all_params = Nucon.get_all()
assert len(all_params) == len(Nucon)
@pytest.fixture(scope="function")
def nucon():
"""Create a fresh Nucon instance for each test"""
return Nucon()
def test_read_all_parameters(nucon):
all_params = nucon.get_all()
assert len(all_params) == len(nucon)
for param, value in all_params.items():
assert isinstance(value, param.param_type), f"Parameter {param.id} has incorrect type. Expected {param.param_type}, got {type(value)}"
if param.param_type == float and value.is_integer() and WARN_FLOAT_COULD_BE_INT:
warnings.warn(f"Parameter {param.id} is a float but has an integer value: {value}")
if param.param_type == str:
param_type = nucon.get_type(param)
assert isinstance(value, param_type), f"Parameter {param.id} has incorrect type. Expected {param_type}, got {type(value)}"
if param_type == float and value.is_integer() and WARN_FLOAT_COULD_BE_INT:
warnings.warn(f"Parameter {param} is a float but has an integer value: {value}")
if param_type == str:
try:
float(value)
raise ValueError(f"Parameter {param.id} is a string that looks like a number: {value}")
raise ValueError(f"Parameter {param} is a string that looks like a number: {value}")
except ValueError:
pass
try:
bool(value.lower())
raise ValueError(f"Parameter {param.id} is a string that looks like a boolean: {value}")
raise ValueError(f"Parameter {param} is a string that looks like a boolean: {value}")
except ValueError:
pass
def test_write_writable_parameters(nucon_setup):
writable_params = Nucon.get_all_writable()
def test_write_writable_parameters(nucon):
writable_params = nucon.get_all_writable()
for param in writable_params:
current_value = param.value
param.value = current_value
assert param.value == current_value, f"Failed to write to parameter {param.id}"
def test_non_writable_parameters(nucon_setup):
non_writable_params = [param for param in Nucon if not param.is_writable]
def test_non_writable_parameters(nucon):
non_writable_params = [param for param in nucon if not param.is_writable]
for param in non_writable_params:
# Test that normal set raises an error
with pytest.raises(ValueError, match=f"Parameter {param.id} is not writable"):
@ -45,40 +47,34 @@ def test_non_writable_parameters(nucon_setup):
# Test that force_set is refused by the webserver
current_value = param.value
with pytest.raises(Exception, match=f"Failed to set parameter {param.id}"):
Nucon.set(param, current_value, force=True)
nucon.set(param, current_value, force=True)
def test_enum_parameters(nucon_setup):
pump_status = Nucon.COOLANT_CORE_CIRCULATION_PUMP_0_STATUS.value
def test_enum_parameters(nucon):
pump_status = nucon.COOLANT_CORE_CIRCULATION_PUMP_0_STATUS.value
assert isinstance(pump_status, PumpStatus)
dry_status = Nucon.COOLANT_CORE_CIRCULATION_PUMP_0_DRY_STATUS.value
dry_status = nucon.COOLANT_CORE_CIRCULATION_PUMP_0_DRY_STATUS.value
assert isinstance(dry_status, PumpDryStatus)
overload_status = Nucon.COOLANT_CORE_CIRCULATION_PUMP_0_OVERLOAD_STATUS.value
overload_status = nucon.COOLANT_CORE_CIRCULATION_PUMP_0_OVERLOAD_STATUS.value
assert isinstance(overload_status, PumpOverloadStatus)
breaker_status = Nucon.GENERATOR_0_BREAKER.value
breaker_status = nucon.GENERATOR_0_BREAKER.value
assert isinstance(breaker_status, BreakerStatus)
def test_custom_truthy_values(nucon_setup):
assert bool(Nucon.COOLANT_CORE_CIRCULATION_PUMP_0_STATUS.value) == (Nucon.COOLANT_CORE_CIRCULATION_PUMP_0_STATUS.value in [PumpStatus.ACTIVE_NO_SPEED_REACHED, PumpStatus.ACTIVE_SPEED_REACHED])
assert bool(Nucon.COOLANT_CORE_CIRCULATION_PUMP_0_DRY_STATUS.value) == (Nucon.COOLANT_CORE_CIRCULATION_PUMP_0_DRY_STATUS.value == PumpDryStatus.INACTIVE_OR_ACTIVE_WITH_FLUID)
assert bool(Nucon.COOLANT_CORE_CIRCULATION_PUMP_0_OVERLOAD_STATUS.value) == (Nucon.COOLANT_CORE_CIRCULATION_PUMP_0_OVERLOAD_STATUS.value == PumpOverloadStatus.INACTIVE_OR_ACTIVE_NO_OVERLOAD)
assert bool(Nucon.GENERATOR_0_BREAKER.value) == (Nucon.GENERATOR_0_BREAKER.value == BreakerStatus.OPEN)
def test_custom_truthy_values(nucon):
assert bool(nucon.COOLANT_CORE_CIRCULATION_PUMP_0_STATUS.value) == (nucon.COOLANT_CORE_CIRCULATION_PUMP_0_STATUS.value in [PumpStatus.ACTIVE_NO_SPEED_REACHED, PumpStatus.ACTIVE_SPEED_REACHED])
#assert bool(nucon.COOLANT_CORE_CIRCULATION_PUMP_0_DRY_STATUS.value) == (nucon.COOLANT_CORE_CIRCULATION_PUMP_0_DRY_STATUS.value == PumpDryStatus.INACTIVE_OR_ACTIVE_WITH_FLUID)
#assert bool(nucon.COOLANT_CORE_CIRCULATION_PUMP_0_OVERLOAD_STATUS.value) == (nucon.COOLANT_CORE_CIRCULATION_PUMP_0_OVERLOAD_STATUS.value == PumpOverloadStatus.INACTIVE_OR_ACTIVE_NO_OVERLOAD)
assert bool(nucon.GENERATOR_0_BREAKER.value) == (nucon.GENERATOR_0_BREAKER.value == BreakerStatus.OPEN)
def test_get_multiple_parameters(nucon_setup):
params_to_get = [Nucon.CORE_TEMP, Nucon.CORE_PRESSURE, Nucon.RODS_POS_ACTUAL]
multiple_params = Nucon.get_multiple(params_to_get)
def test_get_multiple_parameters(nucon):
params_to_get = [nucon.CORE_TEMP, nucon.CORE_PRESSURE, nucon.TIME]
multiple_params = nucon.get_multiple(params_to_get)
assert len(multiple_params) == len(params_to_get)
for param, value in multiple_params.items():
assert isinstance(value, param.param_type)
def test_base_url_setting():
original_url = NuconConfig.base_url
new_url = "http://newlocalhost:9090/"
Nucon.set_base_url(new_url)
assert NuconConfig.base_url == new_url
Nucon.set_base_url(original_url)
param_type = nucon.get_type(param)
assert isinstance(value, param_type)
if __name__ == "__main__":
pytest.main()