mirror of
https://github.com/tiqi-group/pydase.git
synced 2025-06-07 05:50:41 +02:00
moving set_nested_value_in_dict to Serializer, renaming module
This commit is contained in:
parent
4ef4bab36e
commit
3440a632ad
@ -20,7 +20,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.serialization import Serializer
|
from pydase.utils.serializer import Serializer
|
||||||
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,
|
||||||
)
|
)
|
||||||
|
@ -1,7 +1,7 @@
|
|||||||
import logging
|
import logging
|
||||||
from typing import TYPE_CHECKING, Any
|
from typing import TYPE_CHECKING, Any
|
||||||
|
|
||||||
from pydase.utils.helpers import set_nested_value_in_dict
|
from pydase.utils.serializer import Serializer
|
||||||
|
|
||||||
if TYPE_CHECKING:
|
if TYPE_CHECKING:
|
||||||
from pydase import DataService
|
from pydase import DataService
|
||||||
@ -32,5 +32,5 @@ class DataServiceCache:
|
|||||||
# Construct the full path
|
# Construct the full path
|
||||||
full_path = f"{parent_path}.{name}" if parent_path else name
|
full_path = f"{parent_path}.{name}" if parent_path else name
|
||||||
|
|
||||||
set_nested_value_in_dict(self._cache, full_path, value)
|
Serializer.update_serialization_dict(self._cache, full_path, value)
|
||||||
logger.debug(f"Cache updated at path: {full_path}, with value: {value}")
|
logger.debug(f"Cache updated at path: {full_path}, with value: {value}")
|
||||||
|
@ -272,89 +272,6 @@ def get_nested_value_from_DataService_by_path_and_key(
|
|||||||
return current_data.get(key, None)
|
return current_data.get(key, None)
|
||||||
|
|
||||||
|
|
||||||
def set_nested_value_in_dict(data_dict: dict[str, Any], path: str, value: Any) -> None:
|
|
||||||
"""
|
|
||||||
Set the value associated with a specific key in a dictionary given a path.
|
|
||||||
|
|
||||||
This function traverses the dictionary according to the path provided and
|
|
||||||
sets the value at that path. The path is a string with dots connecting
|
|
||||||
the levels and brackets indicating list indices.
|
|
||||||
|
|
||||||
Args:
|
|
||||||
cache (dict): The cache dictionary to set the value in.
|
|
||||||
path (str): The path to where the value should be set in the dictionary.
|
|
||||||
value (Any): The value to be set at the specified path in the dictionary.
|
|
||||||
|
|
||||||
Examples:
|
|
||||||
Let's consider the following dictionary:
|
|
||||||
|
|
||||||
cache = {
|
|
||||||
"attr1": {"type": "int", "value": 10},
|
|
||||||
"attr2": {
|
|
||||||
"type": "MyClass",
|
|
||||||
"value": {"attr3": {"type": "float", "value": 20.5}}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
The function can be used to set the value of 'attr1' as follows:
|
|
||||||
set_nested_value_in_cache(cache, "attr1", 15)
|
|
||||||
|
|
||||||
It can also be used to set the value of 'attr3', which is nested within 'attr2',
|
|
||||||
as follows:
|
|
||||||
set_nested_value_in_cache(cache, "attr2.attr3", 25.0)
|
|
||||||
"""
|
|
||||||
|
|
||||||
parts = path.split(".")
|
|
||||||
current_dict: dict[str, Any] = data_dict
|
|
||||||
index: Optional[int] = None
|
|
||||||
|
|
||||||
for attr_name in parts:
|
|
||||||
# Check if the key contains an index part like '[<index>]'
|
|
||||||
if "[" in attr_name and attr_name.endswith("]"):
|
|
||||||
attr_name, index_part = attr_name.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: {attr_name}")
|
|
||||||
|
|
||||||
current_dict = cast(dict[str, Any], current_dict.get(attr_name, None))
|
|
||||||
|
|
||||||
if not isinstance(current_dict, dict):
|
|
||||||
# key does not exist in dictionary, e.g. when class does not have this
|
|
||||||
# attribute
|
|
||||||
return
|
|
||||||
|
|
||||||
if index is not None:
|
|
||||||
if 0 <= index < len(current_dict["value"]):
|
|
||||||
try:
|
|
||||||
current_dict = cast(dict[str, Any], current_dict["value"][index])
|
|
||||||
except Exception as e:
|
|
||||||
logger.error(f"Could not change {path}. Exception: {e}")
|
|
||||||
return
|
|
||||||
else:
|
|
||||||
# TODO: appending to a list will probably be done here
|
|
||||||
logger.error(f"Could not change {path}...")
|
|
||||||
return
|
|
||||||
|
|
||||||
# When the attribute is a class instance, the attributes are nested in the
|
|
||||||
# "value" key
|
|
||||||
if (
|
|
||||||
current_dict["type"] not in STANDARD_TYPES
|
|
||||||
and current_dict["type"] != "method"
|
|
||||||
):
|
|
||||||
current_dict = cast(dict[str, Any], current_dict.get("value", None)) # type: ignore
|
|
||||||
|
|
||||||
index = None
|
|
||||||
|
|
||||||
# setting the new value
|
|
||||||
try:
|
|
||||||
current_dict["value"] = value
|
|
||||||
except Exception as e:
|
|
||||||
logger.error(e)
|
|
||||||
|
|
||||||
|
|
||||||
def convert_arguments_to_hinted_types(
|
def convert_arguments_to_hinted_types(
|
||||||
args: dict[str, Any], type_hints: dict[str, Any]
|
args: dict[str, Any], type_hints: dict[str, Any]
|
||||||
|
@ -2,11 +2,15 @@ import inspect
|
|||||||
import logging
|
import logging
|
||||||
from collections.abc import Callable
|
from collections.abc import Callable
|
||||||
from enum import Enum
|
from enum import Enum
|
||||||
from typing import Any, Optional
|
from typing import Any, Optional, cast
|
||||||
|
|
||||||
import pydase.units as u
|
import pydase.units as u
|
||||||
from pydase.data_service.abstract_data_service import AbstractDataService
|
from pydase.data_service.abstract_data_service import AbstractDataService
|
||||||
from pydase.utils.helpers import get_component_class_names
|
from pydase.utils.helpers import (
|
||||||
|
STANDARD_TYPES,
|
||||||
|
get_component_class_names,
|
||||||
|
parse_list_attr_and_index,
|
||||||
|
)
|
||||||
|
|
||||||
logger = logging.getLogger(__name__)
|
logger = logging.getLogger(__name__)
|
||||||
|
|
||||||
@ -212,6 +216,79 @@ class Serializer:
|
|||||||
"doc": doc,
|
"doc": doc,
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@staticmethod
|
||||||
|
def update_serialization_dict(
|
||||||
|
serialization_dict: dict[str, Any], path: str, value: Any
|
||||||
|
) -> None:
|
||||||
|
"""
|
||||||
|
Set the value associated with a specific key in a dictionary given a path.
|
||||||
|
|
||||||
|
This function traverses the dictionary according to the path provided and
|
||||||
|
sets the value at that path. The path is a string with dots connecting
|
||||||
|
the levels and brackets indicating list indices.
|
||||||
|
|
||||||
|
Args:
|
||||||
|
data_dict (dict): The cache dictionary to set the value in.
|
||||||
|
path (str): The path to where the value should be set in the dictionary.
|
||||||
|
value (Any): The value to be set at the specified path in the dictionary.
|
||||||
|
|
||||||
|
Examples:
|
||||||
|
Let's consider the following dictionary:
|
||||||
|
|
||||||
|
cache = {
|
||||||
|
"attr1": {"type": "int", "value": 10},
|
||||||
|
"attr2": {
|
||||||
|
"type": "MyClass",
|
||||||
|
"value": {"attr3": {"type": "float", "value": 20.5}}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
The function can be used to set the value of 'attr1' as follows:
|
||||||
|
set_nested_value_in_cache(cache, "attr1", 15)
|
||||||
|
|
||||||
|
It can also be used to set the value of 'attr3', which is nested within
|
||||||
|
'attr2', as follows:
|
||||||
|
set_nested_value_in_cache(cache, "attr2.attr3", 25.0)
|
||||||
|
"""
|
||||||
|
|
||||||
|
parts, attr_name = path.split(".")[:-1], path.split(".")[-1]
|
||||||
|
current_dict: dict[str, Any] = serialization_dict
|
||||||
|
index: Optional[int] = None
|
||||||
|
|
||||||
|
for path_part in 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 = cast(dict[str, Any], current_dict.get(path_part, None))
|
||||||
|
|
||||||
|
if not isinstance(current_dict, dict):
|
||||||
|
# key does not exist in dictionary, e.g. when class does not have this
|
||||||
|
# attribute
|
||||||
|
return
|
||||||
|
|
||||||
|
if index is not None:
|
||||||
|
try:
|
||||||
|
current_dict = cast(dict[str, Any], current_dict["value"][index])
|
||||||
|
except Exception as e:
|
||||||
|
# TODO: appending to a list will probably be done here
|
||||||
|
logger.error(f"Could not change {path}... {e}")
|
||||||
|
return
|
||||||
|
|
||||||
|
# When the attribute is a class instance, the attributes are nested in the
|
||||||
|
# "value" key
|
||||||
|
if (
|
||||||
|
current_dict["type"] not in STANDARD_TYPES
|
||||||
|
and current_dict["type"] != "method"
|
||||||
|
):
|
||||||
|
current_dict = cast(dict[str, Any], current_dict.get("value", None)) # type: ignore
|
||||||
|
|
||||||
|
index = None
|
||||||
|
|
||||||
|
# setting the new value
|
||||||
|
serialized_value = dump(value)
|
||||||
|
current_dict[attr_name]["value"] = serialized_value["value"]
|
||||||
|
current_dict[attr_name]["type"] = serialized_value["type"]
|
||||||
|
|
||||||
|
|
||||||
def dump(obj: Any) -> dict[str, Any]:
|
def dump(obj: Any) -> dict[str, Any]:
|
||||||
return Serializer.serialize_object(obj)
|
return Serializer.serialize_object(obj)
|
@ -6,7 +6,7 @@ import pytest
|
|||||||
import pydase
|
import pydase
|
||||||
import pydase.units as u
|
import pydase.units as u
|
||||||
from pydase.components.coloured_enum import ColouredEnum
|
from pydase.components.coloured_enum import ColouredEnum
|
||||||
from pydase.utils.serialization import dump
|
from pydase.utils.serializer import dump
|
||||||
|
|
||||||
|
|
||||||
@pytest.mark.parametrize(
|
@pytest.mark.parametrize(
|
||||||
|
Loading…
x
Reference in New Issue
Block a user