First working concept

Author: sveinse

canopen/nmt.py

@@ -93,13 +93,18 @@ def state(self) -> str:
         - 'RESET'
         - 'RESET COMMUNICATION'
         """
+        logger.warning("Accessing NmtBase.state attribute is deprecated")
         if self._state in NMT_STATES:
             return NMT_STATES[self._state]
         else:
             return self._state
 
     @state.setter
     def state(self, new_state: str):
+        logger.warning("Accessing NmtBase.state setter is deprecated")
+        self.set_state(new_state)
+
+    def set_state(self, new_state: str):
         if new_state in NMT_COMMANDS:
             code = NMT_COMMANDS[new_state]
         else:

canopen/pdo/base.py

@@ -8,6 +8,7 @@
     from collections import Mapping
 import logging
 import binascii
+import asyncio
 
 if TYPE_CHECKING:
     from ..network import Network
@@ -59,6 +60,11 @@ def read(self):
         for pdo_map in self.map.values():
             pdo_map.read()
 
+    async def aread(self):
+        """Read PDO configuration from node using SDO."""
+        for pdo_map in self.map.values():
+            await pdo_map.aread()
+
     def save(self):
         """Save PDO configuration to node using SDO."""
         for pdo_map in self.map.values():
@@ -195,7 +201,8 @@ def __init__(self, pdo_node: PdoBase, com_record, map_array):
         #: Set explicitly or using the :meth:`start()` method.
         self.period: Optional[float] = None
         self.callbacks = []
-        self.receive_condition = threading.Condition()
+        #self.receive_condition = threading.Condition()  # FIXME
+        self.receive_condition = asyncio.Condition()
         self.is_received: bool = False
         self._task = None
 
@@ -296,6 +303,22 @@ def is_periodic(self) -> bool:
         # Unknown transmission type, assume non-periodic
         return False
 
+    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.receive_condition:
+                self.is_received = True
+                self.data = data
+                if self.timestamp is not None:
+                    self.period = timestamp - self.timestamp
+                self.timestamp = timestamp
+                self.receive_condition.notify_all()
+                for callback in self.callbacks:
+                    callback(self)
+
+    def on_message_async(self, can_id, data, timestamp):
+        asyncio.create_task(self.aon_message(can_id, data, timestamp))
+
     def on_message(self, can_id, data, timestamp):
         is_transmitting = self._task is not None
         if can_id == self.cob_id and not is_transmitting:
@@ -367,6 +390,55 @@ def read(self) -> None:
 
         self.subscribe()
 
+    async def aread(self) -> None:
+        """Read PDO configuration for this map using SDO."""
+        cob_id = await self.com_record[1].aget_raw()
+        self.cob_id = cob_id & 0x1FFFFFFF
+        logger.info("COB-ID is 0x%X", self.cob_id)
+        self.enabled = cob_id & PDO_NOT_VALID == 0
+        logger.info("PDO is %s", "enabled" if self.enabled else "disabled")
+        self.rtr_allowed = cob_id & RTR_NOT_ALLOWED == 0
+        logger.info("RTR is %s", "allowed" if self.rtr_allowed else "not allowed")
+        self.trans_type = await self.com_record[2].aget_raw()
+        logger.info("Transmission type is %d", self.trans_type)
+        if self.trans_type >= 254:
+            try:
+                self.inhibit_time = await self.com_record[3].aget_raw()
+            except (KeyError, SdoAbortedError) as e:
+                logger.info("Could not read inhibit time (%s)", e)
+            else:
+                logger.info("Inhibit time is set to %d ms", self.inhibit_time)
+
+            try:
+                self.event_timer = await self.com_record[5].aget_raw()
+            except (KeyError, SdoAbortedError) as e:
+                logger.info("Could not read event timer (%s)", e)
+            else:
+                logger.info("Event timer is set to %d ms", self.event_timer)
+
+            try:
+                self.sync_start_value = await self.com_record[6].aget_raw()
+            except (KeyError, SdoAbortedError) as e:
+                logger.info("Could not read SYNC start value (%s)", e)
+            else:
+                logger.info("SYNC start value is set to %d ms", self.sync_start_value)
+
+        self.clear()
+        nof_entries = await self.map_array[0].aget_raw()
+        for subindex in range(1, nof_entries + 1):
+            value = await self.map_array[subindex].aget_raw()
+            index = value >> 16
+            subindex = (value >> 8) & 0xFF
+            size = value & 0xFF
+            if hasattr(self.pdo_node.node, "curtis_hack") and self.pdo_node.node.curtis_hack:  # Curtis HACK: mixed up field order
+                index = value & 0xFFFF
+                subindex = (value >> 16) & 0xFF
+                size = (value >> 24) & 0xFF
+            if index and size:
+                self.add_variable(index, subindex, size)
+
+        self.subscribe()
+
     def save(self) -> None:
         """Save PDO configuration for this map using SDO."""
         logger.info("Setting COB-ID 0x%X and temporarily disabling PDO",
@@ -433,7 +505,8 @@ def subscribe(self) -> None:
         """
         if self.enabled:
             logger.info("Subscribing to enabled PDO 0x%X on the network", self.cob_id)
-            self.pdo_node.network.subscribe(self.cob_id, self.on_message)
+            #self.pdo_node.network.subscribe(self.cob_id, self.on_message)  # FIXME
+            self.pdo_node.network.subscribe(self.cob_id, self.on_message_async)
 
     def clear(self) -> None:
         """Clear all variables from this map."""
@@ -532,6 +605,17 @@ def wait_for_reception(self, timeout: float = 10) -> float:
             self.receive_condition.wait(timeout)
         return self.timestamp if self.is_received else None
 
+    async def await_for_reception(self, timeout: float = 10) -> float:
+        """Wait for the next transmit PDO.
+
+        :param float timeout: Max time to wait in seconds.
+        :return: Timestamp of message received or None if timeout.
+        """
+        async with self.receive_condition:
+            self.is_received = False
+            await self.receive_condition.wait()
+        return self.timestamp if self.is_received else None
+
 
 class Variable(variable.Variable):
     """One object dictionary variable mapped to a PDO."""
@@ -571,6 +655,11 @@ def get_data(self) -> bytes:
 
         return data
 
+    async def aget_data(self) -> bytes:
+        # As long as get_data() is not making any IO, it can be called
+        # directly with no special async variant
+        return self.get_data()
+
     def set_data(self, data: bytes):
         """Set for the given variable the PDO data.
 
@@ -603,3 +692,8 @@ def set_data(self, data: bytes):
             self.pdo_parent.data[byte_offset:byte_offset + len(data)] = data
 
         self.pdo_parent.update()
+
+    async def aset_data(self, data: bytes):
+        # As long as get_data() is not making any IO, it can be called
+        # directly with no special async variant
+        return self.set_data(data)

canopen/profiles/p402.py

@@ -391,6 +391,7 @@ def op_mode(self):
         :raises TypeError: When setting a mode not advertised as supported by the node.
         :raises RuntimeError: If the switch is not confirmed within the configured timeout.
         """
+        logger.warning("Accessing BaseNode402.op_mode property is deprecated")
         try:
             pdo = self.tpdo_pointers[0x6061].pdo_parent
             if pdo.is_periodic:
@@ -406,6 +407,7 @@ def op_mode(self):
 
     @op_mode.setter
     def op_mode(self, mode):
+        logger.warning("Accessing BaseNode402.op_mode setter is deprecated")
         try:
             if not self.is_op_mode_supported(mode):
                 raise TypeError(
@@ -470,6 +472,7 @@ def statusword(self):
         If the object 0x6041 is not configured in any TPDO it will fall back to the SDO
         mechanism and try to get the value.
         """
+        logger.warning("Accessing BaseNode402.statusword property is deprecated")
         try:
             return self.tpdo_values[0x6041]
         except KeyError:
@@ -507,6 +510,7 @@ def controlword(self):
 
     @controlword.setter
     def controlword(self, value):
+        logger.warning("Accessing BaseNode402.controlword setter is deprecated")
         if 0x6040 in self.rpdo_pointers:
             self.rpdo_pointers[0x6040].raw = value
             pdo = self.rpdo_pointers[0x6040].pdo_parent
@@ -536,6 +540,7 @@ def state(self):
         :raises RuntimeError: If the switch is not confirmed within the configured timeout.
         :raises ValueError: Trying to execute a illegal transition in the state machine.
         """
+        logger.warning("Accessing BaseNode402.state property is deprecated")
         for state, mask_val_pair in State402.SW_MASK.items():
             bitmask, bits = mask_val_pair
             if self.statusword & bitmask == bits:
@@ -544,6 +549,7 @@ def state(self):
 
     @state.setter
     def state(self, target_state):
+        logger.warning("Accessing BaseNode402.state setter is deprecated")
         timeout = time.monotonic() + self.TIMEOUT_SWITCH_STATE_FINAL
         while self.state != target_state:
             next_state = self._next_state(target_state)

canopen/sdo/base.py

@@ -73,6 +73,9 @@ def __contains__(self, key: Union[int, str]) -> bool:
     def upload(self, index: int, subindex: int) -> bytes:
         raise NotImplementedError()
 
+    async def aupload(self, index: int, subindex: int) -> bytes:
+        raise NotImplementedError()
+
     def download(
         self,
         index: int,
@@ -82,6 +85,15 @@ def download(
     ) -> None:
         raise NotImplementedError()
 
+    async def adownload(
+        self,
+        index: int,
+        subindex: int,
+        data: bytes,
+        force_segment: bool = False,
+    ) -> None:
+        raise NotImplementedError()
+
 
 class Record(Mapping):
 
@@ -131,10 +143,17 @@ def __init__(self, sdo_node: SdoBase, od: objectdictionary.ObjectDictionary):
     def get_data(self) -> bytes:
         return self.sdo_node.upload(self.od.index, self.od.subindex)
 
+    async def aget_data(self) -> bytes:
+        return await self.sdo_node.aupload(self.od.index, self.od.subindex)
+
     def set_data(self, data: bytes):
         force_segment = self.od.data_type == objectdictionary.DOMAIN
         self.sdo_node.download(self.od.index, self.od.subindex, data, force_segment)
 
+    async def aset_data(self, data: bytes):
+        force_segment = self.od.data_type == objectdictionary.DOMAIN
+        await self.sdo_node.adownload(self.od.index, self.od.subindex, data, force_segment)
+
     def open(self, mode="rb", encoding="ascii", buffering=1024, size=None,
              block_transfer=False, request_crc_support=True):
         """Open the data stream as a file like object.