Modernize Dict and List type annotations.

Since Python 3.9 we can use the built-in types directly.

Author: acolomb

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

@@ -3,7 +3,7 @@
 import logging
 import threading
 from collections.abc import MutableMapping
-from typing import Callable, Dict, Final, Iterator, List, Optional, Union
+from typing import Callable, Final, Iterator, Optional, Union
 
 import can
 from can import Listener
@@ -40,10 +40,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)
@@ -396,7 +396,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

@@ -7,7 +7,7 @@
 import logging
 import struct
 from collections.abc import Mapping, MutableMapping
-from typing import Dict, Iterator, List, Optional, TextIO, Union
+from typing import Iterator, 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

@@ -5,7 +5,7 @@
 import math
 import threading
 from collections.abc import Mapping
-from typing import Callable, Dict, Iterator, List, Optional, TYPE_CHECKING, Union
+from typing import Callable, Iterator, 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()

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.