mirror of
https://github.com/tiqi-group/pydase.git
synced 2025-04-21 00:40:01 +02:00
moves generate_paths_from_DataService_dict to serializer module
This commit is contained in:
parent
59c7d7bb6f
commit
03fee3f88c
@ -12,7 +12,6 @@ from pydase.data_service.callback_manager import CallbackManager
|
|||||||
from pydase.data_service.task_manager import TaskManager
|
from pydase.data_service.task_manager import TaskManager
|
||||||
from pydase.utils.helpers import (
|
from pydase.utils.helpers import (
|
||||||
convert_arguments_to_hinted_types,
|
convert_arguments_to_hinted_types,
|
||||||
generate_paths_from_DataService_dict,
|
|
||||||
get_class_and_instance_attributes,
|
get_class_and_instance_attributes,
|
||||||
get_nested_value_from_DataService_by_path_and_key,
|
get_nested_value_from_DataService_by_path_and_key,
|
||||||
get_object_attr_from_path,
|
get_object_attr_from_path,
|
||||||
@ -20,7 +19,7 @@ from pydase.utils.helpers import (
|
|||||||
parse_list_attr_and_index,
|
parse_list_attr_and_index,
|
||||||
update_value_if_changed,
|
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 (
|
from pydase.utils.warnings import (
|
||||||
warn_if_instance_class_does_not_inherit_from_DataService,
|
warn_if_instance_class_does_not_inherit_from_DataService,
|
||||||
)
|
)
|
||||||
|
@ -6,10 +6,8 @@ from typing import TYPE_CHECKING, Any, Optional, cast
|
|||||||
|
|
||||||
import pydase.units as u
|
import pydase.units as u
|
||||||
from pydase.data_service.data_service_cache import DataServiceCache
|
from pydase.data_service.data_service_cache import DataServiceCache
|
||||||
from pydase.utils.helpers import (
|
from pydase.utils.helpers import get_nested_value_from_DataService_by_path_and_key
|
||||||
generate_paths_from_DataService_dict,
|
from pydase.utils.serializer import generate_paths_from_DataService_dict
|
||||||
get_nested_value_from_DataService_by_path_and_key,
|
|
||||||
)
|
|
||||||
|
|
||||||
if TYPE_CHECKING:
|
if TYPE_CHECKING:
|
||||||
from pydase import DataService
|
from pydase import DataService
|
||||||
|
@ -11,6 +11,7 @@ STANDARD_TYPES = (
|
|||||||
"float",
|
"float",
|
||||||
"bool",
|
"bool",
|
||||||
"str",
|
"str",
|
||||||
|
"method",
|
||||||
"Enum",
|
"Enum",
|
||||||
"NoneType",
|
"NoneType",
|
||||||
"Quantity",
|
"Quantity",
|
||||||
@ -76,77 +77,6 @@ def get_object_attr_from_path(target_obj: Any, path: list[str]) -> Any:
|
|||||||
return target_obj
|
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:
|
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.
|
Extract a nested dictionary or list entry based on the provided key.
|
||||||
|
@ -331,3 +331,72 @@ def get_nested_dict_by_attr_and_index(
|
|||||||
return serialization_dict
|
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
|
||||||
|
Loading…
x
Reference in New Issue
Block a user