Merge pull request #15 from rambo/rambo

Improve efficiency (remove busy-loops, use events instead)

Author: rambo

poetry.lock

@@ -46,7 +46,6 @@ branch = true
 [tool.poetry.dependencies]
 python = "^3.6"
 pyserial = "^3.4"
-async-timeout = "^3.0"
 
 [tool.poetry.dev-dependencies]
 pytest = "^6.2"

pyproject.toml

@@ -3,8 +3,6 @@
 import decimal
 import re
 
-from async_timeout import timeout
-
 from .errors import CommandError
 from .transports.baseclass import AbstractTransport
 
@@ -175,14 +173,19 @@ async def check_error(self, prev_command=""):
     async def command(self, command, cmd_timeout=COMMAND_DEFAULT_TIMEOUT, abort_on_timeout=True):
         """Sends a command, does not wait for response"""
         try:
-            with timeout(cmd_timeout):
+
+            async def _command() -> None:
+                """Wrap the actual work"""
+                nonlocal self
                 async with self.lock:
                     await self.transport.send_command(command)
+
+            await asyncio.wait_for(_command(command), timeout=cmd_timeout)
         except (asyncio.TimeoutError, asyncio.CancelledError) as err:
             # check for the actual error if available
             await self.check_error(command)
             if abort_on_timeout:
-                self.abort_command()
+                await self.abort_command()
             # re-raise the timeout if no other error found
             raise err
         # other errors are allowed to bubble-up as-is
@@ -195,11 +198,14 @@ async def safe_command(self, command, *args, **kwargs):
     async def ask(self, command, cmd_timeout=COMMAND_DEFAULT_TIMEOUT, abort_on_timeout=True):
         """Send a command and waits for response, returns the response"""
         try:
-            with timeout(cmd_timeout):
+
+            async def _ask(command: str) -> str:
+                """Wrap the actual work"""
                 async with self.lock:
                     await self.transport.send_command(command)
                     return await self.transport.get_response()
 
+            return await asyncio.wait_for(_ask(command), timeout=cmd_timeout)
         except (asyncio.TimeoutError, asyncio.CancelledError) as err:
             # check for the actual error if available
             await self.check_error(command)

src/scpi/scpi.py

@@ -4,6 +4,7 @@
 """
 import asyncio
 import logging
+import threading
 
 logger = logging.getLogger(__name__)
 
@@ -19,6 +20,8 @@ class BaseTransport(AbstractTransport):
     message_callback = None
     unsolicited_message_callback = None
     lock = asyncio.Lock()
+    aioevent = asyncio.Event()
+    blevent = threading.Event()
 
     async def quit(self):
         """Must shutdown all background threads (if any)"""

src/scpi/transports/baseclass.py

@@ -1,15 +1,17 @@
 """"Driver" for http://prologix.biz/gpib-usb-controller.html GPIB controller"""
 import asyncio
+import logging
+import threading
 
 import serial
 import serial.threaded
-from async_timeout import timeout
 
 from ..rs232 import RS232SerialProtocol
 from .base import GPIBTransport
 
 SCAN_DEVICE_TIMEOUT = 0.5
 READ_TIMEOUT = 1.0
+LOGGER = logging.getLogger(__name__)
 
 
 class PrologixRS232SerialProtocol(RS232SerialProtocol):
@@ -61,30 +63,35 @@ async def get_response(self):
     async def send_and_read(self, send):
         """Send a line, read the response. NOTE: This is for talking with the controller, device responses
         need to use get_response as usual"""
-        with timeout(READ_TIMEOUT):
+
+        async def _send_and_read(send) -> str:
+            """Wrap the actual work"""
             async with self.lock:
                 response = None
 
                 def set_response(message):
                     """Callback for setting the response"""
-                    nonlocal response
+                    nonlocal response, self
                     response = message
+                    self.blevent.set()
 
+                self.blevent.clear()
                 self.message_callback = set_response
                 self.serialhandler.protocol.write_line(send)
-                while response is None:
-                    await asyncio.sleep(0)
+                await asyncio.get_event_loop().run_in_executor(None, self.blevent.wait)
                 return response
 
+        return await asyncio.wait_for(_send_and_read(send), timeout=READ_TIMEOUT)
+
     async def set_address(self, address, secondary=None):
         """Set the address we want to talk to"""
         if secondary is None:
             await self.send_command("++addr %d" % address)
         else:
             await self.send_command("++addr %d %d" % (address, secondary))
-        # Wait for the address to actually be set
+
         while True:
-            await asyncio.sleep(0)
+            await asyncio.sleep(0.001)
             resp = await self.query_address()
             if resp == (address, secondary):
                 break
@@ -144,13 +151,18 @@ async def scan_devices(self):
         prev_read_tmo_ms = await self.send_and_read("++read_tmo_ms")
         self.serialhandler.protocol.write_line("++read_tmo_ms %d" % int((SCAN_DEVICE_TIMEOUT / 2) * 1000))
         for addr in range(0, 31):  # 0-30 inclusive
-            with timeout(SCAN_DEVICE_TIMEOUT):
-                try:
-                    await self.set_address(addr)
-                    await self.poll()
-                    found_addresses.append(addr)
-                except (asyncio.TimeoutError, asyncio.CancelledError):
-                    pass
+
+            async def _scan_addr(addr) -> None:
+                """Sacn single address"""
+                nonlocal found_addresses
+                await self.set_address(addr)
+                await self.poll()
+                found_addresses.append(addr)
+
+            try:
+                await asyncio.wait_for(_scan_addr(addr), timeout=SCAN_DEVICE_TIMEOUT)
+            except (asyncio.TimeoutError, asyncio.CancelledError):
+                pass
         self.serialhandler.protocol.write_line("++read_tmo_ms " + prev_read_tmo_ms)
         # Wait a moment for things to settle
         await asyncio.sleep(float(prev_read_tmo_ms) / 1000)

src/scpi/transports/gpib/prologix.py

@@ -51,12 +51,13 @@ async def get_response(self):
 
             def set_response(message):
                 """Callback for setting the response"""
-                nonlocal response
+                nonlocal response, self
                 response = message
+                self.blevent.set()
 
+            self.blevent.clear()
             self.message_callback = set_response
-            while response is None:
-                await asyncio.sleep(0)
+            await asyncio.get_event_loop().run_in_executor(None, self.blevent.wait)
             return response
 
     async def abort_command(self):