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 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:
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: 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:
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: 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}'")

View File

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