Fixes from testing

* Use Network.taskgroup to administer asyncio tasks
* Network requires running in an async context to work
* Fixup error messages
* Fix disrepencies between sync and async versions
* Documentation update

Author: sveinse

canopen/emcy.py

@@ -42,7 +42,7 @@ def on_emcy(self, can_id, data, timestamp):
             self.emcy_received.notify_all()
 
         # Call all registered callbacks
-        self.network.dispatch_callbacks(self.callbacks, entry)
+        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.
@@ -89,7 +89,7 @@ def wait(
 
     async def async_wait(
         self, emcy_code: Optional[int] = None, timeout: float = 10
-    ) -> EmcyError:
+    ) -> Optional[EmcyError]:
         """Wait for a new EMCY to arrive.
 
         :param emcy_code: EMCY code to wait for

canopen/network.py

@@ -3,8 +3,9 @@
 import asyncio
 import logging
 import threading
-from collections.abc import Iterator, MutableMapping
+from collections.abc import Coroutine, Iterator, MutableMapping
 from typing import Callable, Final, Optional, Union
+import sys
 
 import can
 
@@ -17,6 +18,12 @@
 from canopen.sync import SyncProducer
 from canopen.timestamp import TimeProducer
 
+# Use backported TaskGroup for Python < 3.13, as asyncio.TaskGroup was added in 3.11
+if sys.version_info < (3, 11):
+    from taskgroup import TaskGroup
+else:
+    from asyncio import TaskGroup
+
 
 logger = logging.getLogger(__name__)
 
@@ -28,6 +35,7 @@ class Network(MutableMapping):
 
     NOTIFIER_CYCLE: float = 1.0  #: Maximum waiting time for one notifier iteration.
     NOTIFIER_SHUTDOWN_TIMEOUT: float = 5.0  #: Maximum waiting time to stop notifiers.
+    FILTER_ERRORS: bool = True  #: If True, exceptions in callbacks will be logged only
 
     def __init__(self, bus: Optional[can.BusABC] = None,
                  notifier: Optional[can.Notifier] = None,
@@ -40,7 +48,6 @@ def __init__(self, bus: Optional[can.BusABC] = None,
         #: :meth:`canopen.Network.connect` is called
         self.bus = bus
         self.loop = loop
-        self._futures: set[asyncio.Future] = set()
         #: A :class:`~canopen.network.NodeScanner` for detecting nodes
         self.scanner = NodeScanner(self)
         #: List of :class:`can.Listener` objects.
@@ -50,6 +57,8 @@ def __init__(self, bus: Optional[can.BusABC] = None,
         self.nodes: dict[int, Union[RemoteNode, LocalNode]] = {}
         self.subscribers: dict[int, list[Callback]] = {}
         self.send_lock = threading.Lock()
+        self.taskgroup: TaskGroup = TaskGroup()
+        self.thread_id: int = threading.get_ident()
         self.sync = SyncProducer(self)
         self.time = TimeProducer(self)
         self.nmt = NmtMaster(0)
@@ -114,9 +123,14 @@ def connect(self, *args, **kwargs) -> Network:
             self.bus = can.Bus(*args, **kwargs)
         logger.info("Connected to '%s'", self.bus.channel_info)
         if self.notifier is None:
-            self.notifier = can.Notifier(self.bus, [], self.NOTIFIER_CYCLE)
-        for listener in self.listeners:
-            self.notifier.add_listener(listener)
+            # The notifier is started without setting the loop paramter, even
+            # when running in async mode. The notifier changes in sublte ways
+            # when the loop parameter is set. All callbacks via the Listener
+            # interface will be called from the separate rx thread, which is
+            # what canopen is designed for. The async mode of the notifier will
+            # send all callbacks to the event loop thread, which is not
+            # compatible with the blocking locks and queues used in canopen.
+            self.notifier = can.Notifier(self.bus, self.listeners, self.NOTIFIER_CYCLE)
         return self
 
     def disconnect(self) -> None:
@@ -145,15 +159,49 @@ def __exit__(self, type, value, traceback):
         self.disconnect()
 
     async def __aenter__(self):
-        # FIXME: When TaskGroup are available, we should use them to manage the
-        # tasks. The user must use the `async with` statement with the Network
-        # to ensure its created.
+        if self.loop is None:
+            self.loop = asyncio.get_running_loop()
+        self.thread_id = threading.get_ident()
+        await self.taskgroup.__aenter__()
         return self
 
     async def __aexit__(self, type, value, traceback):
+        await self.taskgroup.__aexit__(None, None, None)
         self.disconnect()
 
-    async def enable_async_guard(self, enable: bool = True) -> None:
+    @property
+    def is_async(self) -> bool:
+        """Check if canopen has been connected with async"""
+        return self.loop is not None
+
+    def create_task(self, coro: Coroutine, *args, **kwarge) -> asyncio.Task:
+        """Create an async task.
+
+        This function is thread-safe and can be called from any thread. If
+        called from the same thread as the event loop, it will use
+        :code:`asyncio.create_task()` directly. If called from a different
+        thread, it will use :code:`asyncio.run_coroutine_threadsafe()` to
+        schedule the task in the event loop.
+
+        :param coro:
+            The coroutine to run in the event loop.
+        """
+
+        if threading.get_ident() == self.thread_id:
+            # If we are running in the same thread as the event loop
+            # asyncio.create_task() can be used directly.
+            return self.taskgroup.create_task(coro, *args, **kwarge)
+
+        else:
+            async def _create_task():
+                return self.taskgroup.create_task(coro, *args, **kwarge)
+            # The .result() will block until the task is created in
+            # the event loop and return the task object.
+            future = asyncio.run_coroutine_threadsafe(_create_task(), self.loop)
+            return future.result()
+
+    @staticmethod
+    async def enable_async_guard(enable: bool = True) -> None:
         """Enable or disable the async guard for this network.
 
         This makes sure that all functions that are decorated with
@@ -186,6 +234,19 @@ def add_node(
         :param upload_eds:
             Set ``True`` if EDS file should be uploaded from 0x1021.
 
+            .. note::
+                Using this option will fail in async mode, since uploading the
+                EDS requires blocking SDO transfers during node setup. Use a
+                pre-fetched ``object_dictionary`` instead when running under
+                asyncio.
+
+            Example of pre-fetching the object dictionary with async:
+
+            .. code-block:: python
+
+                od = await aimport_from_node(node_id, network)
+                node = network.add_node(node_id, od)
+
         :return:
             The Node object that was added.
         """
@@ -197,20 +258,6 @@ def add_node(
         self[node.id] = node
         return node
 
-    async def aadd_node(
-        self,
-        node: Union[int, RemoteNode, LocalNode],
-        object_dictionary: Union[str, ObjectDictionary, None] = None,
-        upload_eds: bool = False,
-    ) -> RemoteNode:
-        """Add a remote node to the network, async variant.
-
-        See add_node() for description
-        """
-        # The async variant exists because import_from_node might block
-        return await asyncio.to_thread(self.add_node, node,
-                                       object_dictionary, upload_eds)
-
     def create_node(
         self,
         node: Union[int, LocalNode],
@@ -296,40 +343,54 @@ def notify(self, can_id: int, data: bytearray, timestamp: float) -> None:
             self.dispatch_callbacks(self.subscribers[can_id], can_id, data, timestamp)
         self.scanner.on_message_received(can_id)
 
-    def on_error(self, exc: BaseException) -> None:
-        """This method is called to handle any exception in the callbacks."""
-
-        # Exceptions in any callbaks should not affect CAN processing
+    def on_error(self, exc: BaseException, ignore_errors: Optional[bool] = None) -> None:
+        """Handle any exception in the callbacks.
+
+        With self.FILTER_ERRORS set to True, exceptions in callbacks will be logged
+        only, and the program will continue running. This is useful for
+        production systems where you want to log errors but not crash the
+        entire application due to a single callback failure.
+
+        With self.FILTER_ERRORS set to False, exceptions in callbacks will be raised,
+        which will stop the program. This is useful for development and debugging,
+        where you want to catch errors early and fix them. This is also
+        important for unit tests, as only logging errors may hide problems.
+
+        :param exc:
+            The exception that was raised.
+        :param ignore_errors:
+            If True, exceptions in callbacks will be logged only, and the program
+            will continue running. If False, exceptions in callbacks will be raised,
+            which will stop the program. If None, the value of self.FILTER_ERRORS
+        """
         logger.exception("Exception in callback: %s", exc_info=exc)
 
-    def dispatch_callbacks(self, callbacks: list[Callback], *args) -> None:
+        if ignore_errors is None:
+            ignore_errors = self.FILTER_ERRORS
+        if not ignore_errors:
+            raise exc
+
+    def dispatch_callbacks(self, callbacks: list[Callable], *args, **kwargs) -> None:
         """Dispatch a list of callbacks with the given arguments.
 
         :param callbacks:
             List of callbacks to call
         :param args:
             Arguments to pass to the callbacks
+        :param kwargs:
+            Keyword arguments to pass to the callbacks. The "ignore_errors"
+            keyword argument can be used to override the default error handling
+            behavior for this specific call. See :meth:`canopen.Network.on_error`
+            for details.
         """
-        def task_done(future: asyncio.Future) -> None:
-            """Callback to be called when a task is done."""
-            self._futures.discard(future)
-            try:
-                if (exc := future.exception()) is not None:
-                    self.on_error(exc)
-            except (asyncio.CancelledError, asyncio.InvalidStateError) as exc:
-                # Handle cancelled tasks and unfinished tasks gracefully
-                self.on_error(exc)
-
-        # Run the callbacks
+        ignore_errors = kwargs.pop("ignore_errors", None)
         for callback in callbacks:
-            result = callback(*args)
-            if result is not None and self.loop is not None and asyncio.iscoroutine(result):
-                # This function may be called from the rx thread, so it must
-                # be thread-safe. We cannot use asyncio.create_task() here, since
-                # it is not thread-safe.
-                task = asyncio.run_coroutine_threadsafe(result, self.loop)
-                self._futures.add(task)
-                task.add_done_callback(task_done)
+            try:
+                result = callback(*args, **kwargs)
+                if result is not None and asyncio.iscoroutine(result):
+                    self.create_task(result)
+            except Exception as e:
+                self.on_error(e, ignore_errors)
 
     def check(self) -> None:
         """Check that no fatal error has occurred in the receiving thread.
@@ -343,11 +404,6 @@ def check(self) -> None:
                 logger.error("An error has caused receiving of messages to stop")
                 raise exc
 
-    @property
-    def is_async(self) -> bool:
-        """Check if canopen has been connected with async"""
-        return self.loop is not None
-
     def __getitem__(self, node_id: int) -> Union[RemoteNode, LocalNode]:
         return self.nodes[node_id]
 

canopen/nmt.py

@@ -242,12 +242,23 @@ def send_command(self, code: int) -> None:
         # The heartbeat service should start on the transition
         # between INITIALIZING and PRE-OPERATIONAL state
         if old_state == 0 and self._state == 127:
-            # FIXME: Document why this was fixed
-            if self._heartbeat_time_ms == 0:
-                heartbeat_time_ms = self._local_node.sdo[0x1017].raw
+            if self.network.is_async:
+                # In async mode we cannot read the heartbeat directly, so we
+                # create a task to read it asynchronously and start the heartbeat
+                # service when the read is complete.
+                async def start_heartbeat_async():
+                    try:
+                        heartbeat_time_ms = await self._local_node.sdo[0x1017].aread()
+                        self.start_heartbeat(heartbeat_time_ms)
+                    except KeyError:
+                        pass
+                self.network.create_task(start_heartbeat_async())
             else:
-                heartbeat_time_ms = self._heartbeat_time_ms
-            self.start_heartbeat(heartbeat_time_ms)
+                try:
+                    heartbeat_time_ms = self._local_node.sdo[0x1017].raw
+                    self.start_heartbeat(heartbeat_time_ms)
+                except KeyError:
+                    pass
         else:
             self.update_heartbeat()
 

canopen/objectdictionary/eds.py

@@ -205,6 +205,7 @@ def import_from_node(node_id: int, network: canopen.network.Network):
     network.subscribe(0x580 + node_id, sdo_client.on_response)
     # Create file like object for Store EDS variable
     try:
+        # Opening an SDO channel with file-like object is not supported in async
         with sdo_client.open(0x1021, 0, "rt") as eds_fp:
             od = import_eds(eds_fp, node_id)
     except Exception as e:

canopen/pdo/base.py

@@ -738,11 +738,6 @@ def get_data(self) -> bytes:
 
         return data
 
-    async def aget_data(self) -> bytes:
-        # Since get_data() is not making any IO, it can be called
-        # directly with no special async variant
-        return self.get_data()
-
     def set_data(self, data: bytes):
         """Set for the given variable the PDO data.
 
@@ -776,10 +771,11 @@ def set_data(self, data: bytes):
 
         self.pdo_parent.update()
 
+    async def aget_data(self) -> bytes:
+        raise RuntimeError("Read of PDO data asynchronously is not supported, use regular access")
+
     async def aset_data(self, data: bytes):
-        # Since get_data() is not making any IO, it can be called
-        # directly with no special async variant
-        return self.set_data(data)
+        raise RuntimeError("Writing PDO data asynchronously is not supported, use regular access")
 
 
 # For compatibility

canopen/sdo/base.py

@@ -171,6 +171,8 @@ async def alen(self) -> int:
         return await self[0].aread()  # type: ignore[return-value]
 
     def __contains__(self, subindex: object) -> bool:
+        if not isinstance(subindex, int):
+            return False
         return 0 <= subindex <= len(self)
 
 
@@ -181,9 +183,8 @@ def __init__(self, sdo_node: SdoBase, od: objectdictionary.ODVariable):
         self.sdo_node = sdo_node
         variable.Variable.__init__(self, od)
 
-    @ensure_not_async("Use aget_data() instead")
-    def get_data(self) -> bytes:
-        data = self.sdo_node.upload(self.od.index, self.od.subindex)
+    def _truncate_data(self, data: bytes) -> bytes:
+        """Truncate data to the size specified in the object dictionary."""
         response_size = len(data)
 
         # If size is available through variable in OD, then use the smaller of the two sizes.
@@ -196,8 +197,14 @@ def get_data(self) -> bytes:
                 data = data[:var_size]
         return data
 
+    @ensure_not_async("Use aget_data() instead")
+    def get_data(self) -> bytes:
+        data = self.sdo_node.upload(self.od.index, self.od.subindex)
+        return self._truncate_data(data)
+
     async def aget_data(self) -> bytes:
-        return await self.sdo_node.aupload(self.od.index, self.od.subindex)
+        data = await self.sdo_node.aupload(self.od.index, self.od.subindex)
+        return self._truncate_data(data)
 
     @ensure_not_async("Use aset_data() instead")
     def set_data(self, data: bytes):

canopen/sdo/server.py

@@ -250,7 +250,7 @@ async def adownload(
         data: bytes,
         force_segment: bool = False,
     ):
-        """May be called to make a write operation without an Object Dictionary. 
+        """May be called to make a write operation without an Object Dictionary.
 
         :param index:
             Index of object to write.