ITQC/pydase_service_base
24
0
Fork 0
mirror of https://github.com/tiqi-group/pydase_service_base.git synced 2025-06-08 05:20:41 +02:00
Code Issues Packages Projects Releases Wiki Activity

replaces get_object_attr_from_path_list with get_object_attr_from_path

Browse Source
This commit is contained in:
Mose Mueller 2024-04-08 14:58:24 +02:00
parent 02e93136fc
commit 0002a97beb
2 changed files with 32 additions and 15 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

22
pydase_service_base/ionizer_interface/ionizer_server.py
View File

@ -7,7 +7,16 @@ import pydase.components
import pydase.units as u import pydase.units as u
import tiqi_rpc import tiqi_rpc
from pydase.data_service.data_service_observer import DataServiceObserver from pydase.data_service.data_service_observer import DataServiceObserver
from pydase.utils.helpers import get_object_attr_from_path_list import pydase.version
if pydase.version.__major__ == 0 and pydase.version.__minor__ > 7:
from pydase.utils.helpers import get_object_attr_from_path
else:
from pydase.utils.helpers import get_object_attr_from_path_list
def get_object_attr_from_path(target_obj: Any, path: str) -> Any:
return get_object_attr_from_path_list(target_obj, path.split("."))
from pydase_service_base.ionizer_interface.rpc_interface import RPCInterface from pydase_service_base.ionizer_interface.rpc_interface import RPCInterface
@ -52,13 +61,16 @@ class IonizerServer:
if isinstance(value, u.Quantity): if isinstance(value, u.Quantity):
value = value.m value = value.m
if attr_name == "value": if attr_name == "value":
parent_object = get_object_attr_from_path_list( parent_path = ".".join(full_access_path.split(".")[:-1])
self.service, parent_path_list parent_object = get_object_attr_from_path(self.service, parent_path)
)
if isinstance(parent_object, pydase.components.NumberSlider): if isinstance(parent_object, pydase.components.NumberSlider):
# removes the "value" from name -> Ionizer does not know about the # removes the "value" from name -> Ionizer does not know about the
# internals of NumberSlider # internals of NumberSlider
full_access_path = ".".join(full_access_path.split(".")[:-1]) full_access_path = parent_path
if isinstance(parent_object, pydase.components.NumberSlider):
# removes the "value" from name -> Ionizer does not know about the
# internals of NumberSlider
full_access_path = parent_object
logger.debug( logger.debug(
"Updating Ionizer with %s", {"name": full_access_path, "value": value} "Updating Ionizer with %s", {"name": full_access_path, "value": value}

25
pydase_service_base/ionizer_interface/rpc_interface.py
View File

@ -6,7 +6,17 @@ from pydase import DataService
from pydase.components import NumberSlider from pydase.components import NumberSlider
from pydase.data_service.data_service_observer import DataServiceObserver from pydase.data_service.data_service_observer import DataServiceObserver
from pydase.units import Quantity from pydase.units import Quantity
from pydase.utils.helpers import get_object_attr_from_path_list import pydase.version
if pydase.version.__major__ == 0 and pydase.version.__minor__ > 7:
from pydase.utils.helpers import get_object_attr_from_path
else:
from pydase.utils.helpers import get_object_attr_from_path_list
def get_object_attr_from_path(target_obj: Any, path: str) -> Any:
return get_object_attr_from_path_list(target_obj, path.split("."))
from pydase.version import __version__ from pydase.version import __version__
@ -35,9 +45,7 @@ class RPCInterface:
This method is called when Ionizer initilizes the Plugin or refreshes. The This method is called when Ionizer initilizes the Plugin or refreshes. The
widgets need to store the full_access_path in their name attribute. widgets need to store the full_access_path in their name attribute.
""" """
param = get_object_attr_from_path_list( param = get_object_attr_from_path(self._service, full_access_path)
self._service, full_access_path.split(".")
)
if isinstance(param, NumberSlider): if isinstance(param, NumberSlider):
return param.value return param.value
if isinstance(param, DataService): if isinstance(param, DataService):
@ -54,8 +62,8 @@ class RPCInterface:
return param return param
async def set_param(self, full_access_path: str, value: Any) -> None: async def set_param(self, full_access_path: str, value: Any) -> None:
parent_path_list = full_access_path.split(".")[:-1] parent_path = ".".join(full_access_path.split(".")[:-1])
parent_object = get_object_attr_from_path_list(self._service, parent_path_list) parent_object = get_object_attr_from_path(self._service, parent_path)
attr_name = full_access_path.split(".")[-1] attr_name = full_access_path.split(".")[-1]
# I don't want to trigger the execution of a property getter as this might take # I don't want to trigger the execution of a property getter as this might take
# a while when connecting to remote devices # a while when connecting to remote devices
@ -79,10 +87,7 @@ class RPCInterface:
self._state_manager.set_service_attribute_value_by_path(full_access_path, value) self._state_manager.set_service_attribute_value_by_path(full_access_path, value)
async def remote_call(self, full_access_path: str, *args: Any) -> Any: async def remote_call(self, full_access_path: str, *args: Any) -> Any:
full_access_path_list = full_access_path.split(".") method_object = get_object_attr_from_path(self._service, full_access_path)
method_object = get_object_attr_from_path_list(
self._service, full_access_path_list
)
return method_object(*args) return method_object(*args)
async def emit(self, message: str) -> None: async def emit(self, message: str) -> None: