Annotation and fixes

* Comment annotation fixups
* Added SdoServer aupload and adownload
* Fix missing Network.is_async() uses
* Fix workaround in pdo.map.save/asave

Author: sveinse

README.rst

@@ -11,6 +11,28 @@ The library supports Python 3.6+.
 This library is the asyncio port of CANopen. See below for code example.
 
 
+Async status
+------------
+
+The remaining work for feature complete async implementation:
+
+* Implement :code:`ABlockUploadStream`, :code:`ABlockDownloadStream` and
+  :code:`ATextIOWrapper` for async in :code:`SdoClient`
+
+* Implement :code:`EcmyConsumer.wait()` for async
+
+* Implement async in :code:`LssMaster``
+
+* Async implementation of :code:`BaseNode402`
+
+* Implement async variant of :code:`Network.add_node`. This will probably also
+  add need of async variant of :code:`input_from_node` in eds.py
+
+* Update unittests for async
+
+* Update documentation and examples
+
+
 Features
 --------
 

canopen/emcy.py

@@ -33,7 +33,8 @@ def on_emcy(self, can_id, data, timestamp):
         code, register, data = EMCY_STRUCT.unpack(data)
         entry = EmcyError(code, register, data, timestamp)
 
-        with self.emcy_received:  # NOTE: Blocking call
+        # NOTE: Blocking call
+        with self.emcy_received:
             if code & 0xFF00 == 0:
                 # Error reset
                 self.active = []
@@ -43,7 +44,8 @@ def on_emcy(self, can_id, data, timestamp):
             self.emcy_received.notify_all()
 
         for callback in self.callbacks:
-            callback(entry)  # FIXME: Assert if callback is coroutine?
+            # FIXME: Assert if callback is a coroutine?
+            callback(entry)
 
     async def aon_emcy(self, can_id, data, timestamp):
         code, register, data = EMCY_STRUCT.unpack(data)
@@ -78,7 +80,8 @@ def reset(self):
         self.log = []
         self.active = []
 
-    # FIXME: Make async implementation
+    # FIXME: Implement "await" function. (Other name is needed here)
+
     @ensure_not_async  # NOTE: Safeguard for accidental async use
     def wait(
         self, emcy_code: Optional[int] = None, timeout: float = 10
@@ -92,9 +95,11 @@ def wait(
         """
         end_time = time.time() + timeout
         while True:
-            with self.emcy_received:  # NOTE: Blocking call
+            # NOTE: Blocking call
+            with self.emcy_received:
                 prev_log_size = len(self.log)
-                self.emcy_received.wait(timeout)  # NOTE: Blocking call
+                # NOTE: Blocking call
+                self.emcy_received.wait(timeout)
                 if len(self.log) == prev_log_size:
                     # Resumed due to timeout
                     return None

canopen/lss.py

@@ -249,7 +249,7 @@ def send_identify_non_configured_remote_slave(self):
         message[0] = CS_IDENTIFY_NON_CONFIGURED_REMOTE_SLAVE
         self.__send_command(message)
 
-    # FIXME: Make async implementation
+    # FIXME: Make async implementation "afast_scan"
     @ensure_not_async  # NOTE: Safeguard for accidental async use
     def fast_scan(self):
         """This command sends a series of fastscan message
@@ -267,7 +267,8 @@ 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)  # NOTE: Blocking call
+            # NOTE: Blocking call
+            time.sleep(0.01)
             while lss_sub < 4:
                 lss_bit_check = 32
                 while lss_bit_check > 0:
@@ -276,13 +277,15 @@ 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)  # NOTE: Blocking call
+                    # NOTE: Blocking call
+                    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)  # NOTE: Blocking call
+                # NOTE: Blocking call
+                time.sleep(0.01)
 
                 # Now the next 32 bits will be scanned
                 lss_sub += 1
@@ -306,7 +309,7 @@ def __send_fast_scan_message(self, id_number, bit_checker, lss_sub, lss_next):
 
         return False
 
-    # FIXME: Make async implementation
+    # FIXME: Make async implementation "__asend_lss_address"
     @ensure_not_async  # NOTE: Safeguard for accidental async use
     def __send_lss_address(self, req_cs, number):
         message = bytearray(8)
@@ -316,7 +319,8 @@ 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)  # NOTE: Blocking call
+        # NOTE: Blocking call
+        time.sleep(0.2)
 
         return response
 
@@ -371,7 +375,7 @@ def __send_configure(self, req_cs, value1=0, value2=0):
             error_msg = "LSS Error: %d" % error_code
             raise LssError(error_msg)
 
-    # FIXME: Make async implementation
+    # FIXME: Make async implementation "__asend_command"
     @ensure_not_async  # NOTE: Safeguard for accidental async use
     def __send_command(self, message):
         """Send a LSS operation code to the network
@@ -392,7 +396,8 @@ 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: Recreating the queue
+            # FIXME: Recreating the queue
+            self.responses = queue.Queue()
 
         self.network.send_message(self.LSS_TX_COBID, message)
 
@@ -402,7 +407,8 @@ 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(  # NOTE: Blocking call
+            # NOTE: Blocking call
+            response = self.responses.get(
                 block=True, timeout=self.RESPONSE_TIMEOUT)
         except queue.Empty:
             raise LssError("No LSS response received")
@@ -412,7 +418,8 @@ def __send_command(self, message):
     @ensure_not_async  # NOTE: Safeguard for accidental async use
     def on_message_received(self, can_id, data, timestamp):
         # NOTE: Callback. Called from another thread
-        self.responses.put(bytes(data))  # NOTE: Blocking call
+        # NOTE: Blocking call
+        self.responses.put(bytes(data))
 
     async def aon_message_received(self, can_id, data, timestamp):
         await self.aresponses.put(bytes(data))

canopen/network.py

@@ -69,7 +69,7 @@ def __init__(
         self.lss = LssMaster()
         self.lss.network = self
 
-        if self.loop:
+        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)
@@ -163,6 +163,8 @@ def __enter__(self):
     def __exit__(self, type, value, traceback):
         self.disconnect()
 
+    # FIXME: Implement async "aadd_node"
+
     def add_node(
         self,
         node: Union[int, RemoteNode, LocalNode],

canopen/nmt.py

@@ -132,14 +132,16 @@ def __init__(self, node_id: int):
     @ensure_not_async  # NOTE: Safeguard for accidental async use
     def on_heartbeat(self, can_id, data, timestamp):
         # NOTE: Callback. Called from another thread unless async
-        with self.state_update:  # NOTE: Blocking call
+        # NOTE: Blocking call
+        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:
-                callback(new_state)  # FIXME: Assert if callback is coroutine?
+                # FIXME: Assert if callback is coroutine?
+                callback(new_state)
             if new_state == 0:
                 # Boot-up, will go to PRE-OPERATIONAL automatically
                 self._state = 127
@@ -181,9 +183,11 @@ def send_command(self, code: int):
     @ensure_not_async  # NOTE: Safeguard for accidental async use
     def wait_for_heartbeat(self, timeout: float = 10):
         """Wait until a heartbeat message is received."""
-        with self.state_update:  # NOTE: Blocking call
+        # NOTE: Blocking call
+        with self.state_update:
             self._state_received = None
-            self.state_update.wait(timeout)  # NOTE: Blocking call
+            # NOTE: Blocking call
+            self.state_update.wait(timeout)
         if self._state_received is None:
             raise NmtError("No boot-up or heartbeat received")
         return self.state
@@ -204,25 +208,27 @@ def wait_for_bootup(self, timeout: float = 10) -> None:
         end_time = time.time() + timeout
         while True:
             now = time.time()
-            with self.state_update:  # NOTE: Blocking call
+            # NOTE: Blocking call
+            with self.state_update:
                 self._state_received = None
-                self.state_update.wait(end_time - now + 0.1)  # NOTE: Blocking call
+                # NOTE: Blocking call
+                self.state_update.wait(end_time - now + 0.1)
             if now > end_time:
                 raise NmtError("Timeout waiting for boot-up message")
             if self._state_received == 0:
                 break
 
     async def await_for_bootup(self, timeout: float = 10) -> None:
         """Wait until a boot-up message is received."""
-        async def wait_for_bootup():
+        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)
+            await asyncio.wait_for(_wait_for_bootup(), timeout=timeout)
         except asyncio.TimeoutError:
             raise NmtError("Timeout waiting for boot-up message")
 
@@ -282,7 +288,8 @@ 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:
-            heartbeat_time_ms = self._local_node.sdo[0x1017].get_raw()  # FIXME: Blocking?
+            # NOTE: Blocking - OK. Protected in SdoClient
+            heartbeat_time_ms = self._local_node.sdo[0x1017].get_raw()
             self.start_heartbeat(heartbeat_time_ms)
         else:
             self.update_heartbeat()

canopen/node/base.py

@@ -30,3 +30,5 @@ def __init__(
         self.object_dictionary = object_dictionary
 
         self.id = node_id or self.object_dictionary.node_id
+
+    # FIXME: Should associate_network() and remove_network() be a part of the base API?

canopen/node/remote.py

@@ -109,7 +109,7 @@ def add_sdo(self, rx_cobid, tx_cobid):
         client = SdoClient(rx_cobid, tx_cobid, self.object_dictionary)
         self.sdo_channels.append(client)
         if self.network is not None:
-            if self.network.loop:
+            if self.network.is_async():
                 self.network.subscribe(client.tx_cobid, client.aon_response)
             else:
                 self.network.subscribe(client.tx_cobid, client.on_response)
@@ -151,9 +151,11 @@ def __load_configuration_helper(self, index, subindex, name, value):
                     subindex=subindex,
                     name=name,
                     value=value)))
-                self.sdo[index][subindex].set_raw(value)  # FIXME: Blocking?
+                # NOTE: Blocking - OK. Protected in SdoClient
+                self.sdo[index][subindex].set_raw(value)
             else:
-                self.sdo[index].set_raw(value)  # FIXME: Blocking?
+                # FIXME: Blocking - OK. Protected in SdoClient
+                self.sdo[index].set_raw(value)
                 logger.info(str('SDO [{index:#06x}]: {name}: {value:#06x}'.format(
                     index=index,
                     name=name,

canopen/objectdictionary/eds.py

@@ -170,7 +170,7 @@ def import_eds(source, node_id):
     return od
 
 
-# FIXME: Make async variant
+# FIXME: Make async variant "aimport_from_node"
 @ensure_not_async  # NOTE: Safeguard for accidental async use
 def import_from_node(node_id, network):
     """ Download the configuration from the remote node