Add asyncio support

LICENSE.txt

@@ -1,6 +1,7 @@
 MIT License
 
 Copyright (c) 2016 Christian Sandberg
+Copyright (c) 2026 Svein Seldal
 
 Permission is hereby granted, free of charge, to any person obtaining a copy
 of this software and associated documentation files (the "Software"), to deal

README.rst

@@ -8,6 +8,80 @@ automation tasks rather than a standard compliant master implementation.
 
 The library supports Python 3.9 or newer.
 
+The library can be run in two modes: regular mode or in async mode. In regular
+mode, calls to the library may block until a response is ready. In async mode
+it is possible to read and write to more than one node at the same time
+without the need of multiple threads.
+
+
+Asyncio port
+------------
+
+The objective of the library is to provide a canopen implementation in
+either async or non-async environment, with suitable API for both.
+
+To minimize the impact of the async changes, this port is designed to use the
+existing synchronous backend of the library. This means that the library
+uses :code:`asyncio.to_thread()` for many asynchronous operations.
+
+This port remains compatible with using it in a regular non-asyncio
+environment. This is selected with the `loop` parameter in the
+:code:`Network` constructor. If you pass a valid asyncio event loop, the
+library will run in async mode. If you pass `loop=None`, it will run in
+regular blocking mode. It cannot be used in both modes at the same time.
+
+
+Difference between async and non-async version
+----------------------------------------------
+
+This port have some differences with the upstream non-async version of canopen.
+
+* The async use of :code:`Network` must be used in an async context. This is
+  required to setup the async tasks and handles proper cleanup and exception
+  handling.
+
+      async with canopen.Network().connect() as network:
+          # do async stuff with network
+
+* Most async functions follow an "a" prefix naming scheme.
+  E.g. the async variant for :code:`SdoClient.download()` is available
+  as :code:`SdoClient.adownload()`.
+
+* Variables in the regular canopen library uses properties for getting and
+  setting. This is replaced with awaitable methods in the async version.
+
+      var = sdo['Variable'].raw  # synchronous
+      sdo['Variable'].raw = 12   # synchronous
+
+      var = await sdo['Variable']          # async
+      var = await sdo['Variable'].aread()  # async (equivalent)
+      await sdo['Variable'].awrite(12)     # async
+
+* Opt-in :code:`ensure_not_async()` sentinel guard in functions which prevents
+  calling blocking functions in async context. It will raise the exception
+  :code:`RuntimeError` "Calling a blocking function" when this happen. If this
+  is encountered, the code is not using the async variants of the library when
+  it shouldn't.
+
+  To enable it call :code:`canopen.async_guard.enable_async_guard(True)` in
+  your main thread/main loop.
+
+* The callbacks to the message handlers have been changed to be handled by
+  :code:`Network.dispatch_callbacks()`. They are no longer called with any
+  locks held, as this would not work with async. This affects:
+    * :code:`PdoMaps.on_message`
+    * :code:`EmcyConsumer.on_emcy`
+    * :code:`NtmMaster.on_heartbaet`
+
+* SDO block upload and download is not yet supported in async mode.
+
+* :code:`BaseNode402` does not work with async
+
+* :code:`Bits` is not working differently in async mode. In non-async mode,
+  the raw value is read from the node when the :code:`Bits` object is created.
+  In async mode, the raw value must be manually read by calling
+  :code:`await bits.aread()` before accessing the bits.
+
 
 Features
 --------
@@ -156,6 +230,70 @@ The :code:`n` is the PDO index (normally 1 to 4). The second form of access is f
     network.disconnect()
 
 
+Asyncio
+-------
+
+This is the same example as above, but using 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'].awrite(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'].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'].awrite(42, "phys")
+            node.rpdo[1].transmit()
+
+    async def main():
+
+        # 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_running_loop()
+        async with canopen.Network(loop=loop).connect(
+                interface='pcan', bitrate=1000000) as network:
+
+            # 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,68 @@
+""" Utils for async """
+import asyncio
+import functools
+import logging
+import threading
+import traceback
+
+
+_ASYNC_GUARDS: dict[int, bool] = {}
+"""Per-thread boolean indicating allowance of running blocking functions.
+
+:code:`True` indicates that blocking functions are not allowed to be called
+from the current thread.
+"""
+
+logger = logging.getLogger(__name__)
+
+
+def enable_async_guard(enable: bool):
+    """Enable or disable the async guard for the current thread.
+
+    :param enable: True to enable the async guard, False to disable it.
+    """
+    _ASYNC_GUARDS[threading.get_ident()] = enable
+
+
+def is_async_guarded() -> bool:
+    """Check if async guard is enabled for this thread.
+
+    :return: True if async guard is enabled, False otherwise.
+    """
+    return _ASYNC_GUARDS.get(threading.get_ident(), False)
+
+
+def ensure_not_async(fn=None, error_message=None):
+    """Guard a function from being called from the async main thread.
+
+    This function is used to guard functions that are blocking and should not
+    be called from async code. If the function is called while async is
+    running, a RuntimeError will be raised.
+
+    Can be used either as a plain decorator, :code:`@ensure_not_async`, or
+    called with an extra error message, :code:`@ensure_not_async("message")`.
+
+    :param error_message: Optional message appended to the RuntimeError raised
+        when the guard trips.
+    """
+    if isinstance(fn, str):
+        fn, error_message = None, fn
+
+    def decorator(fn):
+        @functools.wraps(fn)
+        def async_guard_wrap(*args, **kwargs):
+            if is_async_guarded():
+                st = "".join(traceback.format_stack())
+                logger.debug("Traceback:\n%s", st.rstrip())
+                msg = ("Calling a blocking function while running async. "
+                       f"Function {fn.__qualname__}() "
+                       f"in {fn.__code__.co_filename}:{fn.__code__.co_firstlineno}")
+                if error_message:
+                    msg += f". {error_message}"
+                raise RuntimeError(msg)
+            return fn(*args, **kwargs)
+        return async_guard_wrap
+
+    if fn is not None:
+        return decorator(fn)
+    return decorator

canopen/emcy.py

@@ -1,11 +1,13 @@
 from __future__ import annotations
 
+import asyncio
 import logging
 import struct
 import threading
 import time
 from typing import Callable, Optional
 
+from canopen.async_guard import ensure_not_async
 import canopen.network
 
 
@@ -24,6 +26,7 @@ def __init__(self):
         self.active: list[EmcyError] = []
         self.callbacks = []
         self.emcy_received = threading.Condition()
+        self.network: canopen.network.Network = canopen.network._UNINITIALIZED_NETWORK
 
     def on_emcy(self, can_id, data, timestamp):
         code, register, data = EMCY_STRUCT.unpack(data)
@@ -38,11 +41,8 @@ def on_emcy(self, can_id, data, timestamp):
             self.log.append(entry)
             self.emcy_received.notify_all()
 
-        for callback in self.callbacks:
-            try:
-                callback(entry)
-            except Exception:
-                logger.exception("Exception in EMCY callback")
+        # Call all registered callbacks
+        self.network.dispatch_callbacks(self.callbacks, entry, ignore_errors=True)
 
     def add_callback(self, callback: Callable[[EmcyError], None]):
         """Get notified on EMCY messages from this node.
@@ -58,6 +58,7 @@ def reset(self):
         self.log = []
         self.active = []
 
+    @ensure_not_async("Use async_wait() instead")
     def wait(
         self, emcy_code: Optional[int] = None, timeout: float = 10
     ) -> Optional[EmcyError]:
@@ -86,6 +87,18 @@ def wait(
                     # This is the one we're interested in
                     return emcy
 
+    async def async_wait(
+        self, emcy_code: Optional[int] = None, timeout: float = 10
+    ) -> Optional[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,10 @@
+import asyncio
 import logging
 import queue
 import struct
 import time
 
+from canopen.async_guard import ensure_not_async
 import canopen.network
 
 
@@ -241,6 +243,7 @@ def send_identify_non_configured_remote_slave(self):
         message[0] = CS_IDENTIFY_NON_CONFIGURED_REMOTE_SLAVE
         self.__send_command(message)
 
+    @ensure_not_async("Use afast_scan() instead")
     def fast_scan(self):
         """This command sends a series of fastscan message
         to find unconfigured slave with lowest number of LSS idenities
@@ -282,6 +285,10 @@ def fast_scan(self):
 
         return False, None
 
+    async def afast_scan(self):
+        """Asynchronous version of fast_scan"""
+        return await asyncio.to_thread(self.fast_scan)
+
     def __send_fast_scan_message(self, id_number, bit_checker, lss_sub, lss_next):
         message = bytearray(8)
         message[0:8] = struct.pack('<BIBBB', CS_FAST_SCAN, id_number, bit_checker, lss_sub, lss_next)
@@ -296,6 +303,7 @@ def __send_fast_scan_message(self, id_number, bit_checker, lss_sub, lss_next):
 
         return False
 
+    @ensure_not_async
     def __send_lss_address(self, req_cs, number):
         message = bytearray(8)
 
@@ -359,6 +367,7 @@ def __send_configure(self, req_cs, value1=0, value2=0):
             error_msg = f"LSS Error: {error_code}"
             raise LssError(error_msg)
 
+    @ensure_not_async
     def __send_command(self, message):
         """Send a LSS operation code to the network