Fixed PR after Review. Removed all type-hint changes. Still in this PR:

- save(): Check for cob-id == None
- save(): Removed always true condition
- use getattr directly for curtis hack
- remove unused variables

Author: bizfsc

canopen/pdo/base.py

@@ -1,7 +1,6 @@
-from __future__ import annotations
 import threading
 import math
-from typing import Callable, Dict, Iterable, List, Optional, Union, TYPE_CHECKING
+from typing import Callable, Dict, Iterable, List, Optional, Union
 from collections.abc import Mapping
 import logging
 import binascii
@@ -10,12 +9,6 @@
 from canopen import objectdictionary
 from canopen import variable
 
-if TYPE_CHECKING:
-    from canopen.network import Network
-    from canopen import LocalNode, RemoteNode
-    from canopen.pdo import RPDO, TPDO
-    from canopen.sdo import SdoRecord
-
 PDO_NOT_VALID = 1 << 31
 RTR_NOT_ALLOWED = 1 << 30
 
@@ -29,10 +22,10 @@ class PdoBase(Mapping):
         Parent object associated with this PDO instance
     """
 
-    def __init__(self, node: Union[LocalNode, RemoteNode]):
-        self.network: Optional[Network] = None
-        self.map: Optional[PdoMaps] = None
-        self.node: Union[LocalNode, RemoteNode] = node
+    def __init__(self, node):
+        self.network = None
+        self.map = None  # instance of PdoMaps
+        self.node = node
 
     def __iter__(self):
         return iter(self.map)
@@ -138,7 +131,7 @@ def __init__(self, com_offset, map_offset, pdo_node: PdoBase, cob_base=None):
         :param pdo_node:
         :param cob_base:
         """
-        self.maps: Dict[int, PdoMap] = {}
+        self.maps: Dict[int, "PdoMap"] = {}
         for map_no in range(512):
             if com_offset + map_no in pdo_node.node.object_dictionary:
                 new_map = PdoMap(
@@ -150,7 +143,7 @@ def __init__(self, com_offset, map_offset, pdo_node: PdoBase, cob_base=None):
                     new_map.predefined_cob_id = cob_base + map_no * 0x100 + pdo_node.node.id
                 self.maps[map_no + 1] = new_map
 
-    def __getitem__(self, key: int) -> PdoMap:
+    def __getitem__(self, key: int) -> "PdoMap":
         return self.maps[key]
 
     def __iter__(self) -> Iterable[int]:
@@ -164,9 +157,9 @@ class PdoMap:
     """One message which can have up to 8 bytes of variables mapped."""
 
     def __init__(self, pdo_node, com_record, map_array):
-        self.pdo_node: Union[TPDO, RPDO] = pdo_node
-        self.com_record: SdoRecord = com_record
-        self.map_array: SdoRecord = map_array
+        self.pdo_node = pdo_node
+        self.com_record = com_record
+        self.map_array = map_array
         #: If this map is valid
         self.enabled: bool = False
         #: COB-ID for this PDO
@@ -184,7 +177,7 @@ def __init__(self, pdo_node, com_record, map_array):
         #: Ignores SYNC objects up to this SYNC counter value (optional)
         self.sync_start_value: Optional[int] = None
         #: List of variables mapped to this PDO
-        self.map: List[PdoVariable] = []
+        self.map: List["PdoVariable"] = []
         self.length: int = 0
         #: Current message data
         self.data = bytearray()
@@ -221,7 +214,7 @@ def __getitem_by_name(self, value):
         raise KeyError('{0} not found in map. Valid entries are {1}'.format(
             value, ', '.join(valid_values)))
 
-    def __getitem__(self, key: Union[int, str]) -> PdoVariable:
+    def __getitem__(self, key: Union[int, str]) -> "PdoVariable":
         if isinstance(key, int):
             # there is a maximum available of 8 slots per PDO map
             if key in range(0, 8):
@@ -235,7 +228,7 @@ def __getitem__(self, key: Union[int, str]) -> PdoVariable:
                 var = self.__getitem_by_name(key)
         return var
 
-    def __iter__(self) -> Iterable[PdoVariable]:
+    def __iter__(self) -> Iterable["PdoVariable"]:
         return iter(self.map)
 
     def __len__(self) -> int:
@@ -310,7 +303,7 @@ def on_message(self, can_id, data, timestamp):
                 for callback in self.callbacks:
                     callback(self)
 
-    def add_callback(self, callback: Callable[[PdoMap], None]) -> None:
+    def add_callback(self, callback: Callable[["PdoMap"], None]) -> None:
         """Add a callback which will be called on receive.
 
         :param callback:
@@ -395,39 +388,38 @@ def save(self) -> None:
             logger.info("Setting SYNC start value to %d", self.sync_start_value)
             self.com_record[6].raw = self.sync_start_value
 
-        if self.map is not None:
-            try:
-                self.map_array[0].raw = 0
-            except SdoAbortedError:
-                # WORKAROUND for broken implementations: If the array has a
-                # fixed number of entries (count not writable), generate dummy
-                # mappings for an invalid object 0x0000:00 to overwrite any
-                # excess entries with all-zeros.
-                self._fill_map(self.map_array[0].raw)
-            subindex = 1
-            for var in self.map:
-                logger.info("Writing %s (0x%X:%d, %d bits) to PDO map",
-                            var.name, var.index, var.subindex, var.length)
-                if getattr(self.pdo_node.node, "curtis_hack", False):  # Curtis HACK: mixed up field order
-                    self.map_array[subindex].raw = (var.index |
-                                                    var.subindex << 16 |
-                                                    var.length << 24)
-                else:
-                    self.map_array[subindex].raw = (var.index << 16 |
-                                                    var.subindex << 8 |
-                                                    var.length)
-                subindex += 1
-            try:
-                self.map_array[0].raw = len(self.map)
-            except SdoAbortedError as e:
-                # WORKAROUND for broken implementations: If the array
-                # number-of-entries parameter is not writable, we have already
-                # generated the required number of mappings above.
-                if e.code != 0x06010002:
-                    # Abort codes other than "Attempt to write a read-only
-                    # object" should still be reported.
-                    raise
-            self._update_data_size()
+        try:
+            self.map_array[0].raw = 0
+        except SdoAbortedError:
+            # WORKAROUND for broken implementations: If the array has a
+            # fixed number of entries (count not writable), generate dummy
+            # mappings for an invalid object 0x0000:00 to overwrite any
+            # excess entries with all-zeros.
+            self._fill_map(self.map_array[0].raw)
+        subindex = 1
+        for var in self.map:
+            logger.info("Writing %s (0x%X:%d, %d bits) to PDO map",
+                        var.name, var.index, var.subindex, var.length)
+            if getattr(self.pdo_node.node, "curtis_hack", False):  # Curtis HACK: mixed up field order
+                self.map_array[subindex].raw = (var.index |
+                                                var.subindex << 16 |
+                                                var.length << 24)
+            else:
+                self.map_array[subindex].raw = (var.index << 16 |
+                                                var.subindex << 8 |
+                                                var.length)
+            subindex += 1
+        try:
+            self.map_array[0].raw = len(self.map)
+        except SdoAbortedError as e:
+            # WORKAROUND for broken implementations: If the array
+            # number-of-entries parameter is not writable, we have already
+            # generated the required number of mappings above.
+            if e.code != 0x06010002:
+                # Abort codes other than "Attempt to write a read-only
+                # object" should still be reported.
+                raise
+        self._update_data_size()
 
         if self.enabled:
             self.com_record[1].raw = self.cob_id | (RTR_NOT_ALLOWED if not self.rtr_allowed else 0x0)
@@ -455,7 +447,7 @@ def add_variable(
         index: Union[str, int],
         subindex: Union[str, int] = 0,
         length: Optional[int] = None,
-    ) -> PdoVariable:
+    ) -> "PdoVariable":
         """Add a variable from object dictionary as the next entry.
 
         :param index: Index of variable as name or number