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

reduces complexity of DataServiceObserver functions

Browse Source
This commit is contained in:
Mose Müller 2023-12-05 11:35:58 +01:00
parent 75ee71cbf8
commit a9c6070ca3
1 changed files with 37 additions and 36 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

61
src/pydase/data_service/data_service_observer.py
View File

@ -87,7 +87,16 @@ class DataServiceObserver(Observer):
def _get_properties_and_their_dependencies(
self, obj: Observable, prefix: str = ""
) -> dict[str, list[str]]:
deps = {}
deps: dict[str, Any] = {}
self._process_observable_properties(obj, deps, prefix)
self._process_nested_observables_properties(obj, deps, prefix)
return deps
def _process_observable_properties(
self, obj: Observable, deps: dict[str, Any], prefix: str
) -> None:
for k, value in vars(type(obj)).items():
prefix = (
f"{prefix}." if prefix != "" and not prefix.endswith(".") else prefix
@ -96,51 +105,43 @@ class DataServiceObserver(Observer):
if isinstance(value, property):
deps[key] = get_property_dependencies(value, prefix)
def _process_nested_observables_properties(
self, obj: Observable, deps: dict[str, Any], prefix: str
) -> None:
for k, value in vars(obj).items():
prefix = (
f"{prefix}." if prefix != "" and not prefix.endswith(".") else prefix
)
key = f"{prefix}{k}"
parent_path = f"{prefix}{k}"
if isinstance(value, Observable):
new_prefix = f"{key}."
new_prefix = f"{parent_path}."
deps.update(
self._get_properties_and_their_dependencies(value, new_prefix)
)
elif isinstance(value, list):
for i, item in enumerate(value):
elif isinstance(value, list | dict):
self._process_collection_item_properties(value, deps, parent_path)
def _process_collection_item_properties(
self,
collection: list[Any] | dict[str, Any],
deps: dict[str, Any],
parent_path: str,
) -> None:
if isinstance(collection, list):
for i, item in enumerate(collection):
if isinstance(item, Observable):
new_prefix = f"{key}[{i}]"
new_prefix = f"{parent_path}[{i}]"
deps.update(
self._get_properties_and_their_dependencies(
item, new_prefix
self._get_properties_and_their_dependencies(item, new_prefix)
)
)
elif isinstance(value, dict):
for key, val in value.items():
elif isinstance(collection, dict):
for key, val in collection.items():
if isinstance(val, Observable):
new_prefix = f"{key}['{key}']"
new_prefix = f"{parent_path}['{key}']"
deps.update(
self._get_properties_and_their_dependencies(val, new_prefix)
)
return deps
def _get_property_values(
self, obj: Observable, prefix: str = ""
) -> dict[str, list[str]]:
values = {}
for k, value in vars(type(obj)).items():
key = f"{prefix}{k}"
if isinstance(value, property):
values[key] = getattr(obj, k)
for k, value in vars(obj).items():
key = f"{prefix}{k}"
if isinstance(value, Observable):
new_prefix = f"{key}." if not key.endswith("]") else key
values.update(self._get_property_values(value, new_prefix))
return values
def add_notification_callback(
self, callback: Callable[[str, Any, dict[str, Any]], None]
) -> None: