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

refactoring serializer module methods

Browse Source
This commit is contained in:
Mose Müller 2023-11-07 18:23:24 +01:00
parent 7c573cdc10
commit 14b5219915
7 changed files with 89 additions and 274 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

28
src/pydase/data_service/data_service.py
View File

@ -13,13 +13,16 @@ from pydase.data_service.task_manager import TaskManager
from pydase.utils.helpers import (
convert_arguments_to_hinted_types,
get_class_and_instance_attributes,
get_nested_value_from_DataService_by_path_and_key,
get_object_attr_from_path,
is_property_attribute,
parse_list_attr_and_index,
update_value_if_changed,
)
from pydase.utils.serializer import Serializer, generate_serialized_data_paths
from pydase.utils.serializer import (
Serializer,
generate_serialized_data_paths,
get_nested_dict_by_path,
)
from pydase.utils.warnings import (
warn_if_instance_class_does_not_inherit_from_DataService,
)
@ -165,21 +168,14 @@ class DataService(rpyc.Service, AbstractDataService):
# Traverse the serialized representation and set the attributes of the class
serialized_class = self.serialize()
for path in generate_serialized_data_paths(json_dict):
value = get_nested_value_from_DataService_by_path_and_key(
json_dict, path=path
)
value_type = get_nested_value_from_DataService_by_path_and_key(
json_dict, path=path, key="type"
)
class_value_type = get_nested_value_from_DataService_by_path_and_key(
serialized_class, path=path, key="type"
)
nested_json_dict = get_nested_dict_by_path(json_dict, path)
value = nested_json_dict["value"]
value_type = nested_json_dict["type"]
nested_class_dict = get_nested_dict_by_path(serialized_class, path)
class_value_type = nested_class_dict.get("type", None)
if class_value_type == value_type:
class_attr_is_read_only = (
get_nested_value_from_DataService_by_path_and_key(
serialized_class, path=path, key="readonly"
)
)
class_attr_is_read_only = nested_class_dict["readonly"]
if class_attr_is_read_only:
logger.debug(
f'Attribute "{path}" is read-only. Ignoring value from JSON '

27
src/pydase/data_service/state_manager.py
View File

@ -6,8 +6,10 @@ from typing import TYPE_CHECKING, Any, Optional, cast
import pydase.units as u
from pydase.data_service.data_service_cache import DataServiceCache
from pydase.utils.helpers import get_nested_value_from_DataService_by_path_and_key
from pydase.utils.serializer import generate_serialized_data_paths
from pydase.utils.serializer import (
generate_serialized_data_paths,
get_nested_dict_by_path,
)
if TYPE_CHECKING:
from pydase import DataService
@ -102,21 +104,14 @@ class StateManager:
serialized_class = self.cache
for path in generate_serialized_data_paths(json_dict):
value = get_nested_value_from_DataService_by_path_and_key(
json_dict, path=path
)
value_type = get_nested_value_from_DataService_by_path_and_key(
json_dict, path=path, key="type"
)
class_value_type = get_nested_value_from_DataService_by_path_and_key(
serialized_class, path=path, key="type"
)
nested_json_dict = get_nested_dict_by_path(json_dict, path)
value = nested_json_dict["value"]
value_type = nested_json_dict["type"]
nested_class_dict = get_nested_dict_by_path(serialized_class, path)
class_value_type = nested_class_dict.get("type", None)
if class_value_type == value_type:
class_attr_is_read_only = (
get_nested_value_from_DataService_by_path_and_key(
serialized_class, path=path, key="readonly"
)
)
class_attr_is_read_only = nested_class_dict["readonly"]
if class_attr_is_read_only:
logger.debug(
f"Attribute {path!r} is read-only. Ignoring value from JSON "

151
src/pydase/utils/helpers.py
View File

@ -1,23 +1,10 @@
import inspect
import logging
import re
from itertools import chain
from typing import Any, Optional, cast
from typing import Any, Optional
logger = logging.getLogger(__name__)
STANDARD_TYPES = (
"int",
"float",
"bool",
"str",
"method",
"Enum",
"NoneType",
"Quantity",
"ColouredEnum",
)
def get_attribute_doc(attr: Any) -> Optional[str]:
"""This function takes an input attribute attr and returns its documentation
@ -77,142 +64,6 @@ def get_object_attr_from_path(target_obj: Any, path: list[str]) -> Any:
return target_obj
def extract_dict_or_list_entry(data: dict[str, Any], key: str) -> dict[str, Any] | None:
"""
Extract a nested dictionary or list entry based on the provided key.
Given a dictionary and a key, this function retrieves the corresponding nested
dictionary or list entry. If the key includes an index in the format "[<index>]",
the function assumes that the corresponding entry in the dictionary is a list, and
it will attempt to retrieve the indexed item from that list.
Args:
data (dict): The input dictionary containing nested dictionaries or lists.
key (str): The key specifying the desired entry within the dictionary. The key
can be a regular dictionary key or can include an index in the format
"[<index>]" to retrieve an item from a nested list.
Returns:
dict | None: The nested dictionary or list item found for the given key. If the
key is invalid, or if the specified index is out of bounds for a list, it
returns None.
Example:
>>> data = {
... "attr1": [
... {"type": "int", "value": 10}, {"type": "string", "value": "hello"}
... ],
... "attr2": {
... "type": "MyClass",
... "value": {"sub_attr": {"type": "float", "value": 20.5}}
... }
... }
>>> extract_dict_or_list_entry(data, "attr1[1]")
{"type": "string", "value": "hello"}
>>> extract_dict_or_list_entry(data, "attr2")
{"type": "MyClass", "value": {"sub_attr": {"type": "float", "value": 20.5}}}
"""
attr_name = key
index: Optional[int] = None
# Check if the key contains an index part like '[<index>]'
if "[" in key and key.endswith("]"):
attr_name, index_part = key.split("[", 1)
index_part = index_part.rstrip("]") # remove the closing bracket
# Convert the index part to an integer
if index_part.isdigit():
index = int(index_part)
else:
logger.error(f"Invalid index format in key: {key}")
current_data: dict[str, Any] | list[dict[str, Any]] | None = data.get(
attr_name, None
)
if not isinstance(current_data, dict):
# key does not exist in dictionary, e.g. when class does not have this
# attribute
return None
if isinstance(current_data["value"], list):
current_data = current_data["value"]
if index is not None and 0 <= index < len(current_data):
current_data = current_data[index]
else:
return None
# When the attribute is a class instance, the attributes are nested in the
# "value" key
if current_data["type"] not in STANDARD_TYPES:
current_data = cast(dict[str, Any], current_data.get("value", None)) # type: ignore
assert isinstance(current_data, dict)
return current_data
def get_nested_value_from_DataService_by_path_and_key(
data: dict[str, Any], path: str, key: str = "value"
) -> Any:
"""
Get the value associated with a specific key from a dictionary given a path.
This function traverses the dictionary according to the path provided and
returns the value associated with the specified key at that path. The path is
a string with dots connecting the levels and brackets indicating list indices.
The function can handle complex dictionaries where data is nested within different
types of objects. It checks the type of each object it encounters and correctly
descends into the object if it is not a standard type (i.e., int, float, bool, str,
Enum).
Args:
data (dict): The input dictionary to get the value from.
path (str): The path to the value in the dictionary.
key (str, optional): The key associated with the value to be returned.
Default is "value".
Returns:
Any: The value associated with the specified key at the given path in the
dictionary.
Examples:
Let's consider the following dictionary:
>>> data = {
>>> "attr1": {"type": "int", "value": 10},
>>> "attr2": {
"type": "MyClass",
"value": {"attr3": {"type": "float", "value": 20.5}}
}
>>> }
The function can be used to get the value of 'attr1' as follows:
>>> get_nested_value_by_path_and_key(data, "attr1")
10
It can also be used to get the value of 'attr3', which is nested within 'attr2',
as follows:
>>> get_nested_value_by_path_and_key(data, "attr2.attr3", "type")
float
"""
# Split the path into parts
parts: list[str] = re.split(r"\.", path) # Split by '.'
current_data: dict[str, Any] | None = data
for part in parts:
if current_data is None:
return
current_data = extract_dict_or_list_entry(current_data, part)
if isinstance(current_data, dict):
return current_data.get(key, None)
def convert_arguments_to_hinted_types(
args: dict[str, Any], type_hints: dict[str, Any]
) -> dict[str, Any] | str:

48
src/pydase/utils/serializer.py
View File

@ -241,23 +241,16 @@ def set_nested_value_by_path(
parent_path_parts, attr_name = path.split(".")[:-1], path.split(".")[-1]
current_dict: dict[str, Any] = serialization_dict
index: Optional[int] = None
try:
for path_part in parent_path_parts:
# Check if the key contains an index part like 'attr_name[<index>]'
path_part, index = parse_list_attr_and_index(path_part)
current_dict = get_nested_dict_by_attr_and_index(
current_dict, path_part, index, allow_append=False
current_dict = get_next_level_dict_by_key(
current_dict, path_part, allow_append=False
)
current_dict = current_dict["value"]
index = None
attr_name, index = parse_list_attr_and_index(attr_name)
current_dict = get_nested_dict_by_attr_and_index(
current_dict, attr_name, index, allow_append=True
current_dict = get_next_level_dict_by_key(
current_dict, attr_name, allow_append=True
)
except (SerializationPathError, SerializationValueError, KeyError) as e:
logger.error(e)
@ -272,10 +265,33 @@ def set_nested_value_by_path(
current_dict.update(serialized_value)
def get_nested_dict_by_attr_and_index(
def get_nested_dict_by_path(
serialization_dict: dict[str, Any],
path: str,
) -> dict[str, Any]:
parent_path_parts, attr_name = path.split(".")[:-1], path.split(".")[-1]
current_dict: dict[str, Any] = serialization_dict
try:
for path_part in parent_path_parts:
current_dict = get_next_level_dict_by_key(
current_dict, path_part, allow_append=False
)
current_dict = current_dict["value"]
current_dict = get_next_level_dict_by_key(
current_dict, attr_name, allow_append=False
)
except (SerializationPathError, SerializationValueError, KeyError) as e:
logger.error(e)
return {}
return current_dict
def get_next_level_dict_by_key(
serialization_dict: dict[str, Any],
attr_name: str,
index: Optional[int],
allow_append: bool = False,
) -> dict[str, Any]:
"""
@ -284,8 +300,8 @@ def get_nested_dict_by_attr_and_index(
Args:
serialization_dict: The base dictionary representing serialized data.
attr_name: The key name representing the attribute in the dictionary.
index: The optional index for list items within the dictionary value.
attr_name: The key name representing the attribute in the dictionary,
e.g. 'list_attr[0]' or 'attr'
allow_append: Flag to allow appending a new entry if `index` is out of range by
one.
@ -297,6 +313,8 @@ def get_nested_dict_by_attr_and_index(
invalid or leads to an IndexError or KeyError.
SerializationValueError: If the expected nested structure is not a dictionary.
"""
# Check if the key contains an index part like 'attr_name[<index>]'
attr_name, index = parse_list_attr_and_index(attr_name)
try:
if index is not None:

14
tests/data_service/test_data_service_cache.py
View File

@ -2,7 +2,7 @@ import logging
import pydase
from pydase.data_service.data_service_cache import DataServiceCache
from pydase.utils.helpers import get_nested_value_from_DataService_by_path_and_key
from pydase.utils.serializer import get_nested_dict_by_path
logger = logging.getLogger()
@ -19,15 +19,7 @@ def test_nested_attributes_cache_callback() -> None:
cache = DataServiceCache(test_service)
test_service.name = "Peepz"
assert (
get_nested_value_from_DataService_by_path_and_key(cache.cache, "name")
== "Peepz"
)
assert get_nested_dict_by_path(cache.cache, "name")["value"] == "Peepz"
test_service.class_attr.name = "Ciao"
assert (
get_nested_value_from_DataService_by_path_and_key(
cache.cache, "class_attr.name"
)
== "Ciao"
)
assert get_nested_dict_by_path(cache.cache, "class_attr.name")["value"] == "Ciao"

66
tests/utils/test_helpers.py
View File

@ -1,70 +1,6 @@
import pytest
from pydase.utils.helpers import (
extract_dict_or_list_entry,
get_nested_value_from_DataService_by_path_and_key,
is_property_attribute,
)
# Sample data for the tests
data_sample = {
"attr1": {"type": "bool", "value": False, "readonly": False, "doc": None},
"class_attr": {
"type": "MyClass",
"value": {"sub_attr": {"type": "float", "value": 20.5}},
},
"list_attr": {
"type": "list",
"value": [
{"type": "int", "value": 0, "readonly": False, "doc": None},
{"type": "float", "value": 1.0, "readonly": False, "doc": None},
],
"readonly": False,
},
}
# Tests for extract_dict_or_list_entry
def test_extract_dict_with_valid_list_index() -> None:
result = extract_dict_or_list_entry(data_sample, "list_attr[1]")
assert result == {"type": "float", "value": 1.0, "readonly": False, "doc": None}
def test_extract_dict_without_list_index() -> None:
result = extract_dict_or_list_entry(data_sample, "attr1")
assert result == {"type": "bool", "value": False, "readonly": False, "doc": None}
def test_extract_dict_with_invalid_key() -> None:
result = extract_dict_or_list_entry(data_sample, "attr_not_exist")
assert result is None
def test_extract_dict_with_invalid_list_index() -> None:
result = extract_dict_or_list_entry(data_sample, "list_attr[5]")
assert result is None
# Tests for get_nested_value_from_DataService_by_path_and_key
def test_get_nested_value_with_default_key() -> None:
result = get_nested_value_from_DataService_by_path_and_key(
data_sample, "list_attr[0]"
)
assert result == 0
def test_get_nested_value_with_custom_key() -> None:
result = get_nested_value_from_DataService_by_path_and_key(
data_sample, "class_attr.sub_attr", "type"
)
assert result == "float"
def test_get_nested_value_with_invalid_path() -> None:
result = get_nested_value_from_DataService_by_path_and_key(
data_sample, "class_attr.nonexistent_attr"
)
assert result is None
from pydase.utils.helpers import is_property_attribute
@pytest.mark.parametrize(

29
tests/utils/test_serialization.py
View File

@ -6,7 +6,12 @@ import pytest
import pydase
import pydase.units as u
from pydase.components.coloured_enum import ColouredEnum
from pydase.utils.serializer import dump, set_nested_value_by_path
from pydase.utils.serializer import (
SerializationPathError,
dump,
get_next_level_dict_by_key,
set_nested_value_by_path,
)
@pytest.mark.parametrize(
@ -347,3 +352,25 @@ def test_update_list_inside_class(setup_dict):
def test_update_class_attribute_inside_list(setup_dict):
set_nested_value_by_path(setup_dict, "attr_list[2].attr3", 50)
assert setup_dict["attr_list"]["value"][2]["value"]["attr3"]["value"] == 50
def test_get_attribute_nested_dict(setup_dict):
nested_dict = get_next_level_dict_by_key(setup_dict, "attr1")
assert nested_dict == setup_dict["attr1"]
def test_get_list_entry_nested_dict(setup_dict):
nested_dict = get_next_level_dict_by_key(setup_dict, "attr_list[0]")
assert nested_dict == setup_dict["attr_list"]["value"][0]
def test_get_invalid_path_nested_dict(setup_dict):
with pytest.raises(SerializationPathError):
get_next_level_dict_by_key(setup_dict, "invalid_path")
def test_get_invalid_list_index(setup_dict):
with pytest.raises(SerializationPathError):
get_next_level_dict_by_key(setup_dict, "attr_list[10]")