Refactor on async callbacks

Author: sveinse

canopen/emcy.py

@@ -2,6 +2,7 @@
 import struct
 import logging
 import threading
+import asyncio
 import time
 from typing import Callable, List, Optional, TYPE_CHECKING
 
@@ -22,13 +23,15 @@ def __init__(self):
         #: Only active EMCYs. Will be cleared on Error Reset
         self.active: List["EmcyError"] = []
         self.callbacks = []
-        self.emcy_received = threading.Condition()
+        self.emcy_received = threading.Condition()  # FIXME Async
+        self.aemcy_received = asyncio.Condition()
 
     def on_emcy(self, can_id, data, timestamp):
+        # NOTE: Callback. Will be called from another thread
         code, register, data = EMCY_STRUCT.unpack(data)
         entry = EmcyError(code, register, data, timestamp)
 
-        with self.emcy_received:
+        with self.emcy_received:  # FIXME: Blocking
             if code & 0xFF00 == 0:
                 # Error reset
                 self.active = []
@@ -37,11 +40,31 @@ def on_emcy(self, can_id, data, timestamp):
             self.log.append(entry)
             self.emcy_received.notify_all()
 
+        # NOTE: Will be called from another thread
         for callback in self.callbacks:
-            callback(entry)
+            callback(entry)  # FIXME: Assert if coroutine?
+
+    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
 
     def add_callback(self, callback: Callable[["EmcyError"], None]):
-        """Get notified on EMCY messages from this node.
+        """Get notified on EMCY messages from this node. The callback must
+           be multi-threaded.
 
         :param callback:
             Callable which must take one argument of an
@@ -66,9 +89,9 @@ def wait(
         """
         end_time = time.time() + timeout
         while True:
-            with self.emcy_received:
+            with self.emcy_received:  # FIXME: Blocking
                 prev_log_size = len(self.log)
-                self.emcy_received.wait(timeout)
+                self.emcy_received.wait(timeout)  # FIXME: Blocking
                 if len(self.log) == prev_log_size:
                     # Resumed due to timeout
                     return None

canopen/lss.py

@@ -3,6 +3,7 @@
 import logging
 import time
 import struct
+import asyncio
 try:
     import queue
 except ImportError:
@@ -90,7 +91,8 @@ def __init__(self):
         self.network: Optional[Network] = None
         self._node_id = 0
         self._data = None
-        self.responses = queue.Queue()
+        self.responses = queue.Queue()  # FIXME Async
+        self.aresponses = asyncio.Queue()
 
     def send_switch_state_global(self, mode):
         """switch mode to CONFIGURATION_STATE or WAITING_STATE
@@ -247,12 +249,12 @@ def send_identify_non_configured_remote_slave(self):
         self.__send_command(message)
 
     def fast_scan(self):
-        """This command sends a series of fastscan message 
+        """This command sends a series of fastscan message
         to find unconfigured slave with lowest number of LSS idenities
 
         :return:
             True if a slave is found.
-            False if there is no candidate. 
+            False if there is no candidate.
             list is the LSS identities [vendor_id, product_code, revision_number, seerial_number]
         :rtype: bool, list
         """
@@ -270,21 +272,21 @@ 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
-                    
+
                     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
 
                 time.sleep(0.01)
-                
+
                 # Now the next 32 bits will be scanned
                 lss_sub += 1
 
             # Now lss_id contains the entire 128 bits scanned
             return True, lss_id
-        
+
         return False, None
 
     def __send_fast_scan_message(self, id_number, bit_checker, lss_sub, lss_next):
@@ -298,7 +300,7 @@ def __send_fast_scan_message(self, id_number, bit_checker, lss_sub, lss_next):
         cs = struct.unpack_from("<B", recv_msg)[0]
         if cs == CS_IDENTIFY_SLAVE:
                 return True
-        
+
         return False
 
     def __send_lss_address(self, req_cs, number):
@@ -383,7 +385,7 @@ def __send_command(self, message):
         response = None
         if not self.responses.empty():
             logger.info("There were unexpected messages in the queue")
-            self.responses = queue.Queue()
+            self.responses = queue.Queue()  # FIXME Async
 
         self.network.send_message(self.LSS_TX_COBID, message)
 
@@ -393,15 +395,19 @@ def __send_command(self, message):
         # Wait for the slave to respond
         # TODO check if the response is LSS response message
         try:
-            response = self.responses.get(
+            response = self.responses.get(  # FIXME: Blocking
                 block=True, timeout=self.RESPONSE_TIMEOUT)
         except queue.Empty:
             raise LssError("No LSS response received")
 
         return response
 
     def on_message_received(self, can_id, data, timestamp):
-        self.responses.put(bytes(data))
+        # NOTE: Callback. Will be called from another thread
+        self.responses.put(bytes(data))  # FIXME: Blocking
+
+    async def aon_message_received(self, can_id, data, timestamp):
+        await self.aresponses.put(bytes(data))
 
 
 class LssError(Exception):

canopen/network.py

@@ -6,9 +6,11 @@
 import logging
 import threading
 from typing import Callable, Dict, Iterable, List, Optional, Union, TYPE_CHECKING
+import asyncio
 
 if TYPE_CHECKING:
     from can import BusABC, Notifier
+    from asyncio import AbstractEventLoop
 
 try:
     import can
@@ -36,14 +38,19 @@
 class Network(MutableMapping):
     """Representation of one CAN bus containing one or more nodes."""
 
-    def __init__(self, bus=None):
+    def __init__(
+        self,
+        bus: Optional[BusABC] = None,
+        loop: Optional[AbstractEventLoop] = None
+    ):
         """
         :param can.BusABC bus:
             A python-can bus instance to re-use.
         """
         #: A python-can :class:`can.BusABC` instance which is set after
         #: :meth:`canopen.Network.connect` is called
         self.bus: Optional[BusABC] = bus
+        self.loop: Optional[asyncio.AbstractEventLoop] = loop
         #: A :class:`~canopen.network.NodeScanner` for detecting nodes
         self.scanner = NodeScanner(self)
         #: List of :class:`can.Listener` objects.
@@ -52,15 +59,19 @@ def __init__(self, bus=None):
         self.notifier: Optional[Notifier] = None
         self.nodes: Dict[int, Union[RemoteNode, LocalNode]] = {}
         self.subscribers: Dict[int, List[Callback]] = {}
-        self.send_lock = threading.Lock()
+        self.send_lock = threading.Lock()  # FIXME Async
         self.sync = SyncProducer(self)
         self.time = TimeProducer(self)
         self.nmt = NmtMaster(0)
         self.nmt.network = self
 
         self.lss = LssMaster()
         self.lss.network = self
-        self.subscribe(self.lss.LSS_RX_COBID, self.lss.on_message_received)
+
+        if self.loop:
+            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)
 
     def subscribe(self, can_id: int, callback: Callback) -> None:
         """Listen for messages with a specific CAN ID.
@@ -119,8 +130,9 @@ def connect(self, *args, **kwargs) -> "Network":
         kwargs_notifier = {}
         if "loop" in kwargs:
             kwargs_notifier["loop"] = kwargs["loop"]
+            self.loop = kwargs["loop"]
             del kwargs["loop"]
-        self.bus = can.interface.Bus(*args, **kwargs)
+        self.bus = can.Bus(*args, **kwargs)
         logger.info("Connected to '%s'", self.bus.channel_info)
         self.notifier = can.Notifier(self.bus, self.listeners, 1, **kwargs_notifier)
         return self
@@ -220,7 +232,9 @@ def send_message(self, can_id: int, data: bytes, remote: bool = False) -> None:
                           arbitration_id=can_id,
                           data=data,
                           is_remote_frame=remote)
-        with self.send_lock:
+        # NOTE: This lock is ok for async, because ther is only one thread
+        #       calling this function when using async, so it'll never lock.
+        with self.send_lock:  # FIXME: Blocking
             self.bus.send(msg)
         self.check()
 
@@ -256,10 +270,13 @@ def notify(self, can_id: int, data: bytearray, timestamp: float) -> None:
         :param timestamp:
             Timestamp of the message, preferably as a Unix timestamp
         """
-        if can_id in self.subscribers:
-            callbacks = self.subscribers[can_id]
+        # NOTE: Callback. Will be called from another thread
+        callbacks = self.subscribers.get(can_id)
+        if callbacks is not None:
             for callback in callbacks:
-                callback(can_id, data, timestamp)
+                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)
         self.scanner.on_message_received(can_id)
 
     def check(self) -> None:
@@ -360,6 +377,7 @@ def __init__(self, network: Network):
         self.network = network
 
     def on_message_received(self, msg):
+        # NOTE: Callback. Will be called from another thread
         if msg.is_error_frame or msg.is_remote_frame:
             return
 
@@ -394,9 +412,11 @@ def __init__(self, network: Optional[Network] = None):
         self.nodes: List[int] = []
 
     def on_message_received(self, can_id: int):
+        # NOTE: Callback. Will be called from another thread
         service = can_id & 0x780
         node_id = can_id & 0x7F
         if node_id not in self.nodes and node_id != 0 and service in self.SERVICES:
+            # NOTE: Assume this is thread-safe
             self.nodes.append(node_id)
 
     def reset(self):