diff --git a/src/pydase/data_service/data_service.py b/src/pydase/data_service/data_service.py index 14097a3..941b12d 100644 --- a/src/pydase/data_service/data_service.py +++ b/src/pydase/data_service/data_service.py @@ -12,7 +12,6 @@ from pydase.data_service.callback_manager import CallbackManager from pydase.data_service.task_manager import TaskManager from pydase.utils.helpers import ( convert_arguments_to_hinted_types, - generate_paths_from_DataService_dict, get_class_and_instance_attributes, get_nested_value_from_DataService_by_path_and_key, get_object_attr_from_path, @@ -20,7 +19,7 @@ from pydase.utils.helpers import ( parse_list_attr_and_index, update_value_if_changed, ) -from pydase.utils.serializer import Serializer +from pydase.utils.serializer import Serializer, generate_paths_from_DataService_dict from pydase.utils.warnings import ( warn_if_instance_class_does_not_inherit_from_DataService, ) diff --git a/src/pydase/data_service/state_manager.py b/src/pydase/data_service/state_manager.py index 8b6df90..5da5f55 100644 --- a/src/pydase/data_service/state_manager.py +++ b/src/pydase/data_service/state_manager.py @@ -6,10 +6,8 @@ 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 ( - generate_paths_from_DataService_dict, - get_nested_value_from_DataService_by_path_and_key, -) +from pydase.utils.helpers import get_nested_value_from_DataService_by_path_and_key +from pydase.utils.serializer import generate_paths_from_DataService_dict if TYPE_CHECKING: from pydase import DataService diff --git a/src/pydase/utils/helpers.py b/src/pydase/utils/helpers.py index 3b35942..fe49296 100644 --- a/src/pydase/utils/helpers.py +++ b/src/pydase/utils/helpers.py @@ -11,6 +11,7 @@ STANDARD_TYPES = ( "float", "bool", "str", + "method", "Enum", "NoneType", "Quantity", @@ -76,77 +77,6 @@ def get_object_attr_from_path(target_obj: Any, path: list[str]) -> Any: return target_obj -def generate_paths_from_DataService_dict( - data: dict, parent_path: str = "" -) -> list[str]: - """ - Recursively generate paths from a dictionary representing a DataService object. - - This function traverses through a nested dictionary, which is typically obtained - from serializing a DataService object. The function generates a list where each - element is a string representing the path to each terminal value in the original - dictionary. - - The paths are represented as strings, with dots ('.') denoting nesting levels and - square brackets ('[]') denoting list indices. - - Args: - data (dict): The input dictionary to generate paths from. This is typically - obtained from serializing a DataService object. - parent_path (str, optional): The current path up to the current level of - recursion. Defaults to ''. - - Returns: - list[str]: A list with paths as elements. - - Note: - The function ignores keys whose "type" is "method", as these represent methods - of the DataService object and not its state. - - Example: - ------- - - >>> { - ... "attr1": {"type": "int", "value": 10}, - ... "attr2": { - ... "type": "list", - ... "value": [{"type": "int", "value": 1}, {"type": "int", "value": 2}], - ... }, - ... "add": { - ... "type": "method", - ... "async": False, - ... "parameters": {"a": "float", "b": "int"}, - ... "doc": "Returns the sum of the numbers a and b.", - ... }, - ... } - >>> print(generate_paths_from_DataService_dict(nested_dict)) - [attr1, attr2[0], attr2[1]] - """ - - paths = [] - for key, value in data.items(): - if value["type"] == "method": - # ignoring methods - continue - new_path = f"{parent_path}.{key}" if parent_path else key - if isinstance(value["value"], dict) and value["type"] != "Quantity": - paths.extend(generate_paths_from_DataService_dict(value["value"], new_path)) # type: ignore - elif isinstance(value["value"], list): - for index, item in enumerate(value["value"]): - indexed_key_path = f"{new_path}[{index}]" - if isinstance(item["value"], dict): - paths.extend( # type: ignore - generate_paths_from_DataService_dict( - item["value"], indexed_key_path - ) - ) - else: - paths.append(indexed_key_path) # type: ignore - else: - paths.append(new_path) # type: ignore - return paths - - 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. diff --git a/src/pydase/utils/serializer.py b/src/pydase/utils/serializer.py index e20a1ee..ca28738 100644 --- a/src/pydase/utils/serializer.py +++ b/src/pydase/utils/serializer.py @@ -331,3 +331,72 @@ def get_nested_dict_by_attr_and_index( return serialization_dict +def generate_paths_from_DataService_dict( + data: dict, parent_path: str = "" +) -> list[str]: + """ + Recursively generate paths from a dictionary representing a DataService object. + + This function traverses through a nested dictionary, which is typically obtained + from serializing a DataService object. The function generates a list where each + element is a string representing the path to each terminal value in the original + dictionary. + + The paths are represented as strings, with dots ('.') denoting nesting levels and + square brackets ('[]') denoting list indices. + + Args: + data (dict): The input dictionary to generate paths from. This is typically + obtained from serializing a DataService object. + parent_path (str, optional): The current path up to the current level of + recursion. Defaults to ''. + + Returns: + list[str]: A list with paths as elements. + + Note: + The function ignores keys whose "type" is "method", as these represent methods + of the DataService object and not its state. + + Example: + ------- + + >>> { + ... "attr1": {"type": "int", "value": 10}, + ... "attr2": { + ... "type": "list", + ... "value": [{"type": "int", "value": 1}, {"type": "int", "value": 2}], + ... }, + ... "add": { + ... "type": "method", + ... "async": False, + ... "parameters": {"a": "float", "b": "int"}, + ... "doc": "Returns the sum of the numbers a and b.", + ... }, + ... } + >>> print(generate_paths_from_DataService_dict(nested_dict)) + [attr1, attr2[0], attr2[1]] + """ + + paths = [] + for key, value in data.items(): + if value["type"] == "method": + # ignoring methods + continue + new_path = f"{parent_path}.{key}" if parent_path else key + if isinstance(value["value"], dict) and value["type"] != "Quantity": + paths.extend(generate_paths_from_DataService_dict(value["value"], new_path)) # type: ignore + elif isinstance(value["value"], list): + for index, item in enumerate(value["value"]): + indexed_key_path = f"{new_path}[{index}]" + if isinstance(item["value"], dict): + paths.extend( # type: ignore + generate_paths_from_DataService_dict( + item["value"], indexed_key_path + ) + ) + else: + paths.append(indexed_key_path) # type: ignore + else: + paths.append(new_path) # type: ignore + return paths