Merge branch 'master' into pdo-getitem-zero

Author: acolomb

README.rst

@@ -6,7 +6,7 @@ The aim of the project is to support the most common parts of the CiA 301
 standard in a simple Pythonic interface. It is mainly targeted for testing and
 automation tasks rather than a standard compliant master implementation.
 
-The library supports Python 3.8 or newer.
+The library supports Python 3.9 or newer.
 
 
 Features

canopen/emcy.py

@@ -1,8 +1,10 @@
+from __future__ import annotations
+
 import logging
 import struct
 import threading
 import time
-from typing import Callable, List, Optional
+from typing import Callable, Optional
 
 import canopen.network
 
@@ -17,9 +19,9 @@ class EmcyConsumer:
 
     def __init__(self):
         #: Log of all received EMCYs for this node
-        self.log: List["EmcyError"] = []
+        self.log: list[EmcyError] = []
         #: Only active EMCYs. Will be cleared on Error Reset
-        self.active: List["EmcyError"] = []
+        self.active: list[EmcyError] = []
         self.callbacks = []
         self.emcy_received = threading.Condition()
 
@@ -39,7 +41,7 @@ def on_emcy(self, can_id, data, timestamp):
         for callback in self.callbacks:
             callback(entry)
 
-    def add_callback(self, callback: Callable[["EmcyError"], None]):
+    def add_callback(self, callback: Callable[[EmcyError], None]):
         """Get notified on EMCY messages from this node.
 
         :param callback:
@@ -55,7 +57,7 @@ def reset(self):
 
     def wait(
         self, emcy_code: Optional[int] = None, timeout: float = 10
-    ) -> "EmcyError":
+    ) -> EmcyError:
         """Wait for a new EMCY to arrive.
 
         :param emcy_code: EMCY code to wait for

canopen/network.py

@@ -2,11 +2,10 @@
 
 import logging
 import threading
-from collections.abc import MutableMapping
-from typing import Callable, Dict, Final, Iterator, List, Optional, Union
+from collections.abc import Iterator, MutableMapping
+from typing import Callable, Final, Optional, Union
 
 import can
-from can import Listener
 
 from canopen.lss import LssMaster
 from canopen.nmt import NmtMaster
@@ -40,10 +39,10 @@ def __init__(self, bus: Optional[can.BusABC] = None):
         self.scanner = NodeScanner(self)
         #: List of :class:`can.Listener` objects.
         #: Includes at least MessageListener.
-        self.listeners = [MessageListener(self)]
+        self.listeners: list[can.Listener] = [MessageListener(self)]
         self.notifier: Optional[can.Notifier] = None
-        self.nodes: Dict[int, Union[RemoteNode, LocalNode]] = {}
-        self.subscribers: Dict[int, List[Callback]] = {}
+        self.nodes: dict[int, Union[RemoteNode, LocalNode]] = {}
+        self.subscribers: dict[int, list[Callback]] = {}
         self.send_lock = threading.Lock()
         self.sync = SyncProducer(self)
         self.time = TimeProducer(self)
@@ -352,7 +351,7 @@ def update(self, data: bytes) -> None:
             self._start()
 
 
-class MessageListener(Listener):
+class MessageListener(can.Listener):
     """Listens for messages on CAN bus and feeds them to a Network instance.
 
     :param network:
@@ -396,7 +395,7 @@ def __init__(self, network: Optional[Network] = None):
             network = _UNINITIALIZED_NETWORK
         self.network: Network = network
         #: A :class:`list` of nodes discovered
-        self.nodes: List[int] = []
+        self.nodes: list[int] = []
 
     def on_message_received(self, can_id: int):
         service = can_id & 0x780

canopen/nmt.py

@@ -2,7 +2,7 @@
 import struct
 import threading
 import time
-from typing import Callable, Dict, Final, List, Optional, TYPE_CHECKING
+from typing import Callable, Final, Optional, TYPE_CHECKING
 
 import canopen.network
 
@@ -12,7 +12,7 @@
 
 logger = logging.getLogger(__name__)
 
-NMT_STATES: Final[Dict[int, str]] = {
+NMT_STATES: Final[dict[int, str]] = {
     0: 'INITIALISING',
     4: 'STOPPED',
     5: 'OPERATIONAL',
@@ -21,7 +21,7 @@
     127: 'PRE-OPERATIONAL'
 }
 
-NMT_COMMANDS: Final[Dict[str, int]] = {
+NMT_COMMANDS: Final[dict[str, int]] = {
     'OPERATIONAL': 1,
     'STOPPED': 2,
     'SLEEP': 80,
@@ -32,7 +32,7 @@
     'RESET COMMUNICATION': 130
 }
 
-COMMAND_TO_STATE: Final[Dict[int, int]] = {
+COMMAND_TO_STATE: Final[dict[int, int]] = {
     1: 5,
     2: 4,
     80: 80,
@@ -117,7 +117,7 @@ def __init__(self, node_id: int):
         #: Timestamp of last heartbeat message
         self.timestamp: Optional[float] = None
         self.state_update = threading.Condition()
-        self._callbacks: List[Callable[[int], None]] = []
+        self._callbacks: list[Callable[[int], None]] = []
 
     def on_heartbeat(self, can_id, data, timestamp):
         with self.state_update:

canopen/node/local.py

@@ -1,7 +1,7 @@
 from __future__ import annotations
 
 import logging
-from typing import Dict, Union
+from typing import Union
 
 import canopen.network
 from canopen import objectdictionary
@@ -25,7 +25,7 @@ def __init__(
     ):
         super(LocalNode, self).__init__(node_id, object_dictionary)
 
-        self.data_store: Dict[int, Dict[int, bytes]] = {}
+        self.data_store: dict[int, dict[int, bytes]] = {}
         self._read_callbacks = []
         self._write_callbacks = []
 

canopen/objectdictionary/__init__.py

@@ -6,8 +6,8 @@
 
 import logging
 import struct
-from collections.abc import Mapping, MutableMapping
-from typing import Dict, Iterator, List, Optional, TextIO, Union
+from collections.abc import Iterator, Mapping, MutableMapping
+from typing import Optional, TextIO, Union
 
 from canopen.objectdictionary.datatypes import *
 from canopen.objectdictionary.datatypes import IntegerN, UnsignedN
@@ -367,9 +367,9 @@ def __init__(self, name: str, index: int, subindex: int = 0):
         #: Description of variable
         self.description: str = ""
         #: Dictionary of value descriptions
-        self.value_descriptions: Dict[int, str] = {}
+        self.value_descriptions: dict[int, str] = {}
         #: Dictionary of bitfield definitions
-        self.bit_definitions: Dict[str, List[int]] = {}
+        self.bit_definitions: dict[str, list[int]] = {}
         #: Storage location of index
         self.storage_location = None
         #: Can this variable be mapped to a PDO

canopen/pdo/base.py

@@ -4,8 +4,8 @@
 import logging
 import math
 import threading
-from collections.abc import Mapping
-from typing import Callable, Dict, Iterator, List, Optional, TYPE_CHECKING, Union
+from collections.abc import Iterator, Mapping
+from typing import Callable, Optional, TYPE_CHECKING, Union
 
 import canopen.network
 from canopen import objectdictionary
@@ -154,7 +154,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(
@@ -200,7 +200,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()

canopen/profiles/p402.py

@@ -1,7 +1,6 @@
 # inspired by the NmtMaster code
 import logging
 import time
-from typing import Dict
 
 from canopen.node import RemoteNode
 from canopen.pdo import PdoMap
@@ -215,8 +214,8 @@ class BaseNode402(RemoteNode):
     def __init__(self, node_id, object_dictionary):
         super(BaseNode402, self).__init__(node_id, object_dictionary)
         self.tpdo_values = {}  # { index: value from last received TPDO }
-        self.tpdo_pointers: Dict[int, PdoMap] = {}
-        self.rpdo_pointers: Dict[int, PdoMap] = {}
+        self.tpdo_pointers: dict[int, PdoMap] = {}
+        self.rpdo_pointers: dict[int, PdoMap] = {}
 
     def setup_402_state_machine(self, read_pdos=True):
         """Configure the state machine by searching for a TPDO that has the StatusWord mapped.

canopen/sdo/base.py

@@ -1,8 +1,8 @@
 from __future__ import annotations
 
 import binascii
-from collections.abc import Mapping
-from typing import Iterator, Optional, Union
+from collections.abc import Iterator, Mapping
+from typing import Optional, Union
 
 import canopen.network
 from canopen import objectdictionary
@@ -148,7 +148,18 @@ def __init__(self, sdo_node: SdoBase, od: objectdictionary.ODVariable):
         variable.Variable.__init__(self, od)
 
     def get_data(self) -> bytes:
-        return self.sdo_node.upload(self.od.index, self.od.subindex)
+        data = self.sdo_node.upload(self.od.index, self.od.subindex)
+        response_size = len(data)
+
+        # If size is available through variable in OD, then use the smaller of the two sizes.
+        # Some devices send U32/I32 even if variable is smaller in OD
+        if self.od.fixed_size:
+            # Get the size in bytes for this variable
+            var_size = len(self.od) // 8
+            if response_size is None or var_size < response_size:
+                # Truncate the data to specified size
+                data = data[:var_size]
+        return data
 
     def set_data(self, data: bytes):
         force_segment = self.od.data_type == objectdictionary.DOMAIN

canopen/sdo/client.py

@@ -112,6 +112,10 @@ def abort(self, abort_code=ABORT_GENERAL_ERROR):
     def upload(self, index: int, subindex: int) -> bytes:
         """May be called to make a read operation without an Object Dictionary.
 
+        No validation against the Object Dictionary is performed, even if an object description
+        would be available.  The length of the returned data depends only on the transferred
+        amount, possibly truncated to the size indicated by the server.
+
         :param index:
             Index of object to read.
         :param subindex:
@@ -128,17 +132,8 @@ def upload(self, index: int, subindex: int) -> bytes:
             response_size = fp.size
             data = fp.read()
 
-        # If size is available through variable in OD, then use the smaller of the two sizes.
-        # Some devices send U32/I32 even if variable is smaller in OD
-        var = self.od.get_variable(index, subindex)
-        if var is not None:
-            # Found a matching variable in OD
-            if var.fixed_size:
-                # Get the size in bytes for this variable
-                var_size = len(var) // 8
-                if response_size is None or var_size < response_size:
-                    # Truncate the data to specified size
-                    data = data[0:var_size]
+        if response_size and response_size < len(data):
+            data = data[:response_size]
         return data
 
     def download(