Add asyncio support [WIP]

README.rst

@@ -1,5 +1,5 @@
-CANopen for Python
-==================
+CANopen for Python, asyncio port
+================================
 
 A Python implementation of the CANopen_ standard.
 The aim of the project is to support the most common parts of the CiA 301
@@ -8,6 +8,41 @@ automation tasks rather than a standard compliant master implementation.
 
 The library supports Python 3.8 or newer.
 
+This library is the asyncio port of CANopen. See below for code example.
+
+
+Branch notes
+------------
+This branch is work in progress, where the intent is to concept test running
+the backend callbacks and unchanged from the sync version. The sync-async
+crossing is done via sync waiting via `asyncio.to_thread()` in each class
+that needs it.
+
+The goal was to simplify the impact of the async changes. Having an async
+backend requires a lot of duplication of code.
+
+
+Async status
+------------
+
+The remaining work for feature complete async implementation:
+
+* Implement :code:`ABlockUploadStream`, :code:`ABlockDownloadStream` and
+  :code:`ATextIOWrapper` for async in :code:`SdoClient`
+
+* Implement :code:`EcmyConsumer.wait()` for async
+
+* Implement async in :code:`LssMaster``
+
+* Async implementation of :code:`BaseNode402`
+
+* Implement async variant of :code:`Network.add_node`. This will probably also
+  add need of async variant of :code:`input_from_node` in eds.py
+
+* Update unittests for async
+
+* Update documentation and examples
+
 
 Features
 --------
@@ -156,6 +191,72 @@ The :code:`n` is the PDO index (normally 1 to 4). The second form of access is f
     network.disconnect()
 
 
+Asyncio
+-------
+
+This library can be used with asyncio.
+
+.. code-block:: python
+
+    import asyncio
+    import canopen
+    import can
+
+    async def my_node(network, nodeid, od):
+
+        # Create the node object and load the OD
+        node = network.add_node(nodeid, od)
+
+        # Read the PDOs from the remote
+        await node.tpdo.aread()
+        await node.rpdo.aread()
+
+        # Set the module state
+        node.nmt.set_state('OPERATIONAL')
+
+        # Set motor speed via SDO
+        await node.sdo['MotorSpeed'].aset_raw(2)
+
+        while True:
+
+            # Wait for TPDO 1
+            t = await node.tpdo[1].await_for_reception(1)
+            if not t:
+                continue
+
+            # Get the TPDO 1 value
+            rpm = node.tpdo[1]['MotorSpeed Actual'].get_raw()
+            print(f'SPEED on motor {nodeid}:', rpm)
+
+            # Sleep a little
+            await asyncio.sleep(0.2)
+
+            # Send RPDO 1 with some data
+            node.rpdo[1]['Some variable'].set_phys(42)
+            node.rpdo[1].transmit()
+
+    async def main():
+
+        # Start with creating a network representing one CAN bus
+        network = canopen.Network()
+
+        # Connect to the CAN bus
+        # Arguments are passed to python-can's can.Bus() constructor
+        # (see https://python-can.readthedocs.io/en/latest/bus.html).
+        # Note the loop parameter to enable asyncio operation
+        loop = asyncio.get_event_loop()
+        network.connect(interface='pcan', bitrate=1000000, loop=loop)
+
+        # Create two independent tasks for two nodes 51 and 52 which will run concurrently
+        task1 = asyncio.create_task(my_node(network, 51, '/path/to/object_dictionary.eds'))
+        task2 = asyncio.create_task(my_node(network, 52, '/path/to/object_dictionary.eds'))
+
+        # Wait for both to complete (which will never happen)
+        await asyncio.gather((task1, task2))
+
+    asyncio.run(main())
+
+
 Debugging
 ---------
 

canopen/async_guard.py

@@ -0,0 +1,30 @@
+""" Utils for async """
+import functools
+import logging
+import threading
+import traceback
+
+# NOTE: Global, but needed to be able to use ensure_not_async() in
+#       decorator context.
+_ASYNC_SENTINELS: dict[int, bool] = {}
+
+logger = logging.getLogger(__name__)
+
+
+def set_async_sentinel(enable: bool):
+    """ Register a function to validate if async is running """
+    _ASYNC_SENTINELS[threading.get_ident()] = enable
+
+
+def ensure_not_async(fn):
+    """ Decorator that will ensure that the function is not called if async
+        is running.
+    """
+    @functools.wraps(fn)
+    def async_guard_wrap(*args, **kwargs):
+        if _ASYNC_SENTINELS.get(threading.get_ident(), False):
+            st = "".join(traceback.format_stack())
+            logger.debug("Traceback:\n%s", st.rstrip())
+            raise RuntimeError(f"Calling a blocking function, {fn.__qualname__}() in {fn.__code__.co_filename}:{fn.__code__.co_firstlineno}, while running async")
+        return fn(*args, **kwargs)
+    return async_guard_wrap

canopen/emcy.py

@@ -1,9 +1,12 @@
+from __future__ import annotations
+import asyncio
 import logging
 import struct
 import threading
 import time
 from typing import Callable, List, Optional
 
+from canopen.async_guard import ensure_not_async
 import canopen.network
 
 
@@ -17,16 +20,20 @@
 
     def __init__(self):
         #: Log of all received EMCYs for this node
-        self.log: List["EmcyError"] = []
+        self.log: List[EmcyError] = []
         #: Only active EMCYs. Will be cleared on Error Reset
-        self.active: List["EmcyError"] = []
+        self.active: List[EmcyError] = []
         self.callbacks = []
         self.emcy_received = threading.Condition()
+        self.network: canopen.network.Network = canopen.network._UNINITIALIZED_NETWORK
 
+    # @callback  # NOTE: called from another thread
+    @ensure_not_async  # NOTE: Safeguard for accidental async use
     def on_emcy(self, can_id, data, timestamp):
         code, register, data = EMCY_STRUCT.unpack(data)
         entry = EmcyError(code, register, data, timestamp)
 
+        # NOTE: Blocking lock
         with self.emcy_received:
             if code & 0xFF00 == 0:
                 # Error reset
@@ -36,10 +43,10 @@
             self.log.append(entry)
             self.emcy_received.notify_all()
 
-        for callback in self.callbacks:
-            callback(entry)
+        # Call all registered callbacks
+        self.network.dispatch_callbacks(self.callbacks, entry)
 
-    def add_callback(self, callback: Callable[["EmcyError"], None]):
+    def add_callback(self, callback: Callable[[EmcyError], None]):
         """Get notified on EMCY messages from this node.
 
         :param callback:
@@ -53,9 +60,10 @@
         self.log = []
         self.active = []
 
+    @ensure_not_async  # NOTE: Safeguard for accidental async use
     def wait(
         self, emcy_code: Optional[int] = None, timeout: float = 10
-    ) -> "EmcyError":
+    ) -> EmcyError:
         """Wait for a new EMCY to arrive.
 
         :param emcy_code: EMCY code to wait for
@@ -65,8 +73,10 @@
         """
         end_time = time.time() + timeout
         while True:
+            # NOTE: Blocking lock
             with self.emcy_received:
                 prev_log_size = len(self.log)
+                # NOTE: Blocking call
                 self.emcy_received.wait(timeout)
                 if len(self.log) == prev_log_size:
                     # Resumed due to timeout
@@ -81,6 +91,18 @@
                     # This is the one we're interested in
                     return emcy
 
+    async def async_wait(
+        self, emcy_code: Optional[int] = None, timeout: float = 10
+    ) -> EmcyError:
+        """Wait for a new EMCY to arrive.
+
+        :param emcy_code: EMCY code to wait for
+        :param timeout: Max time in seconds to wait
+
+        :return: The EMCY exception object or None if timeout
+        """
+        return await asyncio.to_thread(self.wait, emcy_code, timeout)
+
 
 class EmcyProducer:
 

canopen/lss.py

@@ -1,8 +1,11 @@
+import asyncio
 import logging
 import queue
 import struct
 import time
+from typing import Optional, TYPE_CHECKING
 
+from canopen.async_guard import ensure_not_async
 import canopen.network
 
 
@@ -241,6 +244,8 @@ def send_identify_non_configured_remote_slave(self):
         message[0] = CS_IDENTIFY_NON_CONFIGURED_REMOTE_SLAVE
         self.__send_command(message)
 
+    # FIXME: Make async implementation "afast_scan"
+    @ensure_not_async  # NOTE: Safeguard for accidental async use
     def fast_scan(self):
         """This command sends a series of fastscan message
         to find unconfigured slave with lowest number of LSS idenities
@@ -257,6 +262,7 @@ def fast_scan(self):
         lss_next = 0
 
         if self.__send_fast_scan_message(lss_id[0], lss_bit_check, lss_sub, lss_next):
+            # NOTE: Blocking call
             time.sleep(0.01)
             while lss_sub < 4:
                 lss_bit_check = 32
@@ -266,12 +272,14 @@ def fast_scan(self):
                     if not self.__send_fast_scan_message(lss_id[lss_sub], lss_bit_check, lss_sub, lss_next):
                         lss_id[lss_sub] |= 1<<lss_bit_check
 
+                    # NOTE: Blocking call
                     time.sleep(0.01)
 
                 lss_next = (lss_sub + 1) & 3
                 if not self.__send_fast_scan_message(lss_id[lss_sub], lss_bit_check, lss_sub, lss_next):
                     return False, None
 
+                # NOTE: Blocking call
                 time.sleep(0.01)
 
                 # Now the next 32 bits will be scanned
@@ -296,6 +304,8 @@ def __send_fast_scan_message(self, id_number, bit_checker, lss_sub, lss_next):
 
         return False
 
+    # FIXME: Make async implementation "__asend_lss_address"
+    @ensure_not_async  # NOTE: Safeguard for accidental async use
     def __send_lss_address(self, req_cs, number):
         message = bytearray(8)
 
@@ -304,6 +314,7 @@ def __send_lss_address(self, req_cs, number):
         response = self.__send_command(message)
         # some device needs these delays between messages
         # because it can't handle messages arriving with no delay
+        # NOTE: Blocking call
         time.sleep(0.2)
 
         return response
@@ -359,6 +370,9 @@ def __send_configure(self, req_cs, value1=0, value2=0):
             error_msg = f"LSS Error: {error_code}"
             raise LssError(error_msg)
 
+    # FIXME: Make async implementation "__asend_command"
+
+    @ensure_not_async  # NOTE: Safeguard for accidental async use
     def __send_command(self, message):
         """Send a LSS operation code to the network
 
@@ -376,6 +390,7 @@ def __send_command(self, message):
         response = None
         if not self.responses.empty():
             logger.info("There were unexpected messages in the queue")
+            # FIXME: Recreating the queue
             self.responses = queue.Queue()
 
         self.network.send_message(self.LSS_TX_COBID, message)
@@ -386,14 +401,18 @@ def __send_command(self, message):
         # Wait for the slave to respond
         # TODO check if the response is LSS response message
         try:
+            # NOTE: Blocking call
             response = self.responses.get(
                 block=True, timeout=self.RESPONSE_TIMEOUT)
         except queue.Empty:
             raise LssError("No LSS response received")
 
         return response
 
+    # @callback  # NOTE: called from another thread
+    @ensure_not_async  # NOTE: Safeguard for accidental async use
     def on_message_received(self, can_id, data, timestamp):
+        # NOTE: Blocking call
         self.responses.put(bytes(data))