diff --git a/src/pydase/data_service/data_service.py b/src/pydase/data_service/data_service.py index 112c2cc..b676327 100644 --- a/src/pydase/data_service/data_service.py +++ b/src/pydase/data_service/data_service.py @@ -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 ' diff --git a/src/pydase/data_service/state_manager.py b/src/pydase/data_service/state_manager.py index 419dd3c..790c406 100644 --- a/src/pydase/data_service/state_manager.py +++ b/src/pydase/data_service/state_manager.py @@ -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 " diff --git a/src/pydase/utils/helpers.py b/src/pydase/utils/helpers.py index fe49296..9161013 100644 --- a/src/pydase/utils/helpers.py +++ b/src/pydase/utils/helpers.py @@ -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 "[]", - 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 - "[]" 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 '[]' - 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: diff --git a/src/pydase/utils/serializer.py b/src/pydase/utils/serializer.py index c79c1bc..ece26c4 100644 --- a/src/pydase/utils/serializer.py +++ b/src/pydase/utils/serializer.py @@ -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[]' - 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[]' + attr_name, index = parse_list_attr_and_index(attr_name) try: if index is not None: diff --git a/tests/data_service/test_data_service_cache.py b/tests/data_service/test_data_service_cache.py index d89ca5f..35dd400 100644 --- a/tests/data_service/test_data_service_cache.py +++ b/tests/data_service/test_data_service_cache.py @@ -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" diff --git a/tests/utils/test_helpers.py b/tests/utils/test_helpers.py index 28f74ed..ddfcc22 100644 --- a/tests/utils/test_helpers.py +++ b/tests/utils/test_helpers.py @@ -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( diff --git a/tests/utils/test_serialization.py b/tests/utils/test_serialization.py index 0fde167..b40b8a9 100644 --- a/tests/utils/test_serialization.py +++ b/tests/utils/test_serialization.py @@ -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]") + +