Added more support for async

Author: sveinse

canopen/emcy.py

@@ -23,15 +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()  # FIXME Async
+        self.emcy_received = threading.Condition()
         self.aemcy_received = asyncio.Condition()
 
     def on_emcy(self, can_id, data, timestamp):
-        # NOTE: Callback. Will be called from another thread
+        # NOTE: Callback. Called from another thread unless async
         code, register, data = EMCY_STRUCT.unpack(data)
         entry = EmcyError(code, register, data, timestamp)
 
-        with self.emcy_received:  # FIXME: Blocking
+        with self.emcy_received:  # NOTE: Blocking call
             if code & 0xFF00 == 0:
                 # Error reset
                 self.active = []
@@ -40,9 +40,8 @@ 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)  # FIXME: Assert if coroutine?
+            callback(entry)  # FIXME: Assert if callback is coroutine?
 
     async def aon_emcy(self, can_id, data, timestamp):
         code, register, data = EMCY_STRUCT.unpack(data)
@@ -89,9 +88,9 @@ def wait(
         """
         end_time = time.time() + timeout
         while True:
-            with self.emcy_received:  # FIXME: Blocking
+            with self.emcy_received:  # NOTE: Blocking call
                 prev_log_size = len(self.log)
-                self.emcy_received.wait(timeout)  # FIXME: Blocking
+                self.emcy_received.wait(timeout)  # NOTE: Blocking call
                 if len(self.log) == prev_log_size:
                     # Resumed due to timeout
                     return None

canopen/lss.py

@@ -91,7 +91,7 @@ def __init__(self):
         self.network: Optional[Network] = None
         self._node_id = 0
         self._data = None
-        self.responses = queue.Queue()  # FIXME Async
+        self.responses = queue.Queue()
         self.aresponses = asyncio.Queue()
 
     def send_switch_state_global(self, mode):
@@ -264,7 +264,7 @@ def fast_scan(self):
         lss_next = 0
 
         if self.__send_fast_scan_message(lss_id[0], lss_bit_check, lss_sub, lss_next):
-            time.sleep(0.01)
+            time.sleep(0.01)  # NOTE: Blocking call
             while lss_sub < 4:
                 lss_bit_check = 32
                 while lss_bit_check > 0:
@@ -273,13 +273,13 @@ 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)
+                    time.sleep(0.01)  # NOTE: Blocking call
 
                 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)
+                time.sleep(0.01)  # NOTE: Blocking
 
                 # Now the next 32 bits will be scanned
                 lss_sub += 1
@@ -311,7 +311,7 @@ def __send_lss_address(self, req_cs, number):
         response = self.__send_command(message)
         # some device needs these delays between messages
         # because it can't handle messages arriving with no delay
-        time.sleep(0.2)
+        time.sleep(0.2)  # NOTE: Blocking call
 
         return response
 
@@ -385,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()  # FIXME Async
+            self.responses = queue.Queue()  # FIXME: Recreating the queue. Async too?
 
         self.network.send_message(self.LSS_TX_COBID, message)
 
@@ -395,16 +395,16 @@ 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(  # FIXME: Blocking
+            response = self.responses.get(  # NOTE: Blocking call
                 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):
-        # NOTE: Callback. Will be called from another thread
-        self.responses.put(bytes(data))  # FIXME: Blocking
+        # NOTE: Callback. Called from another thread
+        self.responses.put(bytes(data))  # NOTE: Blocking call
 
     async def aon_message_received(self, can_id, data, timestamp):
         await self.aresponses.put(bytes(data))

canopen/network.py

@@ -59,7 +59,7 @@ def __init__(
         self.notifier: Optional[Notifier] = None
         self.nodes: Dict[int, Union[RemoteNode, LocalNode]] = {}
         self.subscribers: Dict[int, List[Callback]] = {}
-        self.send_lock = threading.Lock()  # FIXME Async
+        self.send_lock = threading.Lock()
         self.sync = SyncProducer(self)
         self.time = TimeProducer(self)
         self.nmt = NmtMaster(0)
@@ -232,9 +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)
-        # 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
+        # NOTE: Blocking lock. This is probably ok for async, because async
+        #       only use one thread.
+        with self.send_lock:
             self.bus.send(msg)
         self.check()
 
@@ -270,7 +270,7 @@ def notify(self, can_id: int, data: bytearray, timestamp: float) -> None:
         :param timestamp:
             Timestamp of the message, preferably as a Unix timestamp
         """
-        # NOTE: Callback. Will be called from another thread
+        # NOTE: Callback. Called from another thread unless async
         callbacks = self.subscribers.get(can_id)
         if callbacks is not None:
             for callback in callbacks:
@@ -291,6 +291,10 @@ def check(self) -> None:
                 logger.error("An error has caused receiving of messages to stop")
                 raise exc
 
+    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]
 
@@ -355,6 +359,8 @@ def update(self, data: bytes) -> None:
         :param data:
             New data to transmit
         """
+        # NOTE: Called from callback, which is another thread on non-async use.
+        #       Make sure this is thread-safe.
         new_data = bytearray(data)
         old_data = self.msg.data
         self.msg.data = new_data
@@ -377,7 +383,7 @@ def __init__(self, network: Network):
         self.network = network
 
     def on_message_received(self, msg):
-        # NOTE: Callback. Will be called from another thread
+        # NOTE: Callback. Called from another thread unless async
         if msg.is_error_frame or msg.is_remote_frame:
             return
 
@@ -412,11 +418,14 @@ 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
+        # NOTE: Callback. Called from another thread unless async
         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
+            # NOTE: In the current CPython implementation append on lists are
+            #       atomic which makes this thread-safe. However, other py
+            #       interpreters might not. It should be considered if a better
+            #       mechanism is needed to protect against race.
             self.nodes.append(node_id)
 
     def reset(self):

canopen/nmt.py

@@ -57,7 +57,7 @@ def __init__(self, node_id: int):
         self._state = 0
 
     def on_command(self, can_id, data, timestamp):
-        # NOTE: Callback. Will be called from another thread
+        # NOTE: Callback. Called from another thread unless async
         cmd, node_id = struct.unpack_from("BB", data)
         if node_id in (self.id, 0):
             logger.info("Node %d received command %d", self.id, cmd)
@@ -66,7 +66,7 @@ def on_command(self, can_id, data, timestamp):
                 if new_state != self._state:
                     logger.info("New NMT state %s, old state %s",
                                 NMT_STATES[new_state], NMT_STATES[self._state])
-                # NOTE: Assume thread-safe
+                # FIXME: Is this thread-safe?
                 self._state = new_state
 
     def send_command(self, code: int):
@@ -125,20 +125,20 @@ def __init__(self, node_id: int):
         self._node_guarding_producer = None
         #: Timestamp of last heartbeat message
         self.timestamp: Optional[float] = None
-        self.state_update = threading.Condition()  # FIXME
+        self.state_update = threading.Condition()
         self.astate_update = asyncio.Condition()
         self._callbacks = []
 
     def on_heartbeat(self, can_id, data, timestamp):
-        # NOTE: Callback. Will be called from another thread
-        with self.state_update:  # FIXME: Blocking
+        # NOTE: Callback. Called from another thread unless async
+        with self.state_update:  # NOTE: Blocking call
             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:
-                callback(new_state)  # FIXME: Assert on coroutines?
+                callback(new_state)  # FIXME: Assert if callback is coroutine?
             if new_state == 0:
                 # Boot-up, will go to PRE-OPERATIONAL automatically
                 self._state = 127
@@ -179,9 +179,9 @@ def send_command(self, code: int):
 
     def wait_for_heartbeat(self, timeout: float = 10):
         """Wait until a heartbeat message is received."""
-        with self.state_update:  # FIXME: Blocking
+        with self.state_update:  # NOTE: Blocking call
             self._state_received = None
-            self.state_update.wait(timeout)  # FIXME: Blocking
+            self.state_update.wait(timeout)  # NOTE: Blocking call
         if self._state_received is None:
             raise NmtError("No boot-up or heartbeat received")
         return self.state
@@ -191,9 +191,9 @@ def wait_for_bootup(self, timeout: float = 10) -> None:
         end_time = time.time() + timeout
         while True:
             now = time.time()
-            with self.state_update:  # FIXME: Blocking
+            with self.state_update:  # NOTE: Blocking call
                 self._state_received = None
-                self.state_update.wait(end_time - now + 0.1)  # FIXME: Blocking
+                self.state_update.wait(end_time - now + 0.1)  # NOTE: Blocking call
             if now > end_time:
                 raise NmtError("Timeout waiting for boot-up message")
             if self._state_received == 0:
@@ -235,7 +235,7 @@ def __init__(self, node_id: int, local_node):
         self._local_node = local_node
 
     def on_command(self, can_id, data, timestamp):
-        # NOTE: Callback. Will be called from another thread
+        # NOTE: Callback. Called from another thread unless async
         super(NmtSlave, self).on_command(can_id, data, timestamp)
         self.update_heartbeat()
 
@@ -291,9 +291,9 @@ def stop_heartbeat(self):
             self._send_task = None
 
     def update_heartbeat(self):
-        # NOTE: Called from callback. Might be called from another thread
+        # NOTE: Called from callback. Called from another thread unless async
         if self._send_task is not None:
-            # FIXME: Check if network.PeriodicMessageTask() is thread-safe
+            # FIXME: Make this thread-safe
             self._send_task.update([self._state])
 
 

canopen/node/local.py

@@ -44,12 +44,21 @@ def associate_network(self, network: Network):
         self.rpdo.network = network
         self.nmt.network = network
         self.emcy.network = network
-        network.subscribe(self.sdo.rx_cobid, self.sdo.on_request)  # FIXME: Async CB
-        network.subscribe(0, self.nmt.on_command)  # FIXME: Async CB
+        if network.is_async():
+            network.subscribe(self.sdo.rx_cobid, self.sdo.aon_request)
+            network.subscribe(0, self.nmt.aon_command)
+        else:
+            network.subscribe(self.sdo.rx_cobid, self.sdo.on_request)
+            network.subscribe(0, self.nmt.on_command)
 
     def remove_network(self):
-        self.network.unsubscribe(self.sdo.rx_cobid, self.sdo.on_request)  # FIXME: Async CB
-        self.network.unsubscribe(0, self.nmt.on_command)  # FIXME: Async CB
+        network = self.network
+        if network.is_async():
+            network.unsubscribe(self.sdo.rx_cobid, self.sdo.aon_request)
+            network.unsubscribe(0, self.nmt.aon_command)
+        else:
+            network.unsubscribe(self.sdo.rx_cobid, self.sdo.on_request)
+            network.unsubscribe(0, self.nmt.on_command)
         self.network = None
         self.sdo.network = None
         self.tpdo.network = None

canopen/node/remote.py

@@ -61,7 +61,7 @@ def associate_network(self, network: Network):
         self.tpdo.network = network
         self.rpdo.network = network
         self.nmt.network = network
-        if network.loop:
+        if network.is_async():
             for sdo in self.sdo_channels:
                 network.subscribe(sdo.tx_cobid, sdo.aon_response)
             network.subscribe(0x700 + self.id, self.nmt.aon_heartbeat)
@@ -74,12 +74,18 @@ def associate_network(self, network: Network):
         network.subscribe(0, self.nmt.on_command)
 
     def remove_network(self):
-        # FIXME: Usubscribe async CB
-        for sdo in self.sdo_channels:
-            self.network.unsubscribe(sdo.tx_cobid, sdo.on_response)
-        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)
+        network = self.network
+        if network.is_async():
+            for sdo in self.sdo_channels:
+                network.unsubscribe(sdo.tx_cobid, sdo.aon_response)
+            network.unsubscribe(0x700 + self.id, self.nmt.aon_heartbeat)
+            network.unsubscribe(0x80 + self.id, self.emcy.aon_emcy)
+        else:
+            for sdo in self.sdo_channels:
+                network.unsubscribe(sdo.tx_cobid, sdo.on_response)
+            network.unsubscribe(0x700 + self.id, self.nmt.on_heartbeat)
+            network.unsubscribe(0x80 + self.id, self.emcy.on_emcy)
+        network.unsubscribe(0, self.nmt.on_command)
         self.network = None
         self.sdo.network = None
         self.pdo.network = None

canopen/objectdictionary/eds.py

@@ -172,11 +172,12 @@ def import_from_node(node_id, network):
     :param int node_id: Identifier of the node
     :param network: network object
     """
+    # FIXME: Implement async variant
     # Create temporary SDO client
     sdo_client = SdoClient(0x600 + node_id, 0x580 + node_id, objectdictionary.ObjectDictionary())
     sdo_client.network = network
     # Subscribe to SDO responses
-    network.subscribe(0x580 + node_id, sdo_client.on_response)  # FIXME: Async CB
+    network.subscribe(0x580 + node_id, sdo_client.on_response)
     # Create file like object for Store EDS variable
     try:
         eds_fp = sdo_client.open(0x1021, 0, "rt")

canopen/pdo/base.py

@@ -308,7 +308,7 @@ def is_periodic(self) -> bool:
         return False
 
     def on_message(self, can_id, data, timestamp):
-        # NOTE: Callback. Will be called from another thread
+        # NOTE: Callback. Called from another thread unless async
         is_transmitting = self._task is not None
         if can_id == self.cob_id and not is_transmitting:
             with self.receive_condition:  # FIXME: Blocking

canopen/profiles/p402.py

@@ -462,10 +462,10 @@ def on_TPDOs_update_callback(self, mapobject):
         :param mapobject: The received PDO message.
         :type mapobject: canopen.pdo.Map
         """
-        # NOTE: Callback. Will be called from another thread
+        # NOTE: Callback. Called from another thread unless async
         for obj in mapobject:
-            # NOTE: Assume thread-safe set without locking
-            self.tpdo_values[obj.index] = obj.get_raw()
+            # FIXME: Is this thread-safe?
+            self.tpdo_values[obj.index] = obj.get_raw()  # FIXME: Blocking?
 
     @property
     def statusword(self):