Autofix ruff issues, ignore those requiring refactoring

Author: juliuszt

canopen/emcy.py

@@ -16,9 +16,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()
 

canopen/lss.py

@@ -140,10 +140,7 @@ def send_switch_state_selective(
         )
 
         cs = struct.unpack_from("<B", response)[0]
-        if cs == CS_SWITCH_STATE_SELECTIVE_RESPONSE:
-            return True
-
-        return False
+        return cs == CS_SWITCH_STATE_SELECTIVE_RESPONSE
 
     def inquire_node_id(self):
         """Read the node id.
@@ -312,10 +309,7 @@ def __send_fast_scan_message(self, id_number, bit_checker, lss_sub, lss_next):
             return False
 
         cs = struct.unpack_from("<B", recv_msg)[0]
-        if cs == CS_IDENTIFY_SLAVE:
-            return True
-
-        return False
+        return cs == CS_IDENTIFY_SLAVE
 
     def __send_lss_address(self, req_cs, number):
         message = bytearray(8)
@@ -408,8 +402,8 @@ def __send_command(self, message):
         # TODO check if the response is LSS response message
         try:
             response = self.responses.get(block=True, timeout=self.RESPONSE_TIMEOUT)
-        except queue.Empty:
-            raise LssError("No LSS response received")
+        except queue.Empty as err:
+            raise LssError("No LSS response received") from err
 
         return response
 

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
 
 import can
 from can import Listener
@@ -27,7 +27,7 @@ class Network(MutableMapping):
     NOTIFIER_CYCLE: float = 1.0  #: Maximum waiting time for one notifier iteration.
     NOTIFIER_SHUTDOWN_TIMEOUT: float = 5.0  #: Maximum waiting time to stop notifiers.
 
-    def __init__(self, bus: Optional[can.BusABC] = None):
+    def __init__(self, bus: can.BusABC | None = None):
         """
         :param can.BusABC bus:
             A python-can bus instance to re-use.
@@ -40,9 +40,9 @@ def __init__(self, bus: Optional[can.BusABC] = None):
         #: List of :class:`can.Listener` objects.
         #: Includes at least MessageListener.
         self.listeners = [MessageListener(self)]
-        self.notifier: Optional[can.Notifier] = None
-        self.nodes: Dict[int, Union[RemoteNode, LocalNode]] = {}
-        self.subscribers: Dict[int, List[Callback]] = {}
+        self.notifier: can.Notifier | None = None
+        self.nodes: dict[int, RemoteNode | LocalNode] = {}
+        self.subscribers: dict[int, list[Callback]] = {}
         self.send_lock = threading.Lock()
         self.sync = SyncProducer(self)
         self.time = TimeProducer(self)
@@ -61,7 +61,7 @@ def subscribe(self, can_id: int, callback: Callback) -> None:
         :param callback:
             Function to call when message is received.
         """
-        self.subscribers.setdefault(can_id, list())
+        self.subscribers.setdefault(can_id, [])
         if callback not in self.subscribers[can_id]:
             self.subscribers[can_id].append(callback)
 
@@ -133,8 +133,8 @@ def __exit__(self, type, value, traceback):
 
     def add_node(
         self,
-        node: Union[int, RemoteNode, LocalNode],
-        object_dictionary: Union[str, ObjectDictionary, None] = None,
+        node: int | RemoteNode | LocalNode,
+        object_dictionary: str | ObjectDictionary | None = None,
         upload_eds: bool = False,
     ) -> RemoteNode:
         """Add a remote node to the network.
@@ -163,7 +163,7 @@ def add_node(
     def create_node(
         self,
         node: int,
-        object_dictionary: Union[str, ObjectDictionary, None] = None,
+        object_dictionary: str | ObjectDictionary | None = None,
     ) -> LocalNode:
         """Create a local node in the network.
 
@@ -261,10 +261,10 @@ def check(self) -> None:
                 logger.error("An error has caused receiving of messages to stop")
                 raise exc
 
-    def __getitem__(self, node_id: int) -> Union[RemoteNode, LocalNode]:
+    def __getitem__(self, node_id: int) -> RemoteNode | LocalNode:
         return self.nodes[node_id]
 
-    def __setitem__(self, node_id: int, node: Union[RemoteNode, LocalNode]):
+    def __setitem__(self, node_id: int, node: RemoteNode | LocalNode):
         assert node_id == node.id
         if node_id in self.nodes:
             # Remove old callbacks
@@ -286,7 +286,7 @@ def __len__(self) -> int:
 class _UninitializedNetwork(Network):
     """Empty network implementation as a placeholder before actual initialization."""
 
-    def __init__(self, bus: Optional[can.BusABC] = None):
+    def __init__(self, bus: can.BusABC | None = None):
         """Do not initialize attributes, by skipping the parent constructor."""
 
     def __getattribute__(self, name):
@@ -397,10 +397,10 @@ class NodeScanner:
 
     SERVICES = (0x700, 0x580, 0x180, 0x280, 0x380, 0x480, 0x80)
 
-    def __init__(self, network: Optional[Network] = None):
+    def __init__(self, network: Network | None = None):
         self.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

@@ -102,8 +102,7 @@ def state(self, new_state: str):
             code = NMT_COMMANDS[new_state]
         else:
             raise ValueError(
-                "'%s' is an invalid state. Must be one of %s."
-                % (new_state, ", ".join(NMT_COMMANDS))
+                f"'{new_state}' is an invalid state. Must be one of {', '.join(NMT_COMMANDS)}."
             )
 
         self.send_command(code)
@@ -112,7 +111,7 @@ def state(self, new_state: str):
 class NmtMaster(NmtBase):
 
     def __init__(self, node_id: int):
-        super(NmtMaster, self).__init__(node_id)
+        super().__init__(node_id)
         self._state_received = None
         self._node_guarding_producer: Optional[PeriodicMessageTask] = None
         #: Timestamp of last heartbeat message
@@ -143,7 +142,7 @@ def send_command(self, code: int):
         :param code:
             NMT command code.
         """
-        super(NmtMaster, self).send_command(code)
+        super().send_command(code)
         logger.info("Sending NMT command 0x%X to node %d", code, self.id)
         self.network.send_message(0, [code, self.id])
 
@@ -205,13 +204,13 @@ class NmtSlave(NmtBase):
     """
 
     def __init__(self, node_id: int, local_node):
-        super(NmtSlave, self).__init__(node_id)
+        super().__init__(node_id)
         self._send_task: Optional[PeriodicMessageTask] = None
         self._heartbeat_time_ms = 0
         self._local_node = local_node
 
     def on_command(self, can_id, data, timestamp):
-        super(NmtSlave, self).on_command(can_id, data, timestamp)
+        super().on_command(can_id, data, timestamp)
         self.update_heartbeat()
 
     def send_command(self, code: int) -> None:
@@ -221,7 +220,7 @@ def send_command(self, code: int) -> None:
             NMT command code.
         """
         old_state = self._state
-        super(NmtSlave, self).send_command(code)
+        super().send_command(code)
 
         if self._state == 0:
             logger.info("Sending boot-up message")

canopen/node/__init__.py

@@ -1,2 +1,4 @@
 from canopen.node.local import LocalNode
 from canopen.node.remote import RemoteNode
+
+__all__ = ["LocalNode", "RemoteNode"]

canopen/node/local.py

@@ -1,7 +1,6 @@
 from __future__ import annotations
 
 import logging
-from typing import Dict, Union
 
 import canopen.network
 from canopen import objectdictionary
@@ -20,11 +19,11 @@ class LocalNode(BaseNode):
     def __init__(
         self,
         node_id: int,
-        object_dictionary: Union[ObjectDictionary, str],
+        object_dictionary: ObjectDictionary | str,
     ):
-        super(LocalNode, self).__init__(node_id, object_dictionary)
+        super().__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/node/remote.py

@@ -1,7 +1,7 @@
 from __future__ import annotations
 
 import logging
-from typing import TextIO, Union
+from typing import TextIO
 
 import canopen.network
 from canopen.emcy import EmcyConsumer
@@ -30,10 +30,10 @@ class RemoteNode(BaseNode):
     def __init__(
         self,
         node_id: int,
-        object_dictionary: Union[ObjectDictionary, str, TextIO],
+        object_dictionary: ObjectDictionary | str | TextIO,
         load_od: bool = False,
     ):
-        super(RemoteNode, self).__init__(node_id, object_dictionary)
+        super().__init__(node_id, object_dictionary)
 
         #: Enable WORKAROUND for reversed PDO mapping entries
         self.curtis_hack = False
@@ -164,7 +164,7 @@ def load_configuration(self) -> None:
             if 0x1400 <= obj.index < 0x1C00:
                 # Ignore PDO related objects
                 continue
-            if isinstance(obj, ODRecord) or isinstance(obj, ODArray):
+            if isinstance(obj, (ODRecord, ODArray)):
                 for subobj in obj.values():
                     if (
                         isinstance(subobj, ODVariable)