ITQC/pydase
24
0
Fork 0
mirror of https://github.com/tiqi-group/pydase.git synced 2025-04-20 00:10:03 +02:00
Code Issues Packages Projects Releases Wiki Activity

adds add_prefix_to_full_access_path helper function

Browse Source
This commit is contained in:
Mose Müller 2024-10-01 07:13:22 +02:00
parent a9ea237cf3
commit f47a183c11
2 changed files with 167 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

60
src/pydase/utils/serialization/serializer.py
View File

@ -572,6 +572,66 @@ def generate_serialized_data_paths(
return paths return paths
def add_prefix_to_full_access_path(
serialized_obj: SerializedObject, prefix: str
) -> Any:
"""Recursively adds a specified prefix to all full access paths of the serialized
object.
Args:
data:
The serialized object to process.
prefix:
The prefix string to prepend to each full access path.
Returns:
The modified serialized object with the prefix added to all full access paths.
Example:
```python
>>> data = {
... "full_access_path": "some_path",
... "value": {
... "item": {
... "full_access_path": "some_item_path",
... "value": 1.0
... }
... }
... }
...
... modified_data = add_prefix_to_full_access_path(data, 'prefix.')
{"full_access_path": "prefix.some_path", "value": {"item": {"full_access_path":
"prefix.some_item_path", "value": 1.0}}}
```
"""
try:
serialized_obj["full_access_path"] = prefix + serialized_obj["full_access_path"]
if isinstance(serialized_obj["value"], list):
for value in serialized_obj["value"]:
value["full_access_path"] = prefix + value["full_access_path"]
add_prefix_to_full_access_path(
cast(SerializedObject, value["value"]), prefix
)
elif isinstance(serialized_obj["value"], dict):
for value in cast(
dict[str, SerializedObject], serialized_obj["value"]
).values():
value["full_access_path"] = prefix + value["full_access_path"]
add_prefix_to_full_access_path(
cast(SerializedObject, value["value"]), prefix
)
except TypeError:
# passed object is not a dict (cannot use __get__ on it)
pass
except KeyError:
# passed dictionary is not a serialized object
pass
return serialized_obj
def serialized_dict_is_nested_object(serialized_dict: SerializedObject) -> bool: def serialized_dict_is_nested_object(serialized_dict: SerializedObject) -> bool:
value = serialized_dict["value"] value = serialized_dict["value"]
# We are excluding Quantity here as the value corresponding to the "value" key is # We are excluding Quantity here as the value corresponding to the "value" key is

107
tests/utils/serialization/test_serializer.py
View File

@ -12,6 +12,7 @@ from pydase.utils.decorators import frontend
from pydase.utils.serialization.serializer import ( from pydase.utils.serialization.serializer import (
SerializationPathError, SerializationPathError,
SerializedObject, SerializedObject,
add_prefix_to_full_access_path,
dump, dump,
generate_serialized_data_paths, generate_serialized_data_paths,
get_container_item_by_key, get_container_item_by_key,
@ -1070,3 +1071,109 @@ def test_get_data_paths_from_serialized_object(obj: Any, expected: list[str]) ->
) )
def test_generate_serialized_data_paths(obj: Any, expected: list[str]) -> None: def test_generate_serialized_data_paths(obj: Any, expected: list[str]) -> None:
assert generate_serialized_data_paths(dump(obj=obj)["value"]) == expected assert generate_serialized_data_paths(dump(obj=obj)["value"]) == expected
@pytest.mark.parametrize(
"serialized_obj, prefix, expected",
[
(
{
"full_access_path": "new_attr",
"value": {
"name": {
"full_access_path": "new_attr.name",
"value": "MyService",
}
},
},
"prefix.",
{
"full_access_path": "prefix.new_attr",
"value": {
"name": {
"full_access_path": "prefix.new_attr.name",
"value": "MyService",
}
},
},
),
(
{
"full_access_path": "new_attr",
"value": [
{
"full_access_path": "new_attr[0]",
"value": 1.0,
}
],
},
"prefix.",
{
"full_access_path": "prefix.new_attr",
"value": [
{
"full_access_path": "prefix.new_attr[0]",
"value": 1.0,
}
],
},
),
(
{
"full_access_path": "new_attr",
"value": {
"key": {
"full_access_path": 'new_attr["key"]',
"value": 1.0,
}
},
},
"prefix.",
{
"full_access_path": "prefix.new_attr",
"value": {
"key": {
"full_access_path": 'prefix.new_attr["key"]',
"value": 1.0,
}
},
},
),
(
{
"full_access_path": "new_attr",
"value": {"magnitude": 10, "unit": "meter"},
},
"prefix.",
{
"full_access_path": "prefix.new_attr",
"value": {"magnitude": 10, "unit": "meter"},
},
),
(
{
"full_access_path": "quantity_list",
"value": [
{
"full_access_path": "quantity_list[0]",
"value": {"magnitude": 1.0, "unit": "A"},
}
],
},
"prefix.",
{
"full_access_path": "prefix.quantity_list",
"value": [
{
"full_access_path": "prefix.quantity_list[0]",
"value": {"magnitude": 1.0, "unit": "A"},
}
],
},
),
],
)
def test_add_prefix_to_full_access_path(
serialized_obj: SerializedObject, prefix: str, expected: SerializedObject
) -> None:
assert add_prefix_to_full_access_path(serialized_obj, prefix) == expected