mirror of
https://github.com/tiqi-group/pydase.git
synced 2025-04-23 01:20:03 +02:00
adds support for dynamically adding attributes to DataService instances
This commit is contained in:
Mose Müller
parent
b4c84da57e
commit
9916d6df60
@ -65,23 +65,23 @@ class DataServiceObserver(PropertyObserver):
|
||||
cached_value_dict: SerializedObject | dict[str, Any],
|
||||
) -> None:
|
||||
value_dict = dump(value)
|
||||
if cached_value_dict != {}:
|
||||
if (
|
||||
cached_value_dict["type"] != "method"
|
||||
and cached_value_dict["type"] != value_dict["type"]
|
||||
):
|
||||
logger.warning(
|
||||
"Type of '%s' changed from '%s' to '%s'. This could have unwanted "
|
||||
"side effects! Consider setting it to '%s' directly.",
|
||||
full_access_path,
|
||||
cached_value_dict["type"],
|
||||
value_dict["type"],
|
||||
cached_value_dict["type"],
|
||||
)
|
||||
self.state_manager._data_service_cache.update_cache(
|
||||
if (
|
||||
cached_value_dict != {}
|
||||
and cached_value_dict["type"] != "method"
|
||||
and cached_value_dict["type"] != value_dict["type"]
|
||||
):
|
||||
logger.warning(
|
||||
"Type of '%s' changed from '%s' to '%s'. This could have unwanted "
|
||||
"side effects! Consider setting it to '%s' directly.",
|
||||
full_access_path,
|
||||
value,
|
||||
cached_value_dict["type"],
|
||||
value_dict["type"],
|
||||
cached_value_dict["type"],
|
||||
)
|
||||
self.state_manager._data_service_cache.update_cache(
|
||||
full_access_path,
|
||||
value,
|
||||
)
|
||||
|
||||
def _notify_dependent_property_changes(self, changed_attr_path: str) -> None:
|
||||
changed_props = self.property_deps_dict.get(changed_attr_path, [])
|
||||
|
||||
@ -346,8 +346,7 @@ def set_nested_value_by_path(
|
||||
"RUNNING" if isinstance(value, TaskStatus) else None
|
||||
)
|
||||
else:
|
||||
serialized_value = dump(value)
|
||||
serialized_value["full_access_path"] = path
|
||||
serialized_value = Serializer.serialize_object(value, access_path=path)
|
||||
serialized_value["readonly"] = next_level_serialized_object["readonly"]
|
||||
|
||||
keys_to_keep = set(serialized_value.keys())
|
||||
@ -437,11 +436,20 @@ def get_next_level_dict_by_key(
|
||||
f"Error occured trying to change '{attr_name}[{index}]': {e}"
|
||||
)
|
||||
except KeyError:
|
||||
raise SerializationPathError(
|
||||
f"Error occured trying to access the key '{attr_name}': it is either "
|
||||
"not present in the current dictionary or its value does not contain "
|
||||
"a 'value' key."
|
||||
)
|
||||
if not allow_append:
|
||||
raise SerializationPathError(
|
||||
f"Error occured trying to access the key '{attr_name}': it is either "
|
||||
"not present in the current dictionary or its value does not contain "
|
||||
"a 'value' key."
|
||||
)
|
||||
serialization_dict[attr_name] = {
|
||||
"full_access_path": "",
|
||||
"value": None,
|
||||
"type": "None",
|
||||
"doc": None,
|
||||
"readonly": False,
|
||||
}
|
||||
next_level_serialized_object = serialization_dict[attr_name]
|
||||
|
||||
if not isinstance(next_level_serialized_object, dict):
|
||||
raise SerializationValueError(
|
||||
|
||||
@ -570,17 +570,6 @@ def test_update_invalid_list_index(
|
||||
)
|
||||
|
||||
|
||||
def test_update_invalid_path(
|
||||
setup_dict: dict[str, Any], caplog: pytest.LogCaptureFixture
|
||||
) -> None:
|
||||
set_nested_value_by_path(setup_dict, "invalid_path", 30)
|
||||
assert (
|
||||
"Error occured trying to access the key 'invalid_path': it is either "
|
||||
"not present in the current dictionary or its value does not contain "
|
||||
"a 'value' key." in caplog.text
|
||||
)
|
||||
|
||||
|
||||
def test_update_list_inside_class(setup_dict: dict[str, Any]) -> None:
|
||||
set_nested_value_by_path(setup_dict, "attr2.list_attr[1]", 40)
|
||||
assert setup_dict["attr2"]["value"]["list_attr"]["value"][1]["value"] == 40 # noqa
|
||||
@ -743,3 +732,142 @@ def test_serialized_dict_is_nested_object() -> None:
|
||||
assert not serialized_dict_is_nested_object(serialized_dict["unit"])
|
||||
assert not serialized_dict_is_nested_object(serialized_dict["float"])
|
||||
assert not serialized_dict_is_nested_object(serialized_dict["state"])
|
||||
|
||||
|
||||
class MyService(pydase.DataService):
|
||||
name = "MyService"
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"test_input, expected",
|
||||
[
|
||||
(
|
||||
1,
|
||||
{
|
||||
"new_attr": {
|
||||
"full_access_path": "new_attr",
|
||||
"type": "int",
|
||||
"value": 1,
|
||||
"readonly": False,
|
||||
"doc": None,
|
||||
}
|
||||
},
|
||||
),
|
||||
(
|
||||
1.0,
|
||||
{
|
||||
"new_attr": {
|
||||
"full_access_path": "new_attr",
|
||||
"type": "float",
|
||||
"value": 1.0,
|
||||
"readonly": False,
|
||||
"doc": None,
|
||||
},
|
||||
},
|
||||
),
|
||||
(
|
||||
True,
|
||||
{
|
||||
"new_attr": {
|
||||
"full_access_path": "new_attr",
|
||||
"type": "bool",
|
||||
"value": True,
|
||||
"readonly": False,
|
||||
"doc": None,
|
||||
},
|
||||
},
|
||||
),
|
||||
(
|
||||
u.Quantity(10, "m"),
|
||||
{
|
||||
"new_attr": {
|
||||
"full_access_path": "new_attr",
|
||||
"type": "Quantity",
|
||||
"value": {"magnitude": 10, "unit": "meter"},
|
||||
"readonly": False,
|
||||
"doc": None,
|
||||
},
|
||||
},
|
||||
),
|
||||
(
|
||||
MyEnum.RUNNING,
|
||||
{
|
||||
"new_attr": {
|
||||
"full_access_path": "new_attr",
|
||||
"value": "RUNNING",
|
||||
"type": "Enum",
|
||||
"doc": "MyEnum description",
|
||||
"readonly": False,
|
||||
"name": "MyEnum",
|
||||
"enum": {"RUNNING": "running", "FINISHED": "finished"},
|
||||
}
|
||||
},
|
||||
),
|
||||
(
|
||||
[1.0],
|
||||
{
|
||||
"new_attr": {
|
||||
"full_access_path": "new_attr",
|
||||
"value": [
|
||||
{
|
||||
"full_access_path": "new_attr[0]",
|
||||
"doc": None,
|
||||
"readonly": False,
|
||||
"type": "float",
|
||||
"value": 1.0,
|
||||
}
|
||||
],
|
||||
"type": "list",
|
||||
"doc": None,
|
||||
"readonly": False,
|
||||
}
|
||||
},
|
||||
),
|
||||
(
|
||||
{"key": 1.0},
|
||||
{
|
||||
"new_attr": {
|
||||
"full_access_path": "new_attr",
|
||||
"value": {
|
||||
"key": {
|
||||
"full_access_path": 'new_attr["key"]',
|
||||
"doc": None,
|
||||
"readonly": False,
|
||||
"type": "float",
|
||||
"value": 1.0,
|
||||
}
|
||||
},
|
||||
"type": "dict",
|
||||
"doc": None,
|
||||
"readonly": False,
|
||||
}
|
||||
},
|
||||
),
|
||||
(
|
||||
MyService(),
|
||||
{
|
||||
"new_attr": {
|
||||
"full_access_path": "new_attr",
|
||||
"value": {
|
||||
"name": {
|
||||
"full_access_path": "new_attr.name",
|
||||
"doc": None,
|
||||
"readonly": False,
|
||||
"type": "str",
|
||||
"value": "MyService",
|
||||
}
|
||||
},
|
||||
"type": "DataService",
|
||||
"doc": None,
|
||||
"readonly": False,
|
||||
"name": "MyService",
|
||||
}
|
||||
},
|
||||
),
|
||||
],
|
||||
)
|
||||
def test_dynamically_add_attributes(test_input: Any, expected: dict[str, Any]) -> None:
|
||||
serialized_object: dict[str, SerializedObject] = {}
|
||||
|
||||
set_nested_value_by_path(serialized_object, "new_attr", test_input)
|
||||
assert serialized_object == expected
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user