Implement the framwork for sync back-end

* All callbacks are synchronous and same in both sync and async mode
* Sync waiting is done with `asyncio.to_thread()` from async

Author: sveinse

README.rst

@@ -11,6 +11,17 @@ 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
 ------------
 

canopen/emcy.py

@@ -7,6 +7,7 @@
 from typing import Callable, List, Optional
 
 from canopen.async_guard import ensure_not_async
+from canopen.utils import call_callbacks
 import canopen.network
 
 
@@ -25,7 +26,6 @@ def __init__(self):
         self.active: List[EmcyError] = []
         self.callbacks = []
         self.emcy_received = threading.Condition()
-        self.aemcy_received = asyncio.Condition()
 
     # @callback  # NOTE: called from another thread
     @ensure_not_async  # NOTE: Safeguard for accidental async use
@@ -43,28 +43,9 @@ def on_emcy(self, can_id, data, timestamp):
             self.log.append(entry)
             self.emcy_received.notify_all()
 
-        for callback in self.callbacks:
-            # FIXME: Assert if callback is a coroutine?
-            callback(entry)
-
-    # @callback
-    async def aon_emcy(self, can_id, data, timestamp):
-        code, register, data = EMCY_STRUCT.unpack(data)
-        entry = EmcyError(code, register, data, timestamp)
-
-        async with self.aemcy_received:
-            if code & 0xFF00 == 0:
-                # Error reset
-                self.active = []
-            else:
-                self.active.append(entry)
-            self.log.append(entry)
-            self.aemcy_received.notify_all()
-
-        for callback in self.callbacks:
-            res = callback(entry)
-            if res is not None and asyncio.iscoroutine(res):
-                await res
+        # Call all registered callbacks
+        # FIXME: Add the nework loop to the callback
+        call_callbacks(self.callbacks, None, entry)
 
     def add_callback(self, callback: Callable[[EmcyError], None]):
         """Get notified on EMCY messages from this node.
@@ -111,13 +92,17 @@ def wait(
                     # This is the one we're interested in
                     return emcy
 
-    def async_wait(
+    async def async_wait(
         self, emcy_code: Optional[int] = None, timeout: float = 10
     ) -> EmcyError:
-        # FIXME: Implement this function
-        raise NotImplementedError(
-            "async_wait is not implemented."
-        )
+        """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

@@ -89,7 +89,6 @@ def __init__(self) -> None:
         self._node_id = 0
         self._data = None
         self.responses = queue.Queue()
-        self.aresponses = asyncio.Queue()
 
     def send_switch_state_global(self, mode):
         """switch mode to CONFIGURATION_STATE or WAITING_STATE
@@ -416,10 +415,6 @@ def on_message_received(self, can_id, data, timestamp):
         # NOTE: Blocking call
         self.responses.put(bytes(data))
 
-    # @callback
-    async def aon_message_received(self, can_id, data, timestamp):
-        await self.aresponses.put(bytes(data))
-
 
 class LssError(Exception):
     """Some LSS operation failed."""

canopen/network.py

@@ -62,10 +62,7 @@ def __init__(self, bus: Optional[can.BusABC] = None, notifier: Optional[can.Noti
         # work. See async_guard.py
         set_async_sentinel(self.is_async())
 
-        if self.is_async():
-            self.subscribe(self.lss.LSS_RX_COBID, self.lss.aon_message_received)
-        else:
-            self.subscribe(self.lss.LSS_RX_COBID, self.lss.on_message_received)
+        self.subscribe(self.lss.LSS_RX_COBID, self.lss.on_message_received)
 
     def subscribe(self, can_id: int, callback: Callback) -> None:
         """Listen for messages with a specific CAN ID.
@@ -122,7 +119,7 @@ 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, loop=self.loop)
+            self.notifier = can.Notifier(self.bus, [], self.NOTIFIER_CYCLE)
         for listener in self.listeners:
             self.notifier.add_listener(listener)
         return self
@@ -266,12 +263,10 @@ def notify(self, can_id: int, data: bytearray, timestamp: float) -> None:
         :param timestamp:
             Timestamp of the message, preferably as a Unix timestamp
         """
-        callbacks = self.subscribers.get(can_id)
-        if callbacks is not None:
+        if can_id in self.subscribers:
+            callbacks = self.subscribers[can_id]
             for callback in callbacks:
-                res = callback(can_id, data, timestamp)
-                if res is not None and self.loop is not None and asyncio.iscoroutine(res):
-                    self.loop.create_task(res)
+                callback(can_id, data, timestamp)
         self.scanner.on_message_received(can_id)
 
     def check(self) -> None:

canopen/nmt.py

@@ -6,6 +6,7 @@
 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:
@@ -121,22 +122,19 @@ def __init__(self, node_id: int):
         #: Timestamp of last heartbeat message
         self.timestamp: Optional[float] = None
         self.state_update = threading.Condition()
-        self.astate_update = asyncio.Condition()
         self._callbacks = []
 
     # @callback  # NOTE: called from another thread
     @ensure_not_async  # NOTE: Safeguard for accidental async use
     def on_heartbeat(self, can_id, data, timestamp):
+        new_state, = struct.unpack_from("B", data)
+        # Mask out toggle bit
+        new_state &= 0x7F
+        logger.debug("Received heartbeat can-id %d, state is %d", can_id, new_state)
+
         # NOTE: Blocking lock
         with self.state_update:
             self.timestamp = timestamp
-            new_state, = struct.unpack_from("B", data)
-            # Mask out toggle bit
-            new_state &= 0x7F
-            logger.debug("Received heartbeat can-id %d, state is %d", can_id, new_state)
-            for callback in self._callbacks:
-                # FIXME: Assert if callback is coroutine?
-                callback(new_state)
             if new_state == 0:
                 # Boot-up, will go to PRE-OPERATIONAL automatically
                 self._state = 127
@@ -145,25 +143,8 @@ def on_heartbeat(self, can_id, data, timestamp):
             self._state_received = new_state
             self.state_update.notify_all()
 
-    # @callback
-    async def aon_heartbeat(self, can_id, data, timestamp):
-        async with self.astate_update:
-            self.timestamp = timestamp
-            new_state, = struct.unpack_from("B", data)
-            # Mask out toggle bit
-            new_state &= 0x7F
-            logger.debug("Received heartbeat can-id %d, state is %d", can_id, new_state)
-            for callback in self._callbacks:
-                res = callback(new_state)
-                if res is not None and asyncio.iscoroutine(res):
-                    await res
-            if new_state == 0:
-                # Boot-up, will go to PRE-OPERATIONAL automatically
-                self._state = 127
-            else:
-                self._state = new_state
-            self._state_received = new_state
-            self.astate_update.notify_all()
+        # Call all registered callbacks
+        call_callbacks(self._callbacks, self.network.loop, new_state)
 
     def send_command(self, code: int):
         """Send an NMT command code to the node.
@@ -190,13 +171,7 @@ def wait_for_heartbeat(self, timeout: float = 10):
 
     async def await_for_heartbeat(self, timeout: float = 10):
         """Wait until a heartbeat message is received."""
-        async with self.astate_update:
-            self._state_received = None
-            try:
-                await asyncio.wait_for(self.astate_update.wait(), timeout=timeout)
-            except asyncio.TimeoutError:
-                raise NmtError("No boot-up or heartbeat received")
-        return self.state
+        return await asyncio.to_thread(self.wait_for_heartbeat, timeout)
 
     @ensure_not_async  # NOTE: Safeguard for accidental async use
     def wait_for_bootup(self, timeout: float = 10) -> None:
@@ -216,17 +191,7 @@ def wait_for_bootup(self, timeout: float = 10) -> None:
 
     async def await_for_bootup(self, timeout: float = 10) -> None:
         """Wait until a boot-up message is received."""
-        async def _wait_for_bootup():
-            while True:
-                async with self.astate_update:
-                    self._state_received = None
-                    await self.astate_update.wait()
-                    if self._state_received == 0:
-                        return
-        try:
-            await asyncio.wait_for(_wait_for_bootup(), timeout=timeout)
-        except asyncio.TimeoutError:
-            raise NmtError("Timeout waiting for boot-up message")
+        return await asyncio.to_thread(self.wait_for_bootup, timeout)
 
     def add_heartbeat_callback(self, callback: Callable[[int], None]):
         """Add function to be called on heartbeat reception.

canopen/node/remote.py

@@ -61,25 +61,17 @@ def associate_network(self, network: canopen.network.Network):
         self.nmt.network = network
         for sdo in self.sdo_channels:
             network.subscribe(sdo.tx_cobid, sdo.on_response)
-        if network.is_async():
-            network.subscribe(0x700 + self.id, self.nmt.aon_heartbeat)
-            network.subscribe(0x80 + self.id, self.emcy.aon_emcy)
-        else:
-            network.subscribe(0x700 + self.id, self.nmt.on_heartbeat)
-            network.subscribe(0x80 + self.id, self.emcy.on_emcy)
+        network.subscribe(0x700 + self.id, self.nmt.on_heartbeat)
+        network.subscribe(0x80 + self.id, self.emcy.on_emcy)
         network.subscribe(0, self.nmt.on_command)
 
     def remove_network(self) -> None:
         if not self.has_network():
             return
         for sdo in self.sdo_channels:
             self.network.unsubscribe(sdo.tx_cobid, sdo.on_response)
-        if self.network.is_async():
-            self.network.unsubscribe(0x700 + self.id, self.nmt.aon_heartbeat)
-            self.network.unsubscribe(0x80 + self.id, self.emcy.aon_emcy)
-        else:
-            self.network.unsubscribe(0x700 + self.id, self.nmt.on_heartbeat)
-            self.network.unsubscribe(0x80 + self.id, self.emcy.on_emcy)
+        self.network.unsubscribe(0x700 + self.id, self.nmt.on_heartbeat)
+        self.network.unsubscribe(0x80 + self.id, self.emcy.on_emcy)
         self.network.unsubscribe(0, self.nmt.on_command)
         self.network = canopen.network._UNINITIALIZED_NETWORK
         self.sdo.network = canopen.network._UNINITIALIZED_NETWORK

canopen/pdo/base.py

@@ -13,6 +13,7 @@
 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
@@ -221,7 +222,6 @@ def __init__(self, pdo_node, com_record, map_array):
         self.period: Optional[float] = None
         self.callbacks = []
         self.receive_condition = threading.Condition()
-        self.areceive_condition = asyncio.Condition()
         self.is_received: bool = False
         self._task = None
 
@@ -337,25 +337,9 @@ def on_message(self, can_id, data, timestamp):
                     self.period = timestamp - self.timestamp
                 self.timestamp = timestamp
                 self.receive_condition.notify_all()
-                for callback in self.callbacks:
-                    # FIXME: Assert on couroutines?
-                    callback(self)
 
-    # @callback
-    async def aon_message(self, can_id, data, timestamp):
-        is_transmitting = self._task is not None
-        if can_id == self.cob_id and not is_transmitting:
-            async with self.areceive_condition:
-                self.is_received = True
-                self.data = data
-                if self.timestamp is not None:
-                    self.period = timestamp - self.timestamp
-                self.timestamp = timestamp
-                self.areceive_condition.notify_all()
-                for callback in self.callbacks:
-                    res = callback(self)
-                    if res is not None and asyncio.iscoroutine(res):
-                        await res
+            # Call all registered callbacks
+            call_callbacks(self.callbacks, self.pdo_node.network.loop, self)
 
     def add_callback(self, callback: Callable[[PdoMap], None]) -> None:
         """Add a callback which will be called on receive.
@@ -570,10 +554,7 @@ def subscribe(self) -> None:
         """
         if self.enabled:
             logger.info("Subscribing to enabled PDO 0x%X on the network", self.cob_id)
-            if self.pdo_node.network.is_async():
-                self.pdo_node.network.subscribe(self.cob_id, self.aon_message)
-            else:
-                self.pdo_node.network.subscribe(self.cob_id, self.on_message)
+            self.pdo_node.network.subscribe(self.cob_id, self.on_message)
 
     def clear(self) -> None:
         """Clear all variables from this map."""
@@ -681,14 +662,7 @@ async def await_for_reception(self, timeout: float = 10) -> float:
         :param float timeout: Max time to wait in seconds.
         :return: Timestamp of message received or None if timeout.
         """
-        async with self.areceive_condition:
-            self.is_received = False
-            try:
-                await asyncio.wait_for(self.areceive_condition.wait(), timeout=timeout)
-                # FIXME: Can we assume that self.is_received it set here?
-                return self.timestamp
-            except asyncio.TimeoutError:
-                return None
+        await asyncio.to_thread(self.wait_for_reception, timeout)
 
 
 class PdoVariable(variable.Variable):

canopen/sdo/client.py

@@ -48,7 +48,7 @@ def __init__(self, rx_cobid, tx_cobid, od):
 
     # @callback  # NOTE: called from another thread
     def on_response(self, can_id, data, timestamp):
-        self.responses.put_nowait(bytes(data))
+        self.responses.put(bytes(data))
 
     @ensure_not_async  # NOTE: Safeguard for accidental async use
     def send_request(self, request):

canopen/utils.py

@@ -1,6 +1,7 @@
 """Additional utility functions for canopen."""
 
-from typing import Optional, Union
+import asyncio
+from typing import Optional, Union, Iterable, Callable
 
 
 def pretty_index(index: Optional[Union[int, str]],
@@ -21,3 +22,24 @@ 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