updates generate_serialized_data_paths method, adds tests

This commit is contained in:
Mose Müller 2023-12-20 10:15:25 +01:00
parent e106cc4927
commit 32bda8d910
2 changed files with 145 additions and 33 deletions

View File

@ -346,17 +346,19 @@ def generate_serialized_data_paths(
""" """
Generate a list of access paths for all attributes in a dictionary representing Generate a list of access paths for all attributes in a dictionary representing
data serialized with `pydase.utils.serializer.Serializer`, excluding those that are data serialized with `pydase.utils.serializer.Serializer`, excluding those that are
methods. methods. This function handles nested structures, including lists, by generating
paths for each element in the nested lists.
Args: Args:
data: The dictionary representing serialized data, typically produced by data (dict[str, Any]): The dictionary representing serialized data, typically
`pydase.utils.serializer.Serializer`. produced by `pydase.utils.serializer.Serializer`.
parent_path: The base path to prepend to the keys in the `data` dictionary to parent_path (str, optional): The base path to prepend to the keys in the `data`
form the access paths. Defaults to an empty string. dictionary to form the access paths. Defaults to an empty string.
Returns: Returns:
A list of strings where each string is a dot-notation access path to an list[str]: A list of strings where each string is a dot-notation access path
attribute in the serialized data. to an attribute in the serialized data. For list elements, the path includes
the index in square brackets.
""" """
paths: list[str] = [] paths: list[str] = []
@ -365,17 +367,25 @@ def generate_serialized_data_paths(
# ignoring methods # ignoring methods
continue continue
new_path = f"{parent_path}.{key}" if parent_path else key new_path = f"{parent_path}.{key}" if parent_path else key
if isinstance(value["value"], dict) and value["type"] != "Quantity": paths.append(new_path)
paths.extend(generate_serialized_data_paths(value["value"], new_path)) if serialized_dict_is_nested_object(value):
elif isinstance(value["value"], list): if isinstance(value["value"], list):
for index, item in enumerate(value["value"]): for index, item in enumerate(value["value"]):
indexed_key_path = f"{new_path}[{index}]" indexed_key_path = f"{new_path}[{index}]"
if isinstance(item["value"], dict):
paths.extend(
generate_serialized_data_paths(item["value"], indexed_key_path)
)
else:
paths.append(indexed_key_path) paths.append(indexed_key_path)
else: if serialized_dict_is_nested_object(item):
paths.append(new_path) paths.extend(
generate_serialized_data_paths(
item["value"], indexed_key_path
)
)
continue
paths.extend(generate_serialized_data_paths(value["value"], new_path))
return paths return paths
def serialized_dict_is_nested_object(serialized_dict: dict[str, Any]) -> bool:
return (
serialized_dict["type"] != "Quantity"
and isinstance(serialized_dict["value"], dict)
) or isinstance(serialized_dict["value"], list)

View File

@ -11,6 +11,7 @@ from pydase.utils.serializer import (
dump, dump,
get_nested_dict_by_path, get_nested_dict_by_path,
get_next_level_dict_by_key, get_next_level_dict_by_key,
serialized_dict_is_nested_object,
set_nested_value_by_path, set_nested_value_by_path,
) )
@ -360,7 +361,9 @@ def test_update_invalid_list_index(setup_dict, caplog: pytest.LogCaptureFixture)
) )
def test_update_invalid_path(setup_dict, caplog: pytest.LogCaptureFixture): def test_update_invalid_path(
setup_dict: dict[str, Any], caplog: pytest.LogCaptureFixture
) -> None:
set_nested_value_by_path(setup_dict, "invalid_path", 30) set_nested_value_by_path(setup_dict, "invalid_path", 30)
assert ( assert (
"Error occured trying to access the key 'invalid_path': it is either " "Error occured trying to access the key 'invalid_path': it is either "
@ -369,66 +372,165 @@ def test_update_invalid_path(setup_dict, caplog: pytest.LogCaptureFixture):
) )
def test_update_list_inside_class(setup_dict): def test_update_list_inside_class(setup_dict: dict[str, Any]) -> None:
set_nested_value_by_path(setup_dict, "attr2.list_attr[1]", 40) set_nested_value_by_path(setup_dict, "attr2.list_attr[1]", 40)
assert setup_dict["attr2"]["value"]["list_attr"]["value"][1]["value"] == 40 assert setup_dict["attr2"]["value"]["list_attr"]["value"][1]["value"] == 40
def test_update_class_attribute_inside_list(setup_dict): def test_update_class_attribute_inside_list(setup_dict: dict[str, Any]) -> None:
set_nested_value_by_path(setup_dict, "attr_list[2].attr3", 50) set_nested_value_by_path(setup_dict, "attr_list[2].attr3", 50)
assert setup_dict["attr_list"]["value"][2]["value"]["attr3"]["value"] == 50 assert setup_dict["attr_list"]["value"][2]["value"]["attr3"]["value"] == 50
def test_get_next_level_attribute_nested_dict(setup_dict): def test_get_next_level_attribute_nested_dict(setup_dict: dict[str, Any]) -> None:
nested_dict = get_next_level_dict_by_key(setup_dict, "attr1") nested_dict = get_next_level_dict_by_key(setup_dict, "attr1")
assert nested_dict == setup_dict["attr1"] assert nested_dict == setup_dict["attr1"]
def test_get_next_level_list_entry_nested_dict(setup_dict): def test_get_next_level_list_entry_nested_dict(setup_dict: dict[str, Any]) -> None:
nested_dict = get_next_level_dict_by_key(setup_dict, "attr_list[0]") nested_dict = get_next_level_dict_by_key(setup_dict, "attr_list[0]")
assert nested_dict == setup_dict["attr_list"]["value"][0] assert nested_dict == setup_dict["attr_list"]["value"][0]
def test_get_next_level_invalid_path_nested_dict(setup_dict): def test_get_next_level_invalid_path_nested_dict(setup_dict: dict[str, Any]) -> None:
with pytest.raises(SerializationPathError): with pytest.raises(SerializationPathError):
get_next_level_dict_by_key(setup_dict, "invalid_path") get_next_level_dict_by_key(setup_dict, "invalid_path")
def test_get_next_level_invalid_list_index(setup_dict): def test_get_next_level_invalid_list_index(setup_dict: dict[str, Any]) -> None:
with pytest.raises(SerializationPathError): with pytest.raises(SerializationPathError):
get_next_level_dict_by_key(setup_dict, "attr_list[10]") get_next_level_dict_by_key(setup_dict, "attr_list[10]")
def test_get_attribute(setup_dict): def test_get_attribute(setup_dict: dict[str, Any]) -> None:
nested_dict = get_nested_dict_by_path(setup_dict, "attr1") nested_dict = get_nested_dict_by_path(setup_dict, "attr1")
assert nested_dict["value"] == 1.0 assert nested_dict["value"] == 1.0
def test_get_nested_attribute(setup_dict): def test_get_nested_attribute(setup_dict: dict[str, Any]) -> None:
nested_dict = get_nested_dict_by_path(setup_dict, "attr2.attr3") nested_dict = get_nested_dict_by_path(setup_dict, "attr2.attr3")
assert nested_dict["value"] == 1.0 assert nested_dict["value"] == 1.0
def test_get_list_entry(setup_dict): def test_get_list_entry(setup_dict: dict[str, Any]) -> None:
nested_dict = get_nested_dict_by_path(setup_dict, "attr_list[1]") nested_dict = get_nested_dict_by_path(setup_dict, "attr_list[1]")
assert nested_dict["value"] == 1 assert nested_dict["value"] == 1
def test_get_list_inside_class(setup_dict): def test_get_list_inside_class(setup_dict: dict[str, Any]) -> None:
nested_dict = get_nested_dict_by_path(setup_dict, "attr2.list_attr[1]") nested_dict = get_nested_dict_by_path(setup_dict, "attr2.list_attr[1]")
assert nested_dict["value"] == 1.0 assert nested_dict["value"] == 1.0
def test_get_class_attribute_inside_list(setup_dict): def test_get_class_attribute_inside_list(setup_dict: dict[str, Any]) -> None:
nested_dict = get_nested_dict_by_path(setup_dict, "attr_list[2].attr3") nested_dict = get_nested_dict_by_path(setup_dict, "attr_list[2].attr3")
assert nested_dict["value"] == 1.0 assert nested_dict["value"] == 1.0
def test_get_invalid_list_index(setup_dict, caplog: pytest.LogCaptureFixture): def test_get_invalid_list_index(setup_dict: dict[str, Any]) -> None:
with pytest.raises(SerializationPathError): with pytest.raises(SerializationPathError):
get_nested_dict_by_path(setup_dict, "attr_list[10]") get_nested_dict_by_path(setup_dict, "attr_list[10]")
def test_get_invalid_path(setup_dict, caplog: pytest.LogCaptureFixture): def test_get_invalid_path(setup_dict: dict[str, Any]) -> None:
with pytest.raises(SerializationPathError): with pytest.raises(SerializationPathError):
get_nested_dict_by_path(setup_dict, "invalid_path") get_nested_dict_by_path(setup_dict, "invalid_path")
def test_serialized_dict_is_nested_object() -> None:
serialized_dict = {
"list_attr": {
"type": "list",
"value": [
{"type": "float", "value": 1.4, "readonly": False, "doc": None},
{"type": "float", "value": 2.0, "readonly": False, "doc": None},
],
"readonly": False,
"doc": None,
},
"my_slider": {
"type": "NumberSlider",
"value": {
"max": {
"type": "float",
"value": 101.0,
"readonly": False,
"doc": "The min property.",
},
"min": {
"type": "float",
"value": 1.0,
"readonly": False,
"doc": "The min property.",
},
"step_size": {
"type": "float",
"value": 2.0,
"readonly": False,
"doc": "The min property.",
},
"value": {
"type": "float",
"value": 1.0,
"readonly": False,
"doc": "The value property.",
},
},
"readonly": False,
"doc": None,
},
"string": {
"type": "str",
"value": "Another name",
"readonly": True,
"doc": None,
},
"float": {
"type": "int",
"value": 10,
"readonly": False,
"doc": None,
},
"unit": {
"type": "Quantity",
"value": {"magnitude": 12.0, "unit": "A"},
"readonly": False,
"doc": None,
},
"state": {
"type": "ColouredEnum",
"value": "FAILED",
"readonly": True,
"doc": None,
"enum": {
"RUNNING": "#0000FF80",
"COMPLETED": "hsl(120, 100%, 50%)",
"FAILED": "hsla(0, 100%, 50%, 0.7)",
},
},
"subservice": {
"type": "DataService",
"value": {
"name": {
"type": "str",
"value": "SubService",
"readonly": False,
"doc": None,
}
},
"readonly": False,
"doc": None,
},
}
assert serialized_dict_is_nested_object(serialized_dict["list_attr"])
assert serialized_dict_is_nested_object(serialized_dict["my_slider"])
assert serialized_dict_is_nested_object(serialized_dict["subservice"])
assert not serialized_dict_is_nested_object(
serialized_dict["list_attr"]["value"][0] # type: ignore[index]
)
assert not serialized_dict_is_nested_object(serialized_dict["string"])
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"])