diff --git a/src/pydase/data_service/data_service_observer.py b/src/pydase/data_service/data_service_observer.py index a526486..4562b2e 100644 --- a/src/pydase/data_service/data_service_observer.py +++ b/src/pydase/data_service/data_service_observer.py @@ -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( diff --git a/src/pydase/data_service/state_manager.py b/src/pydase/data_service/state_manager.py index 5798abf..5e39d7b 100644 --- a/src/pydase/data_service/state_manager.py +++ b/src/pydase/data_service/state_manager.py @@ -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) diff --git a/src/pydase/server/web_server/sio_setup.py b/src/pydase/server/web_server/sio_setup.py index 6900e7f..29bb691 100644 --- a/src/pydase/server/web_server/sio_setup.py +++ b/src/pydase/server/web_server/sio_setup.py @@ -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"]) diff --git a/src/pydase/utils/helpers.py b/src/pydase/utils/helpers.py index 09f648f..3e29967 100644 --- a/src/pydase/utils/helpers.py +++ b/src/pydase/utils/helpers.py @@ -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)