Bump modbus-connection to 4.3.0 and adopt its new APIs

pyproject.toml

@@ -17,11 +17,11 @@ classifiers = [
 ]
 # Protocol + the device-modelling framework (modbus_connection.model). The caller
 # injects a ModbusUnit built from whichever backend they choose.
-dependencies = ["modbus-connection>=3.6"]
+dependencies = ["modbus-connection>=4.2"]
 
 [project.optional-dependencies]
 # Concrete backend for the script/query.py CLI (pip install trovis-modbus[cli]).
-cli = ["modbus-connection[pymodbus]>=3.6"]
+cli = ["modbus-connection[pymodbus]>=4.2"]
 
 [project.urls]
 Homepage = "https://github.com/Tom-Bom-badil/trovis-modbus"

script/query.py

@@ -17,7 +17,12 @@
 import sys
 import time
 
-from modbus_connection import ModbusConnection, ModbusError
+from modbus_connection import (
+    ModbusConnection,
+    ModbusError,
+    ModbusSerialParams,
+    ModbusTcpParams,
+)
 from modbus_connection.cli_helper import CountingUnit, print_component
 
 from trovis_modbus import Trovis557x
@@ -82,19 +87,24 @@ def _parse_args(argv: list[str] | None = None) -> argparse.Namespace:
     return parser.parse_args(argv)
 
 
-async def _open(args: argparse.Namespace) -> ModbusConnection:
+def _connection(args: argparse.Namespace) -> ModbusConnection:
+    """Build the connection described by the arguments. Performs no I/O."""
     # Imported here so the module loads (and --help works) without a backend.
-    from modbus_connection.pymodbus import connect_serial, connect_tcp
+    from modbus_connection.pymodbus import PymodbusConnection
 
     if args.transport == "serial":
-        return await connect_serial(
-            args.device,
-            baudrate=args.baudrate,
-            parity=args.parity,
-            stopbits=args.stopbits,
-            bytesize=args.bytesize,
+        return PymodbusConnection(
+            ModbusSerialParams(
+                device=args.device,
+                baudrate=args.baudrate,
+                parity=args.parity,
+                stopbits=args.stopbits,
+                bytesize=args.bytesize,
+            )
         )
-    return await connect_tcp(args.host, port=args.port, framer=args.framer)
+    return PymodbusConnection(
+        ModbusTcpParams(host=args.host, port=args.port, framer=args.framer)
+    )
 
 
 def _format_evidence(
@@ -148,8 +158,9 @@ def _print(device: Trovis557x) -> None:
 
 
 async def _run(args: argparse.Namespace) -> int:
+    connection = _connection(args)
     try:
-        connection = await _open(args)
+        await connection.connect()
     except ModbusError as err:
         print(f"Could not connect: {err}", file=sys.stderr)
         return 1

src/trovis_modbus/data_model.py

@@ -746,13 +746,7 @@ class TrovisComponent(Component):
 
     def _ensure_read_layout_is_configurable(self) -> None:
         """Reject availability changes after the read layout was built."""
-        cached_layout = {
-            "_read_items",
-            "register_items",
-            "bit_items",
-            "_register_blocks",
-            "_bit_blocks",
-        } & self.__dict__.keys()
+        cached_layout = {"_read_items"} & self.__dict__.keys()
         if self.__dict__.get("_plan") is not None:
             cached_layout.add("_plan")
         if cached_layout:
@@ -826,14 +820,7 @@ def _register_field_is_readable(
         if descriptor.scale_register is None:
             return True
 
-        scale_address = (
-            descriptor.scale_register
-            + descriptor.scale_register_stride * (self._index - 1)
-            + self._base_offset
-        )
-        if getattr(self, "scale_in_block", False):
-            scale_address += self._instance_offset
-        return is_span_readable(scale_address, 1, register_ranges)
+        return is_span_readable(self._scale_address(descriptor), 1, register_ranges)
 
     @property
     def readable_field_names(self) -> frozenset[str]:
@@ -846,9 +833,7 @@ def is_field_readable(self, field: str) -> bool:
 
     def metadata_for(self, field: str) -> DatapointMetadata | None:
         """Return neutral TROVIS metadata for a declared field."""
-        descriptor = type(self)._register_fields.get(field)
-        if descriptor is None:
-            descriptor = type(self)._bit_fields.get(field)
+        descriptor = type(self).declared_fields.get(field)
         if descriptor is None:
             return None
         return getattr(descriptor, "trovis_metadata", None)

tests/conftest.py

@@ -116,3 +116,13 @@ def trovis(mock_modbus_unit: MockModbusUnit) -> Trovis557x:
     mock_modbus_unit.holding.update(HOLDING)
     mock_modbus_unit.coils.update(COILS)
     return Trovis557x(mock_modbus_unit)
+
+
+@pytest.fixture
+def unit(mock_modbus_unit: MockModbusUnit) -> MockModbusUnit:
+    """The mock unit the ``trovis`` fixture reads and writes through.
+
+    Request it alongside ``trovis`` to assert on the register and coil stores a
+    write landed in, rather than reaching for the unit a component holds.
+    """
+    return mock_modbus_unit

tests/test_address_ranges_availability.py

@@ -33,10 +33,7 @@ def test_ranges_must_be_configured_before_read_layout(
     from trovis_modbus import Functions
 
     functions = Functions(mock_modbus_unit)
-    if hasattr(type(functions), "register_items"):
-        _ = functions.register_items
-    else:
-        _ = functions._read_items
+    _ = functions._read_items  # build and cache the read layout
 
     register_ranges, coil_ranges = ranges_for_model(5578)
     try:

tests/test_canonical_parity.py

@@ -115,7 +115,7 @@ def _fields() -> list[tuple[str, int, RegisterField | CoilField]]:
         label = type(component).__name__ + (
             f"[{component._index}]" if component._index != 1 else ""
         )
-        for field in {**component._register_fields, **component._bit_fields}.values():
+        for field in component.declared_fields.values():
             out.append((label, component._address(field), field))
     return out