Add typing to examples. (#2435)

Author: janiversen

.github/workflows/ci.yml

@@ -105,7 +105,7 @@ jobs:
       - name: mypy
         if: matrix.run_lint == true
         run: |
-          mypy pymodbus
+          mypy pymodbus examples
 
       - name: ruff
         if: matrix.run_lint == true

check_ci.sh

@@ -8,6 +8,6 @@ trap 'echo "\"${last_command}\" command filed with exit code $?."' EXIT
 codespell
 ruff check --fix --exit-non-zero-on-fix .
 pylint --recursive=y examples pymodbus test
-mypy pymodbus
+mypy pymodbus examples
 pytest -x --cov --numprocesses auto
 echo "Ready to push"

examples/client_async.py

@@ -26,13 +26,15 @@
 The corresponding server must be started before e.g. as:
     python3 server_sync.py
 """
+from __future__ import annotations
+
 import asyncio
 import logging
 import sys
 
 
 try:
-    import helper
+    from examples import helper
 except ImportError:
     print("*** ERROR --> THIS EXAMPLE needs the example directory, please see \n\
           https://pymodbus.readthedocs.io/en/latest/source/examples.html\n\
@@ -52,7 +54,7 @@ def setup_async_client(description=None, cmdline=None):
         server=False, description=description, cmdline=cmdline
     )
     _logger.info("### Create client object")
-    client = None
+    client: modbusClient.ModbusBaseClient | None = None
     if args.comm == "tcp":
         client = modbusClient.AsyncModbusTcpClient(
             args.host,

examples/client_async_calls.py

@@ -38,7 +38,7 @@
 
 
 try:
-    import client_async
+    from examples import client_async
 except ImportError:
     print("*** ERROR --> THIS EXAMPLE needs the example directory, please see \n\
           https://pymodbus.readthedocs.io/en/latest/source/examples.html\n\
@@ -157,6 +157,20 @@ async def async_handle_holding_registers(client):
     assert not rr.isError()  # test that call was OK
     assert rr.registers == arguments["values"]
 
+async def async_write_registers_mypy(client):
+    """Read/write holding registers."""
+    regs1: list[int] = [10] * 8
+    await client.write_registers(1, regs1, slave=SLAVE)
+    rr = await client.read_holding_registers(1, len(regs1), slave=SLAVE)
+    assert not rr.isError()  # test that call was OK
+    assert rr.registers == regs1
+
+    regs2: list[bytes] = [b'\x01\x02', b'\x03\x04']
+    await client.write_registers(1, regs2, slave=SLAVE)
+    rr = await client.read_holding_registers(1, len(regs2), slave=SLAVE)
+    assert not rr.isError()  # test that call was OK
+    assert rr.registers == regs2
+
 
 async def async_handle_input_registers(client):
     """Read input registers."""
@@ -258,6 +272,7 @@ async def run_async_calls(client):
     await async_handle_coils(client)
     await async_handle_discrete_input(client)
     await async_handle_holding_registers(client)
+    await async_write_registers_mypy(client)
     await async_handle_input_registers(client)
     await async_handle_file_records(client)
     await async_execute_information_requests(client)

examples/client_calls.py

@@ -36,7 +36,7 @@
 
 
 try:
-    import client_sync
+    from examples import client_sync
 except ImportError:
     print("*** ERROR --> THIS EXAMPLE needs the example directory, please see \n\
           https://pymodbus.readthedocs.io/en/latest/source/examples.html\n\

examples/client_custom_msg.py

@@ -127,13 +127,13 @@ async def main(host="localhost", port=5020):
         # new modbus function code.
         client.register(CustomModbusPDU)
         slave=1
-        request = CustomRequest(32, slave=slave)
-        result = await client.execute(False, request)
+        request1 = CustomRequest(32, slave=slave)
+        result = await client.execute(False, request1)
         print(result)
 
         # inherited request
-        request = Read16CoilsRequest(32, slave)
-        result = await client.execute(False, request)
+        request2 = Read16CoilsRequest(32, slave)
+        result = await client.execute(False, request2)
         print(result)
 
 

examples/client_payload.py

@@ -12,7 +12,7 @@
 
 
 try:
-    import client_async
+    from examples import client_async
 except ImportError:
     print("*** ERROR --> THIS EXAMPLE needs the example directory, please see \n\
           https://pymodbus.readthedocs.io/en/latest/source/examples.html\n\
@@ -64,7 +64,7 @@ async def run_payload_calls(client):
         # Normally just do:  builder = BinaryPayloadBuilder()
         my_string = "abcdefgh"
         builder.add_string(my_string)
-        builder.add_bits([0, 1, 0, 1, 1, 0, 1, 0])
+        builder.add_bits([False, True, False, True, True, False, True, False])
         builder.add_8bit_int(-0x12)
         builder.add_8bit_uint(0x12)
         builder.add_16bit_int(-0x5678)

examples/client_sync.py

@@ -30,12 +30,14 @@
     python3 server_sync.py
 
 """
+from __future__ import annotations
+
 import logging
 import sys
 
 
 try:
-    import helper
+    from examples import helper
 except ImportError:
     print("*** ERROR --> THIS EXAMPLE needs the example directory, please see \n\
           https://pymodbus.readthedocs.io/en/latest/source/examples.html\n\
@@ -58,7 +60,7 @@ def setup_sync_client(description=None, cmdline=None):
         cmdline=cmdline,
     )
     _logger.info("### Create client object")
-    client = None
+    client: modbusClient.ModbusBaseSyncClient | None = None
     if args.comm == "tcp":
         client = modbusClient.ModbusTcpClient(
             args.host,

examples/helper.py

@@ -4,6 +4,8 @@
 code that are not relevant for the code as such, like e.g.
 get_command_line
 """
+from __future__ import annotations
+
 import argparse
 import logging
 import os
@@ -95,7 +97,7 @@ def get_commandline(server=False, description=None, extras=None, cmdline=None):
     args = parser.parse_args(cmdline)
 
     # set defaults
-    comm_defaults = {
+    comm_defaults: dict[str, list[int | str]] = {
         "tcp": ["socket", 5020],
         "udp": ["socket", 5020],
         "serial": ["rtu", "/dev/ptyp0"],

examples/message_parser.py

@@ -70,7 +70,7 @@ def decode(self, message):
         """Attempt to decode the supplied message."""
         value = message if self.encode else c.encode(message, "hex_codec")
         print("=" * 80)
-        print(f"Decoding Message {value}")
+        print(f"Decoding Message {value!r}")
         print("=" * 80)
         decoders = [
             self.framer(DecodePDU(True)),

examples/modbus_forwarder.py

@@ -16,13 +16,15 @@
 
 **WARNING** This example is a simple solution, that do only forward read requests.
 """
+from __future__ import annotations
+
 import asyncio
 import logging
 import sys
 
 
 try:
-    import helper
+    from examples import helper
 except ImportError:
     print("*** ERROR --> THIS EXAMPLE needs the example directory, please see \n\
           https://pymodbus.readthedocs.io/en/latest/source/examples.html\n\
@@ -58,6 +60,7 @@ async def run_forwarder(args):
     # in RemoteSlaveContext
     # For e.g to forward the requests to slave with slave address 1 use
     # store = RemoteSlaveContext(client, slave=1)
+    store: dict | RemoteSlaveContext
     if args.slaves:
         store = {}
         for i in args.slaves:

examples/package_test_tool.py

@@ -67,7 +67,7 @@ def __init__(
         self,
         params: CommParams,
         is_server: bool,
-        handler: Callable[[bytes], bytes],
+        handler: Callable[[ModbusProtocol, bool, bytes], None],
     ) -> None:
         """Initialize a stub instance."""
         self.stub_handle_data = handler
@@ -109,7 +109,7 @@ def __init__(self, comm: CommType):
         global test_port  # pylint: disable=global-statement
         self.comm = comm
         host = NULLMODEM_HOST
-
+        self.client: modbusClient.AsyncModbusTcpClient | modbusClient.AsyncModbusSerialClient
         if comm == CommType.TCP:
             self.client = modbusClient.AsyncModbusTcpClient(
                         host,
@@ -157,6 +157,7 @@ def __init__(self, comm: CommType):
         self.identity = ModbusDeviceIdentification(
             info_name={"VendorName": "VendorName"}
         )
+        self.server: modbusServer.ModbusTcpServer | modbusServer.ModbusSerialServer
         if comm == CommType.TCP:
             self.server = modbusServer.ModbusTcpServer(
                 self.context,
@@ -174,8 +175,9 @@ def __init__(self, comm: CommType):
         else:
             raise RuntimeError("ERROR: CommType not implemented")
         client_params = self.server.comm_params.copy()
-        client_params.host = client_params.source_address[0]
-        client_params.port = client_params.source_address[1]
+        if client_params.source_address:
+            client_params.host = client_params.source_address[0]
+            client_params.port = client_params.source_address[1]
         client_params.timeout_connect = 1.0
         self.stub = TransportStub(client_params, False, simulate_client)
         test_port += 1
@@ -194,6 +196,7 @@ async def run(self):
 
 async def main(comm: CommType, use_server: bool):
     """Combine setup and run."""
+    test: ServerTester | ClientTester
     if use_server:
         test = ServerTester(comm)
     else:

examples/server_async.py

@@ -35,10 +35,12 @@
 import asyncio
 import logging
 import sys
+from collections.abc import Callable
+from typing import Any
 
 
 try:
-    import helper
+    from examples import helper
 except ImportError:
     print("*** ERROR --> THIS EXAMPLE needs the example directory, please see \n\
           https://pymodbus.readthedocs.io/en/latest/source/examples.html\n\
@@ -70,7 +72,7 @@ def setup_server(description=None, context=None, cmdline=None):
     args = helper.get_commandline(server=True, description=description, cmdline=cmdline)
     if context:
         args.context = context
-    datablock = None
+    datablock: Callable[[], Any]
     if not args.context:
         _logger.info("### Create datastore")
         # The datastores only respond to the addresses that are initialized

examples/server_callback.py

@@ -10,7 +10,7 @@
 
 
 try:
-    import server_async
+    from examples import server_async
 except ImportError:
     print("*** ERROR --> THIS EXAMPLE needs the example directory, please see \n\
           https://pymodbus.readthedocs.io/en/latest/source/examples.html\n\
@@ -62,7 +62,7 @@ def validate(self, address, count=1):
 
 async def run_callback_server(cmdline=None):
     """Define datastore callback for server and do setup."""
-    queue = asyncio.Queue()
+    queue: asyncio.Queue = asyncio.Queue()
     block = CallbackDataBlock(queue, 0x00, [17] * 100)
     block.setValues(1, 15)
     store = ModbusSlaveContext(di=block, co=block, hr=block, ir=block)

examples/server_hook.py

@@ -4,6 +4,8 @@
 This is an example of using the builtin request/response tracer to
 manipulate the messages to/from the modbus server
 """
+from __future__ import annotations
+
 import asyncio
 import logging
 
@@ -20,7 +22,7 @@ class Manipulator:
     """A Class to run the server."""
 
     message_count: int = 1
-    server: ModbusTcpServer = None
+    server: ModbusTcpServer
 
     def server_request_tracer(self, request, *_addr):
         """Trace requests.

examples/server_payload.py

@@ -10,7 +10,7 @@
 
 
 try:
-    import server_async
+    from examples import server_async
 except ImportError:
     print("*** ERROR --> THIS EXAMPLE needs the example directory, please see \n\
           https://pymodbus.readthedocs.io/en/latest/source/examples.html\n\
@@ -37,7 +37,7 @@ def setup_payload_server(cmdline=None):
     # ----------------------------------------------------------------------- #
     builder = BinaryPayloadBuilder(byteorder=Endian.LITTLE, wordorder=Endian.LITTLE)
     builder.add_string("abcdefgh")
-    builder.add_bits([0, 1, 0, 1, 1, 0, 1, 0])
+    builder.add_bits([False, True, False, True, True, False, True, False])
     builder.add_8bit_int(-0x12)
     builder.add_8bit_uint(0x12)
     builder.add_16bit_int(-0x5678)

examples/server_sync.py

@@ -39,8 +39,7 @@
 
 
 try:
-    import helper
-    import server_async
+    from examples import helper, server_async
 except ImportError:
     print("*** ERROR --> THIS EXAMPLE needs the example directory, please see \n\
           https://pymodbus.readthedocs.io/en/latest/source/examples.html\n\

examples/server_updating.py

@@ -37,7 +37,7 @@
 
 
 try:
-    import server_async
+    from examples import server_async
 except ImportError:
     print("*** ERROR --> THIS EXAMPLE needs the example directory, please see \n\
           https://pymodbus.readthedocs.io/en/latest/source/examples.html\n\
@@ -101,8 +101,8 @@ def setup_updating_server(cmdline=None):
 
     # Continuing, use a sequential block without gaps.
     datablock = ModbusSequentialDataBlock(0x00, [17] * 100)
-    context = ModbusSlaveContext(di=datablock, co=datablock, hr=datablock, ir=datablock)
-    context = ModbusServerContext(slaves=context, single=True)
+    slavecontext = ModbusSlaveContext(di=datablock, co=datablock, hr=datablock, ir=datablock)
+    context = ModbusServerContext(slaves=slavecontext, single=True)
     return server_async.setup_server(
         description="Run asynchronous server.", context=context, cmdline=cmdline
     )

examples/simple_async_client.py

@@ -26,6 +26,7 @@ async def run_async_simple_client(comm, host, port, framer=FramerType.SOCKET):
     pymodbus_apply_logging_config("DEBUG")
 
     print("get client")
+    client: ModbusClient.ModbusBaseClient
     if comm == "tcp":
         client = ModbusClient.AsyncModbusTcpClient(
             host,

examples/simple_sync_client.py

@@ -28,6 +28,7 @@ def run_sync_simple_client(comm, host, port, framer=FramerType.SOCKET):
     pymodbus_apply_logging_config("DEBUG")
 
     print("get client")
+    client: ModbusClient.ModbusBaseSyncClient
     if comm == "tcp":
         client = ModbusClient.ModbusTcpClient(
             host,

pymodbus/client/__init__.py

@@ -6,13 +6,14 @@
     "AsyncModbusTlsClient",
     "AsyncModbusUdpClient",
     "ModbusBaseClient",
+    "ModbusBaseSyncClient",
     "ModbusSerialClient",
     "ModbusTcpClient",
     "ModbusTlsClient",
     "ModbusUdpClient",
 ]
 
-from pymodbus.client.base import ModbusBaseClient
+from pymodbus.client.base import ModbusBaseClient, ModbusBaseSyncClient
 from pymodbus.client.serial import AsyncModbusSerialClient, ModbusSerialClient
 from pymodbus.client.tcp import AsyncModbusTcpClient, ModbusTcpClient
 from pymodbus.client.tls import AsyncModbusTlsClient, ModbusTlsClient

pyproject.toml

@@ -186,6 +186,7 @@ overgeneral-exceptions = "builtins.Exception"
 bad-functions = "map,input"
 
 [tool.mypy]
+exclude = '/contrib/'
 strict_optional = true
 show_error_codes = true
 local_partial_types = true