Merge branch 'dev' into ruff

Author: janiversen

pymodbus/client/base.py

@@ -219,7 +219,7 @@ def callback_data(self, data: bytes, addr: tuple = None) -> int:
         self.framer.processIncomingPacket(data, self._handle_response, slave=0)
         return len(data)
 
-    def callback_disconnected(self, _reason: Exception) -> None:
+    def callback_disconnected(self, _reason: Exception | None) -> None:
         """Handle lost connection"""
         for tid in list(self.transaction):
             self.raise_future(

pymodbus/datastore/simulator.py

@@ -1,9 +1,11 @@
 """Pymodbus ModbusSimulatorContext."""
+from __future__ import annotations
+
 import dataclasses
 import random
 import struct
 from datetime import datetime
-from typing import Any, Callable, Dict, List
+from typing import Any, Callable
 
 
 WORD_SIZE = 16
@@ -30,7 +32,7 @@ class Cell:
     access: bool = False
     value: int = 0
     action: int = 0
-    action_kwargs: Dict[str, Any] = None
+    action_kwargs: dict[str, Any] | None = None
     count_read: int = 0
     count_write: int = 0
 
@@ -452,18 +454,18 @@ class ModbusSimulatorContext:
     start_time = int(datetime.now().timestamp())
 
     def __init__(
-        self, config: Dict[str, Any], custom_actions: Dict[str, Callable]
+        self, config: dict[str, Any], custom_actions: dict[str, Callable]
     ) -> None:
         """Initialize."""
-        self.registers: List[int] = []
-        self.fc_offset: Dict[int, int] = {}
+        self.registers: list[int] = []
+        self.fc_offset: dict[int, int] = {}
         self.register_count = 0
         self.type_exception = False
-        self.action_name_to_id: Dict[str, int] = {}
-        self.action_id_to_name: List[str] = []
-        self.action_methods: List[Callable] = []
-        self.registerType_name_to_id: Dict[str, int] = {}
-        self.registerType_id_to_name: List[str] = []
+        self.action_name_to_id: dict[str, int] = {}
+        self.action_id_to_name: list[str] = []
+        self.action_methods: list[Callable] = []
+        self.registerType_name_to_id: dict[str, int] = {}
+        self.registerType_id_to_name: list[str] = []
         Setup(self).setup(config, custom_actions)
 
     # --------------------------------------------

pymodbus/server/async_io.py

@@ -1,11 +1,12 @@
 """Implementation of a Threaded Modbus Server."""
 # pylint: disable=missing-type-doc
+from __future__ import annotations
+
 import asyncio
 import os
 import time
 import traceback
 from contextlib import suppress
-from typing import Union
 
 from pymodbus.datastore import ModbusServerContext
 from pymodbus.device import ModbusControlBlock, ModbusDeviceIdentification
@@ -74,7 +75,7 @@ def callback_connected(self) -> None:
                 traceback.format_exc(),
             )
 
-    def callback_disconnected(self, call_exc: Exception) -> None:
+    def callback_disconnected(self, call_exc: Exception | None) -> None:
         """Call when connection is lost."""
         try:
             if self.handler_task:
@@ -229,9 +230,9 @@ async def _recv_(self):  # pragma: no cover
             result = None
         return result
 
-    def callback_data(self, data: bytes, addr: tuple = None) -> int:
+    def callback_data(self, data: bytes, addr: tuple = ()) -> int:
         """Handle received data."""
-        if addr:
+        if addr != ():
             self.receive_queue.put_nowait((data, addr))
         else:
             self.receive_queue.put_nowait(data)
@@ -554,7 +555,7 @@ class _serverList:
     :meta private:
     """
 
-    active_server: Union[ModbusTcpServer, ModbusUdpServer, ModbusSerialServer] = None
+    active_server: ModbusTcpServer | ModbusUdpServer | ModbusSerialServer
 
     def __init__(self, server):
         """Register new server."""

pymodbus/server/simulator/http_server.py

@@ -1,12 +1,13 @@
 """HTTP server for modbus simulator."""
+from __future__ import annotations
+
 import asyncio
 import contextlib
 import dataclasses
 import importlib
 import json
 import os
 from time import time
-from typing import List
 
 
 try:
@@ -118,7 +119,7 @@ def __init__(
         http_port: int = 8080,
         log_file: str = "server.log",
         json_file: str = "setup.json",
-        custom_actions_module: str = None,
+        custom_actions_module: str | None = None,
     ):
         """Initialize http interface."""
         if AIOHTTP_MISSING:
@@ -139,15 +140,15 @@ def __init__(
             actions_module = importlib.import_module(custom_actions_module)
             custom_actions_dict = actions_module.custom_actions_dict
         else:
-            custom_actions_dict = None
+            custom_actions_dict = {}
         server = setup["server_list"][modbus_server]
         if server["comm"] != "serial":
             server["address"] = (server["host"], server["port"])
             del server["host"]
             del server["port"]
         device = setup["device_list"][modbus_device]
         self.datastore_context = ModbusSimulatorContext(
-            device, custom_actions_dict or None
+            device, custom_actions_dict or {}
         )
         datastore = ModbusServerContext(slaves=self.datastore_context, single=True)
         comm = comm_class[server.pop("comm")]
@@ -200,8 +201,8 @@ def __init__(
             with open(html_file, encoding="utf-8") as handle:
                 self.generator_html[entry][0] = handle.read()
         self.refresh_rate = 0
-        self.register_filter: List[int] = []
-        self.call_list: List[tuple] = []
+        self.register_filter: list[int] = []
+        self.call_list: list[tuple] = []
         self.request_lookup = ServerDecoder.getFCdict()
         self.call_monitor = CallTypeMonitor()
         self.call_response = CallTypeResponse()

pymodbus/transport/transport.py

@@ -374,7 +374,7 @@ def callback_connected(self) -> None:
         """Call when connection is succcesfull."""
         Log.debug("callback_connected called")
 
-    def callback_disconnected(self, exc: Exception) -> None:
+    def callback_disconnected(self, exc: Exception | None) -> None:
         """Call when connection is lost."""
         Log.debug("callback_disconnected called: {}", exc)