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

chore: moving task-based things to TaskManager

Browse Source
This commit is contained in:
Mose Müller 2023-08-02 12:06:21 +02:00
parent 3e925c7087
commit 9ffd666085
2 changed files with 15 additions and 15 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

15
src/pyDataInterface/data_service/data_service.py
View File

@ -54,10 +54,6 @@ class DataService(rpyc.Service, TaskManager):
"""Keep track of the root object. This helps to filter the emission of
notifications. This overwrite the TaksManager's __root__ attribute."""
self._autostart_tasks: dict[str, tuple[Any]]
if "_autostart_tasks" not in self.__dict__:
self._autostart_tasks = {}
self._callbacks: set[Callable[[str, Any], None]] = set()
self._register_callbacks()
@ -102,17 +98,6 @@ class DataService(rpyc.Service, TaskManager):
# allow all other attributes
setattr(self, name, value)
def _start_autostart_tasks(self) -> None:
if self._autostart_tasks is not None:
for service_name, args in self._autostart_tasks.items():
start_method = getattr(self, f"start_{service_name}", None)
if start_method is not None and callable(start_method):
start_method(*args)
else:
logger.warning(
f"No start method found for service '{service_name}'"
)
def _register_callbacks(self) -> None:
self._register_list_change_callbacks(self, f"{self.__class__.__name__}")
self._register_DataService_instance_callbacks(

15
src/pyDataInterface/data_service/task_manager.py
View File

@ -39,6 +39,10 @@ class TaskManager:
self._loop = asyncio.get_event_loop()
self._autostart_tasks: dict[str, tuple[Any]]
if "_autostart_tasks" not in self.__dict__:
self._autostart_tasks = {}
self._tasks: dict[str, TaskDict] = {}
"""A dictionary to keep track of running tasks. The keys are the names of the
tasks and the values are TaskDict instances which include the task itself and
@ -157,6 +161,17 @@ class TaskManager:
setattr(self, f"start_{name}", start_task)
setattr(self, f"stop_{name}", stop_task)
def _start_autostart_tasks(self) -> None:
if self._autostart_tasks is not None:
for service_name, args in self._autostart_tasks.items():
start_method = getattr(self, f"start_{service_name}", None)
if start_method is not None and callable(start_method):
start_method(*args)
else:
logger.warning(
f"No start method found for service '{service_name}'"
)
@abstractmethod
def _emit_notification(self, parent_path: str, name: str, value: Any) -> None:
raise NotImplementedError