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

tasks: need to bind method as soon as instance is passed to __get__

Browse Source 
I cannot keep a reference to the parent class as the Task class is a DataService, as well.
This commit is contained in:
Mose Müller 2024-08-05 16:40:24 +02:00
parent 09ceae90ec
commit 5f78771f66
1 changed files with 9 additions and 10 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

19
src/pydase/task/task.py
View File

@ -26,6 +26,7 @@ class Task(pydase.DataService, Generic[P, R]):
) -> None: ) -> None:
super().__init__() super().__init__()
self._func = func self._func = func
self._bound_func: Callable[P, Coroutine[None, None, R | None]] | None = None
self._task: asyncio.Task[R] | None = None self._task: asyncio.Task[R] | None = None
self._status = TaskStatus.NOT_RUNNING self._status = TaskStatus.NOT_RUNNING
self._result: R | None = None self._result: R | None = None
@ -61,25 +62,23 @@ class Task(pydase.DataService, Generic[P, R]):
self._result = task.result() self._result = task.result()
logger.info("Starting task") logger.info("Starting task")
if inspect.iscoroutinefunction(self._func): if inspect.iscoroutinefunction(self._bound_func):
logger.info("Is coroutine function.") res: Coroutine[None, None, R] = self._bound_func(*args, **kwargs)
res: Coroutine[None, None, R] = self._func(
self._parent_obj, *args, **kwargs
)
self._task = asyncio.create_task(res) self._task = asyncio.create_task(res)
self._task.add_done_callback(task_done_callback) self._task.add_done_callback(task_done_callback)
self._status = TaskStatus.RUNNING self._status = TaskStatus.RUNNING
else:
logger.info("Is not a coroutine function.")
def stop(self) -> None: def stop(self) -> None:
if self._task: if self._task:
self._task.cancel() self._task.cancel()
def __get__(self, obj: Any, obj_type: Any) -> Self: def __get__(self, instance: Any, owner: Any) -> Self:
# need to use this descriptor to bind the function to the instance of the class # need to use this descriptor to bind the function to the instance of the class
# containing the function # containing the function
if instance:
if obj is not None: async def bound_func(*args, **kwargs) -> R | None:
self._parent_obj = obj return await self._func(instance, *args, **kwargs)
self._bound_func = bound_func
return self return self