Merge pull request #14 from Enapter/rnovatorov/dev

implement `Device.run_in_thread`

Author: rnovatorov

.github/workflows/ci.yml

@@ -12,7 +12,7 @@ jobs:
     strategy:
       fail-fast: false
       matrix:
-        python: ["3.7", "3.8", "3.9", "3.10"]
+        python: ["3.8", "3.9", "3.10", "3.11"]
     steps:
       - name: Checkout
         uses: actions/checkout@v2

enapter/__init__.py

@@ -1,4 +1,4 @@
-__version__ = "0.7.3"
+__version__ = "0.8.0"
 
 from . import async_, log, mdns, mqtt, types, vucm
 

enapter/async_/__init__.py

@@ -1,9 +1,7 @@
 from .generator import generator
 from .routine import Routine
-from .run_in_executor import run_in_executor
 
 __all__ = [
     "generator",
     "Routine",
-    "run_in_executor",
 ]

enapter/async_/run_in_executor.py

@@ -1,4 +1,6 @@
 import asyncio
+import concurrent
+import functools
 import traceback
 from typing import Optional, Set
 
@@ -7,12 +9,22 @@
 
 
 class Device(async_.Routine):
-    def __init__(self, channel, cmd_prefix="cmd_", task_prefix="task_") -> None:
+    def __init__(
+        self,
+        channel,
+        cmd_prefix="cmd_",
+        task_prefix="task_",
+        thread_pool_executor=None,
+    ) -> None:
         self.__channel = channel
 
         self.__cmd_prefix = cmd_prefix
         self.__task_prefix = task_prefix
 
+        if thread_pool_executor is None:
+            thread_pool_executor = concurrent.futures.ThreadPoolExecutor(max_workers=1)
+        self.__thread_pool_executor = thread_pool_executor
+
         self.log = Logger(channel=channel)
         self.alerts: Set[str] = set()
 
@@ -34,7 +46,15 @@ async def send_properties(self, properties: Optional[types.JSON] = None) -> None
 
         await self.__channel.publish_properties(properties)
 
+    async def run_in_thread(self, func, *args, **kwargs):
+        loop = asyncio.get_running_loop()
+        return await loop.run_in_executor(
+            self.__thread_pool_executor, functools.partial(func, *args, **kwargs)
+        )
+
     async def _run(self):
+        self._stack.enter_context(self.__thread_pool_executor)
+
         tasks = set()
 
         for name in dir(self):

enapter/vucm/device.py

@@ -1,2 +1,2 @@
-enapter==0.7.3
+enapter==0.8.0
 psutil==5.9.4

examples/vucm/psutil-battery/requirements.txt

@@ -1 +1 @@
-enapter==0.7.3
+enapter==0.8.0

examples/vucm/rl6-simulator/requirements.txt

@@ -1,2 +1,2 @@
-enapter==0.7.3
+enapter==0.8.0
 python-weather==0.4.2

examples/vucm/wttr-in/requirements.txt

@@ -1,2 +1,2 @@
-enapter==0.7.3
+enapter==0.8.0
 python-miio==0.5.12

examples/vucm/zhimi-fan-za5/requirements.txt

@@ -0,0 +1,23 @@
+import asyncio
+import contextlib
+
+import enapter
+
+
+class TestDevice:
+    async def test_run_in_thread(self, fake):
+        async with enapter.vucm.Device(channel=MockChannel(fake)) as device:
+            assert not device._Device__thread_pool_executor._shutdown
+            assert await device.run_in_thread(lambda: 42) == 42
+        assert device._Device__thread_pool_executor._shutdown
+
+
+class MockChannel:
+    def __init__(self, fake):
+        self.hardware_id = fake.hardware_id()
+        self.channel_id = fake.channel_id()
+
+    @contextlib.asynccontextmanager
+    async def subscribe_to_command_requests(self, *args, **kwargs):
+        await asyncio.Event().wait()
+        yield