Fix more typing issues (#1351)

Type hints (mypy) part 1.

Author: alexrudd2

.gitignore

@@ -2,15 +2,14 @@
 *.pyc
 *.swp
 __pycache__/
-.cache/
+.*cache/
 .coverage
 .DS_Store
 .eggs/
 .idea
 .idea/
 .noseids
 .pymodhis
-.pytest_cache/
 .venv
 .vscode
 .vscode/

README.rst

@@ -277,7 +277,7 @@ The default command can be overridden by passing any valid command at the end.::
 
     Usage: pymodbus.server [OPTIONS] COMMAND [ARGS]...
 
-    Reactive modebus server
+    Reactive Modbus server
 
    ╭─ Options ──────────────────────────────────────────────────────────────────────────────────────────────────────────────────╮
    │ --host                                    TEXT     Host address [default: localhost]                                       │

pymodbus/client/base.py

@@ -64,8 +64,8 @@ def run():
     """
 
     state = ModbusTransactionState.IDLE
-    last_frame_end = 0
-    silent_interval = 0
+    last_frame_end: float = 0
+    silent_interval: float = 0
 
     @dataclass
     class _params:  # pylint: disable=too-many-instance-attributes
@@ -163,7 +163,7 @@ def is_socket_open(self) -> bool:
         """Return whether socket/serial is open or not (call **sync**)."""
         raise NotImplementedException
 
-    def idle_time(self) -> int:
+    def idle_time(self) -> float:
         """Time before initiating next transaction (call **sync**).
 
         Applications can call message functions without checking idle_time(),

pymodbus/client/serial.py

@@ -55,7 +55,7 @@ def __init__(
         framer: ModbusFramer = ModbusRtuFramer,
         baudrate: int = Defaults.Baudrate,
         bytesize: int = Defaults.Bytesize,
-        parity: chr = Defaults.Parity,
+        parity: str = Defaults.Parity,
         stopbits: int = Defaults.Stopbits,
         handle_local_echo: bool = Defaults.HandleLocalEcho,
         **kwargs: Any,
@@ -203,16 +203,16 @@ def run():
     """
 
     state = ModbusTransactionState.IDLE
-    inter_char_timeout = 0
-    silent_interval = 0
+    inter_char_timeout: float = 0
+    silent_interval: float = 0
 
     def __init__(
         self,
         port: str,
         framer: ModbusFramer = ModbusRtuFramer,
         baudrate: int = Defaults.Baudrate,
         bytesize: int = Defaults.Bytesize,
-        parity: chr = Defaults.Parity,
+        parity: str = Defaults.Parity,
         stopbits: int = Defaults.Stopbits,
         handle_local_echo: bool = Defaults.HandleLocalEcho,
         **kwargs: Any,

pymodbus/datastore/simulator.py

@@ -3,7 +3,7 @@
 import random
 import struct
 from datetime import datetime
-from typing import Any, Callable, Dict
+from typing import Any, Callable, Dict, List
 
 
 WORD_SIZE = 16
@@ -30,7 +30,7 @@ class Cell:
     access: bool = False
     value: int = 0
     action: int = 0
-    action_kwargs: int = None
+    action_kwargs: Dict[str, Any] = None
     count_read: int = 0
     count_write: int = 0
 
@@ -455,15 +455,15 @@ def __init__(
         self, config: Dict[str, Any], custom_actions: Dict[str, Callable]
     ) -> None:
         """Initialize."""
-        self.registers = []
-        self.fc_offset = {}
+        self.registers: List[int] = []
+        self.fc_offset: Dict[int, int] = {}
         self.register_count = 0
         self.type_exception = False
-        self.action_name_to_id = {}
-        self.action_id_to_name = []
-        self.action_methods = []
-        self.registerType_name_to_id = {}
-        self.registerType_id_to_name = []
+        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/repl/client/main.py

@@ -256,7 +256,7 @@ def _(event):
 
 
 @click.group("pymodbus-repl")
-@click.version_option(version, message=TITLE)
+@click.version_option(str(version), message=TITLE)
 @click.option("--verbose", is_flag=True, default=False, help="Verbose logs")
 @click.option(
     "--broadcast-support",

pymodbus/repl/server/README.md

@@ -28,7 +28,7 @@ Invoke REPL server with `pymodbus.server run` command.
 
  Usage: pymodbus.server [OPTIONS] COMMAND [ARGS]...
 
- Reactive modebus server
+ Reactive Modbus server
 
 ╭─ Options ──────────────────────────────────────────────────────────────────────────────────────────────────────────────────╮
 │ --host                                    TEXT     Host address [default: localhost]                                       │
@@ -63,7 +63,7 @@ docker run -it pymodbus-dev/pymodbus pymodbus.server --help
 ╭─ Options ──────────────────────────────────────────────────────────────────────────────────────────────────────────────────╮
 │ --modbus-server  -s      TEXT     Modbus Server [default: ModbusServerTypes.tcp]                                           │
 │ --framer         -f      TEXT     Modbus framer to use [default: ModbusFramerTypes.socket]                                 │
-│ --modbus-port    -p      TEXT     Modbus port [default: 5020]                                                              │
+│ --modbus-port    -p      INTEGER  Modbus port [default: 5020]                                                              │
 │ --unit-id        -u      INTEGER  Supported Modbus unit id's [default: None]                                               │
 │ --modbus-config          PATH     Path to additional modbus server config [default: None]                                  │
 │ --random         -r      INTEGER  Randomize every `r` reads. 0=never, 1=always,2=every-second-read, and so on. Applicable  │

pymodbus/repl/server/main.py

@@ -90,7 +90,7 @@ def process_extra_args(extra_args: list[str], modbus_config: dict) -> dict:
 app = typer.Typer(
     no_args_is_help=True,
     context_settings=CONTEXT_SETTING,
-    help="Reactive modebus server",
+    help="Reactive Modbus server",
 )
 
 
@@ -138,7 +138,7 @@ def run(
         autocompletion=framers,
         help="Modbus framer to use",
     ),
-    modbus_port: str = typer.Option("5020", "--modbus-port", "-p", help="Modbus port"),
+    modbus_port: int = typer.Option(5020, "--modbus-port", "-p", help="Modbus port"),
     modbus_unit_id: List[int] = typer.Option(
         None, "--unit-id", "-u", help="Supported Modbus unit id's"
     ),
@@ -180,7 +180,6 @@ def run(
     modbus_config = modbus_config.get(modbus_server, {})
     modbus_config = process_extra_args(extra_args, modbus_config)
     if modbus_server != "serial":
-        modbus_port = int(modbus_port)
         handler = modbus_config.pop("handler", "ModbusConnectedRequestHandler")
     else:
         handler = modbus_config.pop("handler", "ModbusSingleRequestHandler")

pymodbus/server/simulator/http_server.py

@@ -5,6 +5,7 @@
 import json
 import os
 from time import time
+from typing import List
 
 
 try:
@@ -44,8 +45,8 @@ class CallTypeMonitor:
     """Define Request/Response monitor"""
 
     active: bool = False
-    range_start: int = ""
-    range_stop: int = ""
+    range_start: int = -1
+    range_stop: int = -1
     function: int = -1
     hex: bool = False
     decode: bool = False
@@ -176,7 +177,7 @@ def __init__(
             with open(file, encoding="utf-8") as handle:
                 self.generator_html[entry][0] = handle.read()
         self.refresh_rate = 0
-        self.register_filter = []
+        self.register_filter: List[int] = []
         self.call_list = []
         self.call_monitor = CallTypeMonitor()
         self.call_response = CallTypeResponse()
@@ -326,15 +327,11 @@ async def build_html_log(self, _params, html):
 
     def helper_build_html_calls_submit_monitor(self, params):
         """Build html calls submit."""
-        if params["range_start"]:
-            self.call_monitor.range_start = int(params["range_start"])
-            if params["range_stop"]:
-                self.call_monitor.range_stop = int(params["range_stop"])
-            else:
-                self.call_monitor.range_stop = self.call_monitor.range_start
+        self.call_monitor.range_start = params["range_start"]
+        if params["range_stop"] != -1:
+            self.call_monitor.range_stop = params["range_stop"]
         else:
-            self.call_monitor.range_start = ""
-            self.call_monitor.range_stop = ""
+            self.call_monitor.range_stop = self.call_monitor.range_start
         if params["function"]:
             self.call_monitor.function = int(params["function"])
         else:
@@ -412,9 +409,19 @@ async def build_html_calls(self, params, html):
         ):
             selected = "selected" if i == self.call_response.error_response else ""
             function_error += f"<option value={i} {selected}>{txt}</option>"
+        range_start_html = (
+            str(self.call_monitor.range_start)
+            if self.call_monitor.range_start != -1
+            else ""
+        )
+        range_stop_html = (
+            str(self.call_monitor.range_stop)
+            if self.call_monitor.range_stop != -1
+            else ""
+        )
         html = (
-            html.replace("FUNCTION_RANGE_START", str(self.call_monitor.range_start))
-            .replace("FUNCTION_RANGE_STOP", str(self.call_monitor.range_stop))
+            html.replace("FUNCTION_RANGE_START", range_start_html)
+            .replace("FUNCTION_RANGE_STOP", range_stop_html)
             .replace("<!--FUNCTION_TYPES-->", function_error)
             .replace(
                 "FUNCTION_SHOW_HEX_CHECKED", "checked" if self.call_monitor.hex else ""
@@ -493,14 +500,8 @@ async def build_json_server(self, params, json_dict):
 
     def helper_build_filter(self, params):
         """Build list of registers matching filter."""
-        if x := params.get("range_start"):
-            range_start = int(x)
-        else:
-            range_start = -1
-        if x := params.get("range_stop"):
-            range_stop = int(x)
-        else:
-            range_stop = range_start
+        range_start = params.get("range_start", -1)
+        range_stop = params.get("range_stop", range_start)
         reg_action = int(params["action"])
         reg_writeable = "writeable" in params
         reg_type = int(params["type"])

pymodbus/version.py

@@ -22,14 +22,14 @@ def __init__(self, package, major, minor, micro, pre=None):
         self.micro = micro
         self.pre = pre
 
-    def short(self):
+    def short(self) -> str:
         """Return a string in canonical short version format: <major>.<minor>.<micro>.<pre>."""
         pre = ""
         if self.pre:
             pre = f".{self.pre}"
         return f"{self.major}.{self.minor}.{self.micro}{pre}"
 
-    def __str__(self):
+    def __str__(self) -> str:
         """Return a string representation of the object.
 
         :returns: A string representation of this object