Make mypy and black happy.

Author: tjhowse

.github/workflows/python-app.yml

@@ -39,12 +39,14 @@ jobs:
       run: |
         python -m pip install --upgrade pip
         python -m pip install .[test]
-    - name: Lint with flake8
+    - name: Linting
       run: |
         # stop the build if there are Python syntax errors or undefined names
         flake8 . --count --select=E9,F63,F7,F82 --show-source --statistics
         # exit-zero treats all errors as warnings. The GitHub editor is 127 chars wide
         flake8 . --count --exit-zero --max-complexity=10 --max-line-length=127 --statistics
+        black --check .
+        mypy .
     - name: Test with pytest
       run: |
         pytest --cov --cov-report=xml

modbus4mqtt/modbus4mqtt.py

@@ -67,11 +67,9 @@ def __init__(
         self.modbus_reconnect_sleep_interval = (
             5  # Wait this many seconds between modbus connection attempts
         )
-        self.modbus_connection_status: ModbusConnectionStatus = (
-            ModbusConnectionStatus.Offline
-        )
-        self._subscription_mids: map[int, str] = {}
-        self.mqtt_connection_status: MqttConnectionStatus = MqttConnectionStatus.Offline
+        self.modbus_connection_status: str = ModbusConnectionStatus.Offline
+        self._subscription_mids: dict[int, str] = {}
+        self.mqtt_connection_status: str = MqttConnectionStatus.Offline
         self.setup_modbus()
 
     def connect(self):
@@ -121,7 +119,7 @@ def connect_modbus(self):
         else:
             self.set_modbus_connection_status(ModbusConnectionStatus.Offline)
 
-    def set_modbus_connection_status(self, status: ModbusConnectionStatus):
+    def set_modbus_connection_status(self, status: str):
         if status == self.modbus_connection_status:
             return
         self.modbus_connection_status = status
@@ -293,7 +291,7 @@ def _on_subscribe(self, client, userdata, mid, reason_code_list, properties):
             logging.info("Subscribed to all set topics.")
             self._set_mqtt_connection_status(MqttConnectionStatus.Online)
 
-    def _set_mqtt_connection_status(self, status: MqttConnectionStatus):
+    def _set_mqtt_connection_status(self, status: str):
         if self.mqtt_connection_status == status:
             return
         self.mqtt_connection_status = status

modbus4mqtt/modbus_interface.py

@@ -5,7 +5,7 @@
 from pymodbus.framer import FramerType
 from pymodbus import ModbusException
 
-from SungrowModbusTcpClient import SungrowModbusTcpClient
+from SungrowModbusTcpClient import SungrowModbusTcpClient  # type: ignore
 from modbus4mqtt.modbus_table import ModbusTable
 
 DEFAULT_READ_BATCHING = 100
@@ -48,18 +48,26 @@ def __init__(
         self._unit: int = device_address
         self._variant: str | None = variant
         self._word_order: WordOrder = word_order
-        self._read_batching: int = read_batching
-        self._write_batching: int = write_batching
-        if not MIN_BATCHING <= self._read_batching <= MAX_BATCHING:
+        self._read_batching: int = (
+            read_batching if read_batching is not None else DEFAULT_READ_BATCHING
+        )
+        self._write_batching: int = (
+            write_batching if write_batching is not None else DEFAULT_WRITE_BATCHING
+        )
+        if not (MIN_BATCHING <= self._read_batching <= MAX_BATCHING):
             logging.warning(
                 f"Bad value for read_batching: {self._read_batching}. Enforcing limits of {MIN_BATCHING} to {MAX_BATCHING}."
             )
-            self._read_batching = max(MIN_BATCHING, min(MAX_BATCHING, self._read_batching))
-        if not MIN_BATCHING <= self._write_batching <= MAX_BATCHING:
+            self._read_batching = max(
+                MIN_BATCHING, min(MAX_BATCHING, self._read_batching)
+            )
+        if not (MIN_BATCHING <= self._write_batching <= MAX_BATCHING):
             logging.warning(
                 f"Bad value for write_batching: {self._write_batching}. Enforcing limits of {MIN_BATCHING} to {MAX_BATCHING}."
             )
-            self._write_batching = max(MIN_BATCHING, min(MAX_BATCHING, self._write_batching))
+            self._write_batching = max(
+                MIN_BATCHING, min(MAX_BATCHING, self._write_batching)
+            )
         if self._write_mode == WriteMode.Single and self._write_batching != 1:
             logging.warning("Overriding write batching to 1 due to single write mode.")
             self._write_batching = 1

modbus4mqtt/modbus_table.py

@@ -46,9 +46,9 @@ def _generate_batched_addresses(
         # of a range of addresses that can be read/written together.
         # If "write_mode" is true, the returned lists will only include
         # registers that've changed since the last read operation.
-        result = []
-        current_batch_start = None
-        current_batch_size = 0
+        result: list[tuple[int, int]] = []
+        current_batch_start: int = -1
+        current_batch_size: int = 0
         previous_addr = None
         if write_mode:
             max_batch_size = self._write_batch_size
@@ -60,17 +60,17 @@ def _generate_batched_addresses(
             if current_batch_size >= max_batch_size or (
                 previous_addr is not None and addr != previous_addr + 1
             ):
-                result.append([current_batch_start, current_batch_size])
+                result.append((current_batch_start, current_batch_size))
                 current_batch_start = addr
                 current_batch_size = 1
             else:
-                if current_batch_start is None:
+                if current_batch_start == -1:
                     current_batch_start = addr
                 current_batch_size += 1
             previous_addr = addr
         # Don't forget to add the last batch
-        if current_batch_start is not None:
-            result.append([current_batch_start, current_batch_size])
+        if current_batch_start != -1:
+            result.append((current_batch_start, current_batch_size))
         return result
 
     def clear_changed_registers(self):

tests/test_modbus_server.py

@@ -8,8 +8,11 @@
 import asyncio
 from paho.mqtt import client as mqtt_client
 
-from pymodbus import ModbusDeviceIdentification
+from pymodbus import (
+    ModbusDeviceIdentification,
+)
 from pymodbus.server import ModbusTcpServer
+from pymodbus.constants import ExcCodes
 from pymodbus.datastore import (
     ModbusDeviceContext,
     ModbusSequentialDataBlock,
@@ -47,7 +50,12 @@ async def set_holding_register(self, address: int, value: int) -> None:
 
     async def get_holding_register(self, address: int) -> int:
         """Get holding register value."""
-        return self.holding_registers.getValues(address + 1, count=1)[0]
+        result = self.holding_registers.getValues(address + 1, count=1)
+        if type(result) is ExcCodes:
+            raise ValueError(f"Error getting holding register at address {address}")
+        if type(result) is list and len(result) > 0:
+            return result[0]
+        raise ValueError(f"Error getting holding register at address {address}")
 
     async def run_async_server(self) -> None:
         """Run server."""

tests/test_modbustable.py

@@ -57,7 +57,7 @@ def test_generate_batched_addresses_simple():
         table.add_register(addr)
     batches = table.get_batched_addresses()
     # Should batch: [1,2] (start=1, len=2), [3] (start=3, len=1), [10,11] (start=10, len=2)
-    assert batches == [[1, 2], [3, 1], [10, 2]]
+    assert batches == [(1, 2), (3, 1), (10, 2)]
 
 
 def test_generate_batched_addresses_max_batch():
@@ -66,7 +66,7 @@ def test_generate_batched_addresses_max_batch():
         table.add_register(addr)
     batches = table.get_batched_addresses()
     # Should batch: [1,2,3,4] (start=1, len=4), [5] (start=5, len=1)
-    assert batches == [[1, 4], [5, 1]]
+    assert batches == [(1, 4), (5, 1)]
 
 
 def test_generate_batched_addresses_max_batch_write_mode():
@@ -80,4 +80,4 @@ def test_generate_batched_addresses_max_batch_write_mode():
         table.set_value(addr, 123, write=True)
     batches = table.get_batched_addresses(write_mode=True)
     # Should batch: [2] (start=2, len=1), [4] (start=4, len=1)
-    assert batches == [[2, 1], [4, 1]]
+    assert batches == [(2, 1), (4, 1)]

tests/test_mqtt.py

@@ -757,6 +757,7 @@ def assert_modbus_call(
             write_mode=modbus4mqtt.modbus_interface.WriteMode.Multi,
             variant=None,
             read_batching=None,
+            write_batching=None,
             word_order=word_order,
         )