ITQC/pydase
24
0
Fork 0
mirror of https://github.com/tiqi-group/pydase.git synced 2025-04-22 01:00:02 +02:00
Code Issues Packages Projects Releases Wiki Activity

refactores __update_attribute_by_path of StateManager

Browse Source
This commit is contained in:
Mose Müller 2023-11-09 11:36:45 +01:00
parent 8dd05ac5e3
commit 784d49d90c
1 changed files with 22 additions and 27 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

49
src/pydase/data_service/state_manager.py
View File

@ -1,12 +1,10 @@
import json import json
import logging import logging
import os import os
from enum import Enum
from pathlib import Path from pathlib import Path
from typing import TYPE_CHECKING, Any, Optional, cast from typing import TYPE_CHECKING, Any, Optional, cast
import pydase.units as u import pydase.units as u
from pydase.data_service.data_service import process_callable_attribute
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_object_attr_from_path_list, get_object_attr_from_path_list,
@ -110,10 +108,9 @@ class StateManager:
logger.debug("Could not load the service state.") logger.debug("Could not load the service state.")
return return
serialized_class = self.cache
for path in generate_serialized_data_paths(json_dict): for path in generate_serialized_data_paths(json_dict):
nested_json_dict = get_nested_dict_by_path(json_dict, path) nested_json_dict = get_nested_dict_by_path(json_dict, path)
nested_class_dict = get_nested_dict_by_path(serialized_class, path) nested_class_dict = get_nested_dict_by_path(self.cache, path)
value, value_type = nested_json_dict["value"], nested_json_dict["type"] value, value_type = nested_json_dict["value"], nested_json_dict["type"]
class_attr_value_type = nested_class_dict.get("type", None) class_attr_value_type = nested_class_dict.get("type", None)
@ -157,10 +154,9 @@ class StateManager:
current_value_dict = get_nested_dict_by_path(self.cache, path) current_value_dict = get_nested_dict_by_path(self.cache, path)
# This will also filter out methods as they are 'read-only'
if current_value_dict["readonly"]: if current_value_dict["readonly"]:
logger.debug( logger.debug(f"Attribute {path!r} is read-only. Ignoring new value...")
f"Attribute {path!r} is read-only. Ignoring value from JSON " "file..."
)
return return
converted_value = self.__convert_value_if_needed(value, current_value_dict) converted_value = self.__convert_value_if_needed(value, current_value_dict)
@ -194,28 +190,27 @@ class StateManager:
# index # index
attr_name, index = parse_list_attr_and_index(attr_name) 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])
attr_cache_type = get_nested_dict_by_path(self.cache, path)["type"]
# Traverse the object according to the path parts # 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_list(self.service, parent_path_list)
if self.__attr_value_should_change(target_obj, attr_name):
if attr_cache_type in ("ColouredEnum", "Enum"):
enum_attr = get_object_attr_from_path_list(target_obj, [attr_name])
setattr(target_obj, attr_name, enum_attr.__class__[value])
elif attr_cache_type == "list":
list_obj = get_object_attr_from_path_list(target_obj, [attr_name])
list_obj[index] = value
else:
setattr(target_obj, attr_name, value)
def __attr_value_should_change(self, parent_object: Any, attr_name: str) -> bool:
# If the attribute is a property, change it using the setter without getting # If the attribute is a property, change it using the setter without getting
# the property value (would otherwise be bad for expensive getter methods) # the property value (would otherwise be bad for expensive getter methods)
if is_property_attribute(target_obj, attr_name): if is_property_attribute(parent_object, attr_name):
setattr(target_obj, attr_name, value) return True
return return True
attr = get_object_attr_from_path(target_obj, [attr_name])
if attr is None:
# If the attribute does not exist, abort setting the value. An error
# message has already been logged.
# This will never happen as this function is only called when the
# attribute exists in the cache.
return
if isinstance(attr, Enum):
setattr(target_obj, attr_name, attr.__class__[value])
elif isinstance(attr, list) and index is not None:
attr[index] = value
elif callable(attr):
process_callable_attribute(attr, value["args"])
else:
setattr(target_obj, attr_name, value)