Repository: XKNX/xknx
Pull request #936 (2022-04-30 22:45:36)

Fix task registry for restarting tasks
<!--
  You are awesome! Thanks for contributing to our project!
  Please, DO NOT DELETE ANY TEXT from this template!.
-->
## Description
<!--
Please include a summary of the change and which issue is fixed.
Please also include relevant motivation and context.
List any dependencies that are required for this change.
-->
TaskRegistry takes functions returning coroutines instead of coroutines directly. This enables restarting tasks.

Without this we would get (if we awaited for)
```py
Task exception was never retrieved
future: <Task finished name='Task-3' coro=<test() done, defined at <stdin>:1> exception=RuntimeError('cannot reuse already awaited coroutine')>
RuntimeError: cannot reuse already awaited coroutine
```

Fixes # (issue)

## Type of change
<!--
Please tick the applicable options.
NOTE: Ticking multiple options most likely indicates
that your change is to big and it is suggested to split it into several smaller PRs.
-->

- [ ] Bug fix (non-breaking change which fixes an issue)
- [ ] New feature (non-breaking change which adds functionality)
- [ ] Breaking change (fix or feature that would cause existing functionality to not work as expected)
- [ ] This change requires a documentation update

## Checklist:

- [ ] I have performed a self-review of my own code
- [ ] The documentation has been adjusted accordingly
- [ ] The changes generate no new warnings
- [ ] Tests have been added that prove the fix is effective or that the feature works
- [ ] The changes are documented in the changelog
- [ ] The Homeassistant plugin has been adjusted in case of new config options


farmio: @marvin-w Should we rename the attribute `task` to avoid confusion?
codecov[bot]: # [Codecov](https://codecov.io/gh/XKNX/xknx/pull/936?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=XKNX) Report
> Merging [#936](https://codecov.io/gh/XKNX/xknx/pull/936?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=XKNX) (c946b3d) into [main](https://codecov.io/gh/XKNX/xknx/commit/6c5a440643c13f4cd48346d4bf59694a3f47031b?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=XKNX) (6c5a440) will **increase** coverage by `0.01%`.
> The diff coverage is `100.00%`.

[![Impacted file tree graph](https://codecov.io/gh/XKNX/xknx/pull/936/graphs/tree.svg?width=650&height=150&src=pr&token=irWbIygS84&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=XKNX)](https://codecov.io/gh/XKNX/xknx/pull/936?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=XKNX)

```diff
@@            Coverage Diff             @@
##             main     #936      +/-   ##
==========================================
+ Coverage   95.95%   95.96%   +0.01%     
==========================================
  Files         130      130              
  Lines        7785     7781       -4     
==========================================
- Hits         7470     7467       -3     
+ Misses        315      314       -1     
```


| [Impacted Files](https://codecov.io/gh/XKNX/xknx/pull/936?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=XKNX) | Coverage Δ | |
|---|---|---|
| [xknx/devices/light.py](https://codecov.io/gh/XKNX/xknx/pull/936/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=XKNX#diff-eGtueC9kZXZpY2VzL2xpZ2h0LnB5) | `99.60% <ø> (ø)` | |
| [xknx/core/task\_registry.py](https://codecov.io/gh/XKNX/xknx/pull/936/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=XKNX#diff-eGtueC9jb3JlL3Rhc2tfcmVnaXN0cnkucHk=) | `100.00% <100.00%> (+1.56%)` | :arrow_up: |
| [xknx/devices/binary\_sensor.py](https://codecov.io/gh/XKNX/xknx/pull/936/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=XKNX#diff-eGtueC9kZXZpY2VzL2JpbmFyeV9zZW5zb3IucHk=) | `98.92% <100.00%> (+0.01%)` | :arrow_up: |
| [xknx/devices/datetime.py](https://codecov.io/gh/XKNX/xknx/pull/936/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=XKNX#diff-eGtueC9kZXZpY2VzL2RhdGV0aW1lLnB5) | `95.23% <100.00%> (+0.11%)` | :arrow_up: |
| [xknx/devices/switch.py](https://codecov.io/gh/XKNX/xknx/pull/936/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=XKNX#diff-eGtueC9kZXZpY2VzL3N3aXRjaC5weQ==) | `100.00% <100.00%> (ø)` | |
| [xknx/xknx.py](https://codecov.io/gh/XKNX/xknx/pull/936/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=XKNX#diff-eGtueC94a254LnB5) | `88.88% <100.00%> (-0.12%)` | :arrow_down: |

marvin-w: > @marvin-w Should we rename the attribute `task` to avoid confusion?

Sure, if you want.

Patch:
diff --git a/changelog.md b/changelog.md
index 9e2ac1c4..858cc8a3 100644
--- a/changelog.md
+++ b/changelog.md
@@ -2,6 +2,14 @@
 
 ## Unreleased changes
 
+### Bugfixes
+
+- Fix exposure of datetime, time and date objects to the Bus again
+
+### Internals
+
+- TaskRegistry takes functions returning coroutines instead of coroutines directly
+
 ## 0.21.0 Search and connect 2022-04-30
 
 ### Discovery
diff --git a/xknx/core/task_registry.py b/xknx/core/task_registry.py
index b4f848b2..f306adfb 100644
--- a/xknx/core/task_registry.py
+++ b/xknx/core/task_registry.py
@@ -3,11 +3,11 @@ from __future__ import annotations
 
 import asyncio
 import logging
-from typing import TYPE_CHECKING, Any, Coroutine, Generator, Union
+from typing import TYPE_CHECKING, Any, Callable, Coroutine, Generator
 
 from xknx.core import XknxConnectionState
 
-AsyncCallbackType = Union[Generator[Any, None, Any], Coroutine[Any, Any, None]]
+AsyncCallbackType = Callable[[], Coroutine[Any, Any, None]]
 
 if TYPE_CHECKING:
     from xknx import XKNX
@@ -23,21 +23,16 @@ class Task:
         name: str,
         task: AsyncCallbackType,
         restart_after_reconnect: bool = False,
-        loop: asyncio.AbstractEventLoop | None = None,
     ) -> None:
         """Initialize Task class."""
         self.name = name
         self.task = task
         self.restart_after_reconnect = restart_after_reconnect
         self._task: asyncio.Task[None] | None = None
-        self._loop = loop
 
     def start(self) -> None:
         """Start a task."""
-        if self._loop is not None:
-            self._task = self._loop.create_task(self.task, name=self.name)
-        else:
-            self._task = asyncio.create_task(self.task, name=self.name)
+        self._task = asyncio.create_task(self.task(), name=self.name)
 
     def __await__(self) -> Generator[None, None, None]:
         """Wait for task to be finished."""
@@ -71,15 +66,9 @@ class TaskRegistry:
 
     def __init__(self, xknx: XKNX) -> None:
         """Initialize TaskRegistry class."""
-        self._main_loop: asyncio.AbstractEventLoop | None = None
-
         self.xknx = xknx
         self.tasks: list[Task] = []
 
-    async def register_loop(self) -> None:
-        """Register the main loop for thread safe interaction."""
-        self._main_loop = asyncio.get_running_loop()
-
     def register(
         self,
         name: str,
@@ -91,10 +80,7 @@ class TaskRegistry:
         self.unregister(name)
 
         _task: Task = Task(
-            name=name,
-            task=task,
-            restart_after_reconnect=restart_after_reconnect,
-            loop=self._main_loop,
+            name=name, task=task, restart_after_reconnect=restart_after_reconnect
         )
 
         if track_task:
diff --git a/xknx/devices/binary_sensor.py b/xknx/devices/binary_sensor.py
index dcc01426..d8813241 100644
--- a/xknx/devices/binary_sensor.py
+++ b/xknx/devices/binary_sensor.py
@@ -11,6 +11,7 @@ A BinarySensor may also have Actions attached which are executed after state was
 from __future__ import annotations
 
 import asyncio
+from functools import partial
 import time
 from typing import TYPE_CHECKING, Iterator, cast
 
@@ -92,7 +93,8 @@ class BinarySensor(Device):
             if self.ignore_internal_state and self._context_timeout:
                 self.bump_and_get_counter(state)
                 self._context_task = self.xknx.task_registry.register(
-                    self._context_task_name, self._counter_task(self._context_timeout)
+                    name=self._context_task_name,
+                    task=partial(self._counter_task, self._context_timeout),
                 )
                 self._context_task.start()
             else:
@@ -156,8 +158,8 @@ class BinarySensor(Device):
         """Create Task for resetting state if 'reset_after' is configured."""
         if self.reset_after is not None and self.state:
             self._reset_task = self.xknx.task_registry.register(
-                self._reset_task_name,
-                self._reset_state(self.reset_after),
+                name=self._reset_task_name,
+                task=partial(self._reset_state, self.reset_after),
                 track_task=True,
             )
             self._reset_task.start()
diff --git a/xknx/devices/datetime.py b/xknx/devices/datetime.py
index a3cf3eda..a18737d8 100644
--- a/xknx/devices/datetime.py
+++ b/xknx/devices/datetime.py
@@ -7,6 +7,7 @@ broadcasting localtime periodically.
 from __future__ import annotations
 
 import asyncio
+from functools import partial
 import time
 from typing import TYPE_CHECKING, Iterator
 
@@ -65,8 +66,8 @@ class DateTime(Device):
 
         if self.localtime:
             task: Task = self.xknx.task_registry.register(
-                f"datetime.broadcast_{id(self)}",
-                broadcast_loop(self, minutes=minutes),
+                name=f"datetime.broadcast_{id(self)}",
+                task=partial(broadcast_loop, self, minutes),
                 restart_after_reconnect=True,
             )
             task.start()
diff --git a/xknx/devices/light.py b/xknx/devices/light.py
index 6c919bdd..a1549f5d 100644
--- a/xknx/devices/light.py
+++ b/xknx/devices/light.py
@@ -368,7 +368,8 @@ class Light(Device):
         if self._individual_color_debounce_telegram_counter > 0:
             # task registry cancels existing task
             self.xknx.task_registry.register(
-                self._individual_color_debounce_task_name, debouncer()
+                name=self._individual_color_debounce_task_name,
+                task=debouncer,
             ).start()
             return
         self.xknx.task_registry.unregister(self._individual_color_debounce_task_name)
diff --git a/xknx/devices/switch.py b/xknx/devices/switch.py
index 2fe6aa76..fed84fd6 100644
--- a/xknx/devices/switch.py
+++ b/xknx/devices/switch.py
@@ -9,6 +9,7 @@ It provides functionality for
 from __future__ import annotations
 
 import asyncio
+from functools import partial
 import logging
 from typing import TYPE_CHECKING, Iterator
 
@@ -87,8 +88,8 @@ class Switch(Device):
         if await self.switch.process(telegram):
             if self.reset_after is not None and self.switch.value:
                 self._reset_task = self.xknx.task_registry.register(
-                    self._reset_task_name,
-                    self._reset_state(self.reset_after),
+                    name=self._reset_task_name,
+                    task=partial(self._reset_state, self.reset_after),
                     track_task=True,
                 )
                 self._reset_task.start()
diff --git a/xknx/xknx.py b/xknx/xknx.py
index 0f07fca7..0c0caa09 100644
--- a/xknx/xknx.py
+++ b/xknx/xknx.py
@@ -115,7 +115,7 @@ class XKNX:
         """Start XKNX module. Connect to KNX/IP devices and start state updater."""
         if self.connection_config.threaded:
             await self.connection_manager.register_loop()
-            await self.task_registry.register_loop()
+        self.task_registry.start()
         self.knxip_interface = knx_interface_factory(
             xknx=self,
             connection_config=self.connection_config,
@@ -128,7 +128,6 @@ class XKNX:
         await self.knxip_interface.start()
         await self.telegram_queue.start()
         self.state_updater.start()
-        self.task_registry.start()
         self.started.set()
         if self.daemon_mode:
             await self.loop_until_sigint()


Test patch:
diff --git a/test/core_tests/task_registry_test.py b/test/core_tests/task_registry_test.py
index 0cfcf44a..7c4b83e5 100644
--- a/test/core_tests/task_registry_test.py
+++ b/test/core_tests/task_registry_test.py
@@ -22,7 +22,7 @@ class TestTaskRegistry:
 
         task = xknx.task_registry.register(
             name="test",
-            task=callback(),
+            task=callback,
         )
         assert len(xknx.task_registry.tasks) == 1
         task.start()
@@ -39,7 +39,7 @@ class TestTaskRegistry:
 
         task = xknx.task_registry.register(
             name="test",
-            task=callback(),
+            task=callback,
         )
         assert len(xknx.task_registry.tasks) == 1
         task.start()
@@ -60,7 +60,7 @@ class TestTaskRegistry:
 
         task = xknx.task_registry.register(
             name="test",
-            task=callback(),
+            task=callback,
         )
         assert len(xknx.task_registry.tasks) == 1
         task.start()
@@ -70,7 +70,7 @@ class TestTaskRegistry:
     #
     # TEST CONNECTION HANDLING
     #
-    async def test_reconnect_handling(self):
+    async def test_reconnect_handling(self, time_travel):
         """Test reconnect handling."""
 
         xknx = XKNX()
@@ -79,27 +79,43 @@ class TestTaskRegistry:
         await xknx.connection_manager.connection_state_changed(
             XknxConnectionState.CONNECTED
         )
+        # pylint: disable=attribute-defined-outside-init
+        self.test = 0
 
         async def callback() -> None:
             """Reset tasks."""
-            await asyncio.sleep(100)
+            try:
+                while True:
+                    await asyncio.sleep(100)
+                    self.test += 1
+            except asyncio.CancelledError:
+                self.test -= 1
 
         task = xknx.task_registry.register(
-            name="test", task=callback(), restart_after_reconnect=True
+            name="test", task=callback, restart_after_reconnect=True
         )
         assert len(xknx.task_registry.tasks) == 1
         task.start()
+        assert task._task is not None
+        await time_travel(100)
+        assert self.test == 1
         await xknx.connection_manager.connection_state_changed(
             XknxConnectionState.DISCONNECTED
         )
         assert task._task is None
+        assert self.test == 0
         await xknx.connection_manager.connection_state_changed(
             XknxConnectionState.CONNECTED
         )
         assert task._task is not None
+        assert self.test == 0
+        await time_travel(100)
+        assert self.test == 1
         assert len(xknx.task_registry.tasks) == 1
 
         xknx.task_registry.stop()
         assert len(xknx.task_registry.tasks) == 0
         assert task._task is None
+        await asyncio.sleep(0)  # iterate loop to cancel task
+        assert self.test == 0
         assert len(xknx.connection_manager._connection_state_changed_cbs) == 0