Compare commits
3 Commits
08ecbb461d
...
d5a0212824
Author | SHA1 | Date | |
---|---|---|---|
d5a0212824 | |||
9adee35a6f | |||
a6b5b68777 |
@ -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
|
||||||
|
@ -105,7 +105,7 @@ class NuconParameter:
|
|||||||
return self.id
|
return self.id
|
||||||
|
|
||||||
class Nucon:
|
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.base_url = f'http://{host}:{port}/'
|
||||||
self.dummy_mode = False
|
self.dummy_mode = False
|
||||||
self._parameters = self._create_parameters()
|
self._parameters = self._create_parameters()
|
||||||
@ -159,16 +159,16 @@ class Nucon:
|
|||||||
'COOLANT_CORE_CIRCULATION_PUMP_1_SPEED': (float, False),
|
'COOLANT_CORE_CIRCULATION_PUMP_1_SPEED': (float, False),
|
||||||
'COOLANT_CORE_CIRCULATION_PUMP_2_SPEED': (float, False),
|
'COOLANT_CORE_CIRCULATION_PUMP_2_SPEED': (float, False),
|
||||||
'RODS_STATUS': (RodsState, False),
|
'RODS_STATUS': (RodsState, False),
|
||||||
'RODS_MOVEMENT_SPEED': (float, False),
|
#'RODS_MOVEMENT_SPEED': (float, False),
|
||||||
'RODS_MOVEMENT_SPEED_DECREASED_HIGH_TEMPERATURE': (bool, False),
|
#'RODS_MOVEMENT_SPEED_DECREASED_HIGH_TEMPERATURE': (bool, False),
|
||||||
'RODS_DEFORMED': (bool, False),
|
#'RODS_DEFORMED': (bool, False),
|
||||||
'RODS_TEMPERATURE': (float, False),
|
#'RODS_TEMPERATURE': (float, False),
|
||||||
'RODS_MAX_TEMPERATURE': (float, False),
|
#'RODS_MAX_TEMPERATURE': (float, False),
|
||||||
'RODS_POS_ORDERED': (float, True),
|
#'RODS_POS_ORDERED': (float, True),
|
||||||
'RODS_POS_ACTUAL': (float, False),
|
#'RODS_POS_ACTUAL': (float, False),
|
||||||
'RODS_POS_REACHED': (bool, False),
|
#'RODS_POS_REACHED': (bool, False),
|
||||||
'RODS_QUANTITY': (int, False),
|
#'RODS_QUANTITY': (int, False),
|
||||||
'RODS_ALIGNED': (bool, False),
|
#'RODS_ALIGNED': (bool, False),
|
||||||
'GENERATOR_0_KW': (float, False),
|
'GENERATOR_0_KW': (float, False),
|
||||||
'GENERATOR_1_KW': (float, False),
|
'GENERATOR_1_KW': (float, False),
|
||||||
'GENERATOR_2_KW': (float, False),
|
'GENERATOR_2_KW': (float, False),
|
||||||
@ -210,16 +210,22 @@ class Nucon:
|
|||||||
value = self._query(parameter)
|
value = self._query(parameter)
|
||||||
|
|
||||||
if parameter.enum_type:
|
if parameter.enum_type:
|
||||||
|
try:
|
||||||
return parameter.enum_type(value)
|
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:
|
elif parameter.param_type == bool:
|
||||||
if isinstance(value, str):
|
if isinstance(value, str):
|
||||||
if value.lower() not in ('true', 'false'):
|
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'
|
return value.lower() == 'true'
|
||||||
else:
|
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:
|
else:
|
||||||
|
try:
|
||||||
return parameter.param_type(value)
|
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:
|
def set(self, parameter: Union[str, NuconParameter], value: Union[float, int, bool, str, Enum], force: bool = False) -> None:
|
||||||
if isinstance(parameter, str):
|
if isinstance(parameter, str):
|
||||||
@ -240,12 +246,20 @@ class Nucon:
|
|||||||
|
|
||||||
self._set_value(parameter, str(value))
|
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:
|
def _query(self, parameter: NuconParameter) -> str:
|
||||||
response = requests.get(self.base_url, params={"variable": parameter.id})
|
response = requests.get(self.base_url, params={"variable": parameter.id})
|
||||||
|
|
||||||
if response.status_code != 200:
|
if response.status_code != 200:
|
||||||
raise Exception(f"Failed to query parameter {parameter.id}. Status code: {response.status_code}")
|
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()
|
return response.text.strip()
|
||||||
|
|
||||||
def _set_value(self, parameter: NuconParameter, value: str) -> None:
|
def _set_value(self, parameter: NuconParameter, value: str) -> None:
|
||||||
@ -301,6 +315,8 @@ class Nucon:
|
|||||||
self.dummy_mode = dummy_mode
|
self.dummy_mode = dummy_mode
|
||||||
|
|
||||||
def __getattr__(self, name):
|
def __getattr__(self, name):
|
||||||
|
if isinstance(name, int):
|
||||||
|
return self.__getattr__(list(self._parameters.keys())[name])
|
||||||
if name in self._parameters:
|
if name in self._parameters:
|
||||||
return self._parameters[name]
|
return self._parameters[name]
|
||||||
raise AttributeError(f"'{self.__class__.__name__}' object has no attribute '{name}'")
|
raise AttributeError(f"'{self.__class__.__name__}' object has no attribute '{name}'")
|
||||||
|
@ -4,39 +4,41 @@ from nucon import Nucon, PumpStatus, PumpDryStatus, PumpOverloadStatus, BreakerS
|
|||||||
|
|
||||||
WARN_FLOAT_COULD_BE_INT = False
|
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):
|
@pytest.fixture(scope="function")
|
||||||
all_params = Nucon.get_all()
|
def nucon():
|
||||||
assert len(all_params) == len(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():
|
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)}"
|
param_type = nucon.get_type(param)
|
||||||
if param.param_type == float and value.is_integer() and WARN_FLOAT_COULD_BE_INT:
|
assert isinstance(value, param_type), f"Parameter {param.id} has incorrect type. Expected {param_type}, got {type(value)}"
|
||||||
warnings.warn(f"Parameter {param.id} is a float but has an integer value: {value}")
|
if param_type == float and value.is_integer() and WARN_FLOAT_COULD_BE_INT:
|
||||||
if param.param_type == str:
|
warnings.warn(f"Parameter {param} is a float but has an integer value: {value}")
|
||||||
|
if param_type == str:
|
||||||
try:
|
try:
|
||||||
float(value)
|
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:
|
except ValueError:
|
||||||
pass
|
pass
|
||||||
try:
|
try:
|
||||||
bool(value.lower())
|
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:
|
except ValueError:
|
||||||
pass
|
pass
|
||||||
|
|
||||||
def test_write_writable_parameters(nucon_setup):
|
def test_write_writable_parameters(nucon):
|
||||||
writable_params = Nucon.get_all_writable()
|
writable_params = nucon.get_all_writable()
|
||||||
for param in writable_params:
|
for param in writable_params:
|
||||||
current_value = param.value
|
current_value = param.value
|
||||||
param.value = current_value
|
param.value = current_value
|
||||||
assert param.value == current_value, f"Failed to write to parameter {param.id}"
|
assert param.value == current_value, f"Failed to write to parameter {param.id}"
|
||||||
|
|
||||||
def test_non_writable_parameters(nucon_setup):
|
def test_non_writable_parameters(nucon):
|
||||||
non_writable_params = [param for param in Nucon if not param.is_writable]
|
non_writable_params = [param for param in nucon if not param.is_writable]
|
||||||
for param in non_writable_params:
|
for param in non_writable_params:
|
||||||
# Test that normal set raises an error
|
# Test that normal set raises an error
|
||||||
with pytest.raises(ValueError, match=f"Parameter {param.id} is not writable"):
|
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
|
# Test that force_set is refused by the webserver
|
||||||
current_value = param.value
|
current_value = param.value
|
||||||
with pytest.raises(Exception, match=f"Failed to set parameter {param.id}"):
|
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):
|
def test_enum_parameters(nucon):
|
||||||
pump_status = Nucon.COOLANT_CORE_CIRCULATION_PUMP_0_STATUS.value
|
pump_status = nucon.COOLANT_CORE_CIRCULATION_PUMP_0_STATUS.value
|
||||||
assert isinstance(pump_status, PumpStatus)
|
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)
|
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)
|
assert isinstance(overload_status, PumpOverloadStatus)
|
||||||
|
|
||||||
breaker_status = Nucon.GENERATOR_0_BREAKER.value
|
breaker_status = nucon.GENERATOR_0_BREAKER.value
|
||||||
assert isinstance(breaker_status, BreakerStatus)
|
assert isinstance(breaker_status, BreakerStatus)
|
||||||
|
|
||||||
def test_custom_truthy_values(nucon_setup):
|
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_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_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.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)
|
assert bool(nucon.GENERATOR_0_BREAKER.value) == (nucon.GENERATOR_0_BREAKER.value == BreakerStatus.OPEN)
|
||||||
|
|
||||||
def test_get_multiple_parameters(nucon_setup):
|
def test_get_multiple_parameters(nucon):
|
||||||
params_to_get = [Nucon.CORE_TEMP, Nucon.CORE_PRESSURE, Nucon.RODS_POS_ACTUAL]
|
params_to_get = [nucon.CORE_TEMP, nucon.CORE_PRESSURE, nucon.TIME]
|
||||||
multiple_params = Nucon.get_multiple(params_to_get)
|
multiple_params = nucon.get_multiple(params_to_get)
|
||||||
assert len(multiple_params) == len(params_to_get)
|
assert len(multiple_params) == len(params_to_get)
|
||||||
for param, value in multiple_params.items():
|
for param, value in multiple_params.items():
|
||||||
assert isinstance(value, param.param_type)
|
param_type = nucon.get_type(param)
|
||||||
|
assert isinstance(value, 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)
|
|
||||||
|
|
||||||
if __name__ == "__main__":
|
if __name__ == "__main__":
|
||||||
pytest.main()
|
pytest.main()
|
Loading…
Reference in New Issue
Block a user