Lint and type hints for REPL

pymodbus/constants.py

@@ -142,7 +142,7 @@ class Defaults(Singleton):  # pylint: disable=too-few-public-methods
     TransactionId = 0
     Strict = True
     ProtocolId = 0
-    Slave = 0x00
+    Slave: int = 0x00
     Baudrate = 19200
     Parity = "N"
     Bytesize = 8

pymodbus/repl/client/main.py

@@ -1,25 +1,9 @@
 """Pymodbus REPL Entry point."""
-# pylint: disable=anomalous-backslash-in-string
-# flake8: noqa
 import logging
 import pathlib
-import sys
-
-
-try:
-    import click
-except ImportError:
-    print('click not installed!! Install with "pip install click"')
-    sys.exit(1)
-try:
-    from prompt_toolkit import PromptSession, print_formatted_text
-except ImportError:
-    print(
-        "prompt toolkit is not installed!! "
-        'Install with "pip install prompt_toolkit --upgrade"'
-    )
-    sys.exit(1)
 
+import click
+from prompt_toolkit import PromptSession, print_formatted_text
 from prompt_toolkit.auto_suggest import AutoSuggestFromHistory
 from prompt_toolkit.formatted_text import HTML
 from prompt_toolkit.history import FileHistory
@@ -46,9 +30,7 @@
 
 _logger = logging.getLogger()
 
-click.disable_unicode_literals_warning = True
-
-TITLE = f"""
+TITLE = rf"""
 ----------------------------------------------------------------------------
 __________          _____             .___  __________              .__
 \______   \___.__. /     \   ____   __| _/  \______   \ ____ ______ |  |
@@ -124,8 +106,8 @@ def convert(self, value, param, ctx):
         return None
 
 
-def process_args(args: list, string: bool = True):
-    """Internal function to parse arguments provided on command line.
+def _process_args(args: list, string: bool = True):
+    """Parse arguments provided on command line.
 
     :param args: Array of argument values
     :param string: True if arguments values are strings, false if argument values are integers
@@ -226,7 +208,7 @@ def _(event):
                     text = text.strip().split()
                     cmd = text[0].split(".")[1]
                     args = text[1:]
-                    kwargs, execute = process_args(args, string=False)
+                    kwargs, execute = _process_args(args, string=False)
                     if execute:
                         if text[0] in CLIENT_ATTRIBUTES:
                             result = Result(getattr(client, cmd))
@@ -242,7 +224,7 @@ def _(event):
                         result.raw()
                     if words[0] == "result.decode":
                         args = words[1:]
-                        kwargs, execute = process_args(args)
+                        kwargs, execute = _process_args(args)
                         if execute:
                             result.decode(**kwargs)
         except KeyboardInterrupt:

pymodbus/repl/client/mclient.py

@@ -220,10 +220,10 @@ def read_input_registers(self, address, count=1, slave=Defaults.Slave, **kwargs)
 
     def readwrite_registers(
         self,
-        read_address,
-        read_count,
-        write_address,
-        write_registers,
+        read_address=0,
+        read_count=0,
+        write_address=0,
+        values=0,
         slave=Defaults.Slave,
         **kwargs,
     ):
@@ -235,7 +235,7 @@ def readwrite_registers(
         :param read_address: register offset to read from
         :param read_count: Number of registers to read
         :param write_address: register offset to write to
-        :param write_registers: List of register values to write (comma separated)
+        :param values: List of register values to write (comma separated)
         :param slave: Modbus slave unit ID
         :param kwargs:
         :return:
@@ -244,7 +244,7 @@ def readwrite_registers(
             read_address=read_address,
             read_count=read_count,
             write_address=write_address,
-            write_registers=write_registers,
+            values=values,
             slave=slave,
             **kwargs,
         )
@@ -321,15 +321,14 @@ def report_slave_id(self, slave=Defaults.Slave, **kwargs):
             }
         return ExtendedRequestSupport._process_exception(resp)
 
-    def read_exception_status(self, **kwargs):
-        """Read tcontents of eight Exception Status output.
-
-        In a remote device.
+    def read_exception_status(self, slave=Defaults.Slave, **kwargs):
+        """Read contents of eight Exception Status output in a remote device.
 
+        :param slave: Modbus slave unit ID
         :param kwargs:
         :return:
         """
-        request = ReadExceptionStatusRequest(**kwargs)
+        request = ReadExceptionStatusRequest(slave, **kwargs)
         resp = self.execute(request)  # pylint: disable=no-member
         if not resp.isError():
             return {"function_code": resp.function_code, "status": resp.status}

pymodbus/repl/server/README.md

@@ -1,6 +1,6 @@
 # Pymodbus REPL Server
 
-Pymodbus REPL server helps to quicky spin an [asynchronous server](../../../examples/common/asyncio_server.py) from command line.
+Pymodbus REPL server helps to quickly start an [asynchronous server](../../../examples/common/asyncio_server.py) from the command line.
 
 Support both `Modbus TCP` and `Modbus RTU` server.
 
@@ -9,7 +9,7 @@ Some features offered are
 
 ---
 1. Runs a [reactive server](../../server/reactive/main.py) in `REPL` and `NON REPL` mode.
-2. Exposes REST API's to manipulate the behaviour of the server in non repl mode.
+2. Exposes REST API's to manipulate the behaviour of the server in non REPL mode.
 3. Ability to manipulate the out-going response dynamically (either via REPL console or via REST API request).
 4. Ability to delay the out-going response dynamically (either via REPL console or via REST API request).
 5. Auto revert to normal response after pre-defined number of manipulated responses.

pymodbus/repl/server/main.py

@@ -1,6 +1,4 @@
 """Repl server main."""
-from __future__ import annotations
-
 import asyncio
 import json
 import sys
@@ -71,11 +69,11 @@ def servers(incomplete: str) -> List[str]:
     return _completer(incomplete, _servers)
 
 
-def process_extra_args(extra_args: list[str], modbus_config: dict) -> dict:
+def process_extra_args(extra_args: List[str], modbus_config: dict) -> dict:
     """Process extra args passed to server."""
     options_stripped = [x.strip().replace("--", "") for x in extra_args[::2]]
-    extra_args = dict(list(zip(options_stripped, extra_args[1::2])))
-    for option, value in extra_args.items():
+    extra_args_dict = dict(list(zip(options_stripped, extra_args[1::2])))
+    for option, value in extra_args_dict.items():
         if option in modbus_config:
             try:
                 modbus_config[option] = type(modbus_config[option])(value)

pymodbus/server/simulator/http_server.py

@@ -173,8 +173,8 @@ def __init__(
             "server_json": [None, self.build_json_server],
         }
         for entry in self.generator_html:  # pylint: disable=consider-using-dict-items
-            file = os.path.join(self.web_path, "generator", entry)
-            with open(file, encoding="utf-8") as handle:
+            html_file = os.path.join(self.web_path, "generator", entry)
+            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] = []

requirements.txt

@@ -28,6 +28,7 @@ aiohttp==3.8.3
 typer[all]==0.7.0
 prompt-toolkit==3.0.36
 pygments==2.14.0
+click>=8.0.0
 
 # install:serial
 pyserial>=3.5

test/test_repl_client.py

@@ -1,5 +1,5 @@
 """Test client sync."""
-from pymodbus.repl.client.main import process_args
+from pymodbus.repl.client.main import _process_args
 from pymodbus.server.reactive.default_config import DEFAULT_CONFIG
 
 
@@ -10,39 +10,39 @@ def test_repl_default_config():
 
 
 def test_repl_client_process_args():
-    """Test argument processing in repl.client.main ( process_args function)."""
-    resp = process_args(["address=11"], False)
+    """Test argument processing in repl.client.main ( _process_args function)."""
+    resp = _process_args(["address=11"], False)
     assert resp == ({"address": 11}, True)
 
-    resp = process_args(["address=0x11"], False)
+    resp = _process_args(["address=0x11"], False)
     assert resp == ({"address": 17}, True)
 
-    resp = process_args(["address=0b11"], False)
+    resp = _process_args(["address=0b11"], False)
     assert resp == ({"address": 3}, True)
 
-    resp = process_args(["address=0o11"], False)
+    resp = _process_args(["address=0o11"], False)
     assert resp == ({"address": 9}, True)
 
-    resp = process_args(["address=11", "value=0x10"], False)
+    resp = _process_args(["address=11", "value=0x10"], False)
     assert resp == ({"address": 11, "value": 16}, True)
 
-    resp = process_args(["value=11", "address=0x10"], False)
+    resp = _process_args(["value=11", "address=0x10"], False)
     assert resp == ({"address": 16, "value": 11}, True)
 
-    resp = process_args(["address=0b11", "value=0x10"], False)
+    resp = _process_args(["address=0b11", "value=0x10"], False)
     assert resp == ({"address": 3, "value": 16}, True)
 
     try:
-        resp = process_args(["address=0xhj", "value=0x10"], False)
+        resp = _process_args(["address=0xhj", "value=0x10"], False)
     except ValueError:
         pass
 
     try:
-        resp = process_args(["address=11ah", "value=0x10"], False)
+        resp = _process_args(["address=11ah", "value=0x10"], False)
     except ValueError:
         pass
 
     try:
-        resp = process_args(["address=0b12", "value=0x10"], False)
+        resp = _process_args(["address=0b12", "value=0x10"], False)
     except ValueError:
         pass