Added callback dispatcher

Author: sveinse

canopen/emcy.py

@@ -7,7 +7,6 @@
 from typing import Callable, List, Optional
 
 from canopen.async_guard import ensure_not_async
-from canopen.utils import call_callbacks
 import canopen.network
 
 
@@ -26,6 +25,7 @@ def __init__(self):
         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
@@ -44,8 +44,7 @@ def on_emcy(self, can_id, data, timestamp):
             self.emcy_received.notify_all()
 
         # Call all registered callbacks
-        # FIXME: Add the nework loop to the callback
-        call_callbacks(self.callbacks, None, entry)
+        self.network.dispatch_callbacks(self.callbacks, entry)
 
     def add_callback(self, callback: Callable[[EmcyError], None]):
         """Get notified on EMCY messages from this node.

canopen/network.py

@@ -40,6 +40,7 @@ def __init__(self, bus: Optional[can.BusABC] = None, notifier: Optional[can.Noti
         #: :meth:`canopen.Network.connect` is called
         self.bus: Optional[BusABC] = bus
         self.loop: Optional[asyncio.AbstractEventLoop] = loop
+        self._tasks: set[asyncio.Task] = set()
         #: A :class:`~canopen.network.NodeScanner` for detecting nodes
         self.scanner = NodeScanner(self)
         #: List of :class:`can.Listener` objects.
@@ -119,6 +120,12 @@ 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:
+            # Do not start a can notifier with the async loop. It changes the
+            # behavior of the notifier callbacks. Instead of running the
+            # callbacks from a separate thread, it runs the callbacks in the
+            # same thread as the event loop where blocking calls are not allowed.
+            # This library needs to support both async and sync, so we need to
+            # use the notifier in a separate thread.
             self.notifier = can.Notifier(self.bus, [], self.NOTIFIER_CYCLE)
         for listener in self.listeners:
             self.notifier.add_listener(listener)
@@ -148,6 +155,15 @@ def __enter__(self):
     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.
+        return self
+
+    async def __aexit__(self, type, value, traceback):
+        self.disconnect()
+
     # FIXME: Implement async "aadd_node"
 
     def add_node(
@@ -264,11 +280,44 @@ def notify(self, can_id: int, data: bytearray, timestamp: float) -> None:
             Timestamp of the message, preferably as a Unix timestamp
         """
         if can_id in self.subscribers:
-            callbacks = self.subscribers[can_id]
-            for callback in callbacks:
-                callback(can_id, data, timestamp)
+            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
+        logger.exception("Exception in callback: %s", exc_info=exc)
+
+    def dispatch_callbacks(self, callbacks: List[Callback], *args) -> 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
+        """
+        def task_done(task: asyncio.Task) -> None:
+            """Callback to be called when a task is done."""
+            self._tasks.discard(task)
+
+            # FIXME: This section should probably be migrated to a TaskGroup.
+            # However, this is not available yet in Python 3.8 - 3.10.
+            try:
+                if (exc := task.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
+        for callback in callbacks:
+            result = callback(*args)
+            if result is not None and asyncio.iscoroutine(result):
+                task = asyncio.create_task(result)
+                self._tasks.add(task)
+                task.add_done_callback(task_done)
+
     def check(self) -> None:
         """Check that no fatal error has occurred in the receiving thread.
 
@@ -397,7 +446,7 @@ def on_message_received(self, msg):
             self.network.notify(msg.arbitration_id, msg.data, msg.timestamp)
         except Exception as e:
             # Exceptions in any callbaks should not affect CAN processing
-            logger.error(str(e))
+            self.network.on_error(e)
 
     def stop(self) -> None:
         """Override abstract base method to release any resources."""

canopen/nmt.py

@@ -6,7 +6,6 @@
 from typing import Callable, Optional, TYPE_CHECKING
 
 from canopen.async_guard import ensure_not_async
-from canopen.utils import call_callbacks
 import canopen.network
 
 if TYPE_CHECKING:
@@ -144,7 +143,7 @@ def on_heartbeat(self, can_id, data, timestamp):
             self.state_update.notify_all()
 
         # Call all registered callbacks
-        call_callbacks(self._callbacks, self.network.loop, new_state)
+        self.network.dispatch_callbacks(self._callbacks, new_state)
 
     def send_command(self, code: int):
         """Send an NMT command code to the node.

canopen/node/remote.py

@@ -59,6 +59,7 @@ def associate_network(self, network: canopen.network.Network):
         self.tpdo.network = network
         self.rpdo.network = network
         self.nmt.network = network
+        self.emcy.network = network
         for sdo in self.sdo_channels:
             network.subscribe(sdo.tx_cobid, sdo.on_response)
         network.subscribe(0x700 + self.id, self.nmt.on_heartbeat)
@@ -79,6 +80,7 @@ def remove_network(self) -> None:
         self.tpdo.network = canopen.network._UNINITIALIZED_NETWORK
         self.rpdo.network = canopen.network._UNINITIALIZED_NETWORK
         self.nmt.network = canopen.network._UNINITIALIZED_NETWORK
+        self.emcy.network = canopen.network._UNINITIALIZED_NETWORK
 
     def add_sdo(self, rx_cobid, tx_cobid):
         """Add an additional SDO channel.

canopen/pdo/base.py

@@ -13,7 +13,6 @@
 from canopen import variable
 from canopen.async_guard import ensure_not_async
 from canopen.sdo import SdoAbortedError
-from canopen.utils import call_callbacks
 
 if TYPE_CHECKING:
     from canopen import LocalNode, RemoteNode
@@ -339,7 +338,7 @@ def on_message(self, can_id, data, timestamp):
                 self.receive_condition.notify_all()
 
             # Call all registered callbacks
-            call_callbacks(self.callbacks, self.pdo_node.network.loop, self)
+            self.pdo_node.network.dispatch_callbacks(self.callbacks, self)
 
     def add_callback(self, callback: Callable[[PdoMap], None]) -> None:
         """Add a callback which will be called on receive.

canopen/utils.py

@@ -1,7 +1,6 @@
 """Additional utility functions for canopen."""
 
-import asyncio
-from typing import Optional, Union, Iterable, Callable
+from typing import Optional, Union
 
 
 def pretty_index(index: Optional[Union[int, str]],
@@ -22,24 +21,3 @@ def pretty_index(index: Optional[Union[int, str]],
         sub_str = f"{sub!r}"
 
     return ":".join(s for s in (index_str, sub_str) if s)
-
-
-def call_callbacks(callbacks: Iterable[Callable], loop: asyncio.AbstractEventLoop | None = None, *args, **kwargs) -> bool:
-    """Call a list of callbacks with the given arguments.
-
-    """
-
-    def dispatch():
-        for callback in callbacks:
-            result = callback(*args, **kwargs)
-            if result is not None and asyncio.iscoroutine(result):
-                asyncio.create_task(result)
-
-    # If the loop is running, call the callbacks from the loop to minimize
-    # blocking and multithreading issues.
-    if loop is not None and loop.is_running():
-        loop.call_soon_threadsafe(dispatch)
-        return False
-    else:
-        dispatch()
-        return True