get_object_attr_from_path expects string instead of list now

This commit is contained in:
Mose Müller 2024-03-28 10:16:30 +01:00
parent c02c75aab5
commit bb5205b2e4
4 changed files with 21 additions and 17 deletions

View File

@ -8,7 +8,7 @@ from pydase.observer_pattern.observable.observable_object import ObservableObjec
from pydase.observer_pattern.observer.property_observer import (
PropertyObserver,
)
from pydase.utils.helpers import get_object_attr_from_path_list
from pydase.utils.helpers import get_object_attr_from_path
from pydase.utils.serialization.serializer import SerializedObject, dump
logger = logging.getLogger(__name__)
@ -92,7 +92,7 @@ class DataServiceObserver(PropertyObserver):
if prop not in self.changing_attributes:
self._notify_changed(
prop,
get_object_attr_from_path_list(self.observable, prop.split(".")),
get_object_attr_from_path(self.observable, prop),
)
def add_notification_callback(

View File

@ -8,7 +8,7 @@ from typing import TYPE_CHECKING, Any, cast
import pydase.units as u
from pydase.data_service.data_service_cache import DataServiceCache
from pydase.utils.helpers import (
get_object_attr_from_path_list,
get_object_attr_from_path,
is_property_attribute,
parse_list_attr_and_index,
)
@ -235,23 +235,25 @@ class StateManager:
return float(value)
return value
def __update_attribute_by_path(self, path: str, value: Any) -> None:
parent_path_list, attr_name = path.split(".")[:-1], path.split(".")[-1]
def __update_attribute_by_path(
self, path: str, serialized_value: SerializedObject
) -> None:
parent_path, attr_name = ".".join(path.split(".")[:-1]), path.split(".")[-1]
# If attr_name corresponds to a list entry, extract the attr_name and the
# index
attr_name, index = parse_list_attr_and_index(attr_name)
# Update path to reflect the attribute without list indices
path = ".".join([*parent_path_list, attr_name])
path = f"{parent_path}.{attr_name}" if parent_path != "" else attr_name
attr_cache_type = get_nested_dict_by_path(self.cache_value, path)["type"]
# Traverse the object according to the path parts
target_obj = get_object_attr_from_path_list(self.service, parent_path_list)
target_obj = get_object_attr_from_path(self.service, parent_path)
if attr_cache_type in ("ColouredEnum", "Enum"):
enum_attr = get_object_attr_from_path_list(target_obj, [attr_name])
enum_attr = get_object_attr_from_path(target_obj, attr_name)
# take the value of the existing enum class
# TODO: this might break when you set a value from a different enum
if isinstance(value, enum.Enum):
@ -271,10 +273,11 @@ class StateManager:
attributes default to being loadable.
"""
parent_object = get_object_attr_from_path_list(
self.service, full_access_path.split(".")[:-1]
parent_path, attr_name = (
".".join(full_access_path.split(".")[:-1]),
full_access_path.split(".")[-1],
)
attr_name = full_access_path.split(".")[-1]
parent_object = get_object_attr_from_path(self.service, parent_path)
if is_property_attribute(parent_object, attr_name):
prop = getattr(type(parent_object), attr_name)

View File

@ -7,7 +7,7 @@ import socketio # type: ignore[import-untyped]
from pydase.data_service.data_service_observer import DataServiceObserver
from pydase.data_service.state_manager import StateManager
from pydase.utils.helpers import get_object_attr_from_path_list
from pydase.utils.helpers import get_object_attr_from_path
from pydase.utils.logging import SocketIOHandler
from pydase.utils.serialization.deserializer import Deserializer, loads
from pydase.utils.serialization.serializer import SerializedObject, dump
@ -154,8 +154,8 @@ def setup_sio_events(sio: socketio.AsyncServer, state_manager: StateManager) ->
@sio.event
async def trigger_method(sid: str, data: TriggerMethodDict) -> Any:
try:
method = get_object_attr_from_path_list(
state_manager.service, data["access_path"].split(".")
method = get_object_attr_from_path(
state_manager.service, data["access_path"]
)
args = Deserializer.deserialize(data["args"])
kwargs: dict[str, Any] = Deserializer.deserialize(data["kwargs"])

View File

@ -30,13 +30,13 @@ def get_class_and_instance_attributes(obj: object) -> dict[str, Any]:
return dict(chain(type(obj).__dict__.items(), obj.__dict__.items()))
def get_object_attr_from_path_list(target_obj: Any, path: list[str]) -> Any:
def get_object_attr_from_path(target_obj: Any, path: str) -> Any:
"""
Traverse the object tree according to the given path.
Args:
target_obj: The root object to start the traversal from.
path: A list of attribute names representing the path to traverse.
path: Access path of the object.
Returns:
The attribute at the end of the path. If the path includes a list index,
@ -46,7 +46,8 @@ def get_object_attr_from_path_list(target_obj: Any, path: list[str]) -> Any:
Raises:
ValueError: If a list index in the path is not a valid integer.
"""
for part in path:
path_list = path.split(".") if path != "" else []
for part in path_list:
try:
# Try to split the part into attribute and index
attr, index_str = part.split("[", maxsplit=1)