Merge branch 'master' into sdo-upload-disregard-od

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
@@ -36,7 +36,7 @@ Incomplete support for creating slave nodes also exists.
 Installation
 ------------
 
-Install from PyPI_ using :program:`pip`::
+Install from PyPI_ using ``pip``::
 
     $ pip install canopen
 
@@ -56,7 +56,7 @@ Unit tests can be run using the pytest_ framework::
     $ pip install -r requirements-dev.txt
     $ pytest -v
 
-You can also use :mod:`unittest` standard library module::
+You can also use ``unittest`` standard library module::
 
     $ python3 -m unittest discover test -v
 

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/objectdictionary/eds.py

@@ -473,7 +473,7 @@ def export_record(var, eds):
     for rate in od.device_information.allowed_baudrates.union(
             {10e3, 20e3, 50e3, 125e3, 250e3, 500e3, 800e3, 1000e3}):
         eds.set(
-            "DeviceInfo", f"BaudRate_{rate//1000}",
+            "DeviceInfo", f"BaudRate_{int(rate//1000)}",
             int(rate in od.device_information.allowed_baudrates))
 
     if device_commisioning and (od.bitrate or od.node_id):

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
@@ -150,7 +150,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(
@@ -196,7 +196,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()
@@ -403,19 +403,27 @@ def save(self) -> None:
             logger.info("Skip saving %s: COB-ID was never set", self.com_record.od.name)
             return
         logger.info("Setting COB-ID 0x%X and temporarily disabling PDO", self.cob_id)
-        self.com_record[1].raw = self.cob_id | PDO_NOT_VALID | (RTR_NOT_ALLOWED if not self.rtr_allowed else 0x0)
-        if self.trans_type is not None:
-            logger.info("Setting transmission type to %d", self.trans_type)
-            self.com_record[2].raw = self.trans_type
-        if self.inhibit_time is not None:
-            logger.info("Setting inhibit time to %d us", (self.inhibit_time * 100))
-            self.com_record[3].raw = self.inhibit_time
-        if self.event_timer is not None:
-            logger.info("Setting event timer to %d ms", self.event_timer)
-            self.com_record[5].raw = self.event_timer
-        if self.sync_start_value is not None:
-            logger.info("Setting SYNC start value to %d", self.sync_start_value)
-            self.com_record[6].raw = self.sync_start_value
+        self.com_record[1].raw = (
+            self.cob_id
+            | PDO_NOT_VALID
+            | (RTR_NOT_ALLOWED if not self.rtr_allowed else 0)
+        )
+
+        def _set_com_record(
+            subindex: int, value: Optional[int], log_fmt: str, log_factor: int = 1
+        ):
+            if value is None:
+                return
+            if self.com_record[subindex].writable:
+                logger.info(f"Setting {log_fmt}", value * log_factor)
+                self.com_record[subindex].raw = value
+            else:
+                logger.info(f"Cannot set {log_fmt}, not writable", value * log_factor)
+
+        _set_com_record(2, self.trans_type, "transmission type to %d")
+        _set_com_record(3, self.inhibit_time, "inhibit time to %d us", 100)
+        _set_com_record(5, self.event_timer, "event timer to %d ms")
+        _set_com_record(6, self.sync_start_value, "SYNC start value to %d")
 
         try:
             self.map_array[0].raw = 0
@@ -425,20 +433,21 @@ def save(self) -> None:
             # 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%04X:%02X, %d bits) to PDO map",
-                        var.name, var.index, var.subindex, var.length)
+        for var, entry in zip(self.map, self.map_array.values()):
+            if not entry.od.writable:
+                continue
+            logger.info(
+                "Writing %s (0x%04X:%02X, %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)
+                entry.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
+                entry.raw = var.index << 16 | var.subindex << 8 | var.length
         try:
             self.map_array[0].raw = len(self.map)
         except SdoAbortedError as e:

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