From 497b8d4a5c69248b52ca038fafca663b0b4f6ba5 Mon Sep 17 00:00:00 2001 From: Paulus Schoutsen Date: Wed, 5 Aug 2026 01:53:35 -0400 Subject: [PATCH 1/7] Bump modbus-connection to 4.0.0a3 and adopt its new APIs 4.0.0 rebuilds connection management, but nothing in this library owns a connection, so the migration is small: every component declares absolute addresses at base_offset 0, so the breaking readable-range shift is a no-op, and all twelve components share one range map, so ComponentGroup's new range merge collapses to a single map. - script/query.py builds its connection from ModbusTcpParams / ModbusSerialParams instead of the connect_* factories, which 4.0 demotes. Construction performs no I/O; the script still connects eagerly so a connect failure stays distinguishable from a device that refuses a read. - Component.declared_fields (new in 4.0.0a2) replaces reaching into Component._register_fields / _bit_fields in metadata_for() and across the tests. The two metadata audits and the canonical-parity sweep now cover every declared field instead of only those the default model reads. - The read-layout guard watched four cache attributes that no longer exist in 4.0; only _read_items and _plan remain. - Cover the connection lifetime 4.0 guarantees: a dropped link heals on the next update over the same handles, and close() is permanent (ClientClosedError), both now modelled by the mock backend. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01WYw6yc9R1sgG2qybAp6p5j --- README.md | 2 +- pyproject.toml | 4 +- script/query.py | 46 +++++++----- src/trovis_modbus/data_model.py | 12 +--- tests/test_canonical_parity.py | 2 +- tests/test_device.py | 37 +++++++++- tests/test_metadata_ranges.py | 6 +- tests/test_operational_datapoints.py | 100 ++++++++++++++------------- tests/test_temporal_values.py | 12 ++-- 9 files changed, 130 insertions(+), 91 deletions(-) diff --git a/README.md b/README.md index 47e3d95..7db97b4 100755 --- a/README.md +++ b/README.md @@ -17,7 +17,7 @@ The library `trovis-modbus`: - provides the auto-discovery and auto-configuration logic required to determine the controller model and its capabilities, the configured hydronic system, the sensor and input assignments, and the relevant controller functions, parameters, and configuration settings. - contains the controller-specific data model. It knows the available registers and coils, their data types and metadata, and the rules required for safe reads and writes. -- does _NOT_ create or own the Modbus transport. Applications utilizing the library provide a [`modbus_connection.ModbusUnit`](https://github.com/home-assistant-libs/modbus-connection) and may use any backend supported by `modbus-connection` (pymodbus, tmodbus, ...). +- does _NOT_ create or own the Modbus transport. Applications utilizing the library provide a [`modbus_connection.ModbusUnit`](https://github.com/home-assistant-libs/modbus-connection) and may use any backend supported by `modbus-connection` (pymodbus, tmodbus, ...). Since `modbus-connection` 4.0 a connection manages its own link: it connects on demand and re-establishes itself after a drop, over the same unit handle. An application does not have to rebuild the `Trovis557x` object (nor, in Home Assistant, reload its config entry) when the link goes down. ## Data provided by the library diff --git a/pyproject.toml b/pyproject.toml index ba8dced..8c40e49 100755 --- a/pyproject.toml +++ b/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.0.0a3"] [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.0.0a3"] [project.urls] Homepage = "https://github.com/Tom-Bom-badil/trovis-modbus" diff --git a/script/query.py b/script/query.py index d90eada..478738a 100755 --- a/script/query.py +++ b/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 ModbusConnection as 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,14 +158,18 @@ def _print(device: Trovis557x) -> None: async def _run(args: argparse.Namespace) -> int: - try: - connection = await _open(args) - except ModbusError as err: - print(f"Could not connect: {err}", file=sys.stderr) - return 1 - + # Constructing the connection performs no I/O; a request would establish the + # link on its own. This one-shot dump connects eagerly anyway, to tell a + # connect failure apart from a device that answers but refuses a read. + connection = _connection(args) counting = CountingUnit(connection.for_unit(args.unit)) try: + try: + await connection.connect() + except ModbusError as err: + print(f"Could not connect: {err}", file=sys.stderr) + return 1 + probe = await Trovis557x.async_probe(counting) device = Trovis557x( counting, diff --git a/src/trovis_modbus/data_model.py b/src/trovis_modbus/data_model.py index c9675c6..00418ee 100755 --- a/src/trovis_modbus/data_model.py +++ b/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: @@ -846,9 +840,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) diff --git a/tests/test_canonical_parity.py b/tests/test_canonical_parity.py index 04ac268..f7d5c08 100755 --- a/tests/test_canonical_parity.py +++ b/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 diff --git a/tests/test_device.py b/tests/test_device.py index a69fff0..ee90bd8 100755 --- a/tests/test_device.py +++ b/tests/test_device.py @@ -5,6 +5,7 @@ from datetime import date, time import pytest +from modbus_connection import ClientClosedError from modbus_connection.mock import MockModbusConnection, MockModbusUnit from trovis_modbus import ( @@ -166,9 +167,7 @@ async def test_full_update_consolidates_reads() -> None: unit = _CountingUnit(inner) device = Trovis557x(unit) # type: ignore[arg-type] - field_count = sum( - len(c._register_fields) + len(c._bit_fields) for c in device.components - ) + field_count = sum(len(c.readable_field_names) for c in device.components) await device.async_update() # Many fields collapse into a small number of range-aware block reads — far @@ -218,6 +217,38 @@ async def test_consolidated_reads_decode_correctly() -> None: assert device.sensors.vf3 == pytest.approx(32.0) +async def test_update_survives_a_dropped_connection() -> None: + """A dropped link heals on the next update — the device is not rebuilt.""" + connection = MockModbusConnection() + unit = connection.for_unit(1) + unit.holding.update(HOLDING) + unit.coils.update(COILS) + device = Trovis557x(unit) + await device.async_update() + assert device.rk1.flow_setpoint == pytest.approx(55.0) + + lost: list[int] = [] + connection.on_connection_lost(lambda: lost.append(1)) + connection.simulate_connection_lost() + assert lost == [1] + + # The same device and unit handles keep working; the request reconnects. + await device.async_update() + assert device.rk1.flow_setpoint == pytest.approx(55.0) + + +async def test_update_after_close_raises() -> None: + """Closing is the owner's permanent end of the connection, not a drop.""" + connection = MockModbusConnection() + unit = connection.for_unit(1) + unit.holding.update(HOLDING) + device = Trovis557x(unit) + await connection.close() + + with pytest.raises(ClientClosedError): + await device.async_update() + + async def test_update_listener(trovis: Trovis557x) -> None: calls: list[int] = [] unsubscribe = trovis.rk4.add_update_listener(lambda: calls.append(1)) diff --git a/tests/test_metadata_ranges.py b/tests/test_metadata_ranges.py index 7a6c457..b2ea0a8 100755 --- a/tests/test_metadata_ranges.py +++ b/tests/test_metadata_ranges.py @@ -57,7 +57,7 @@ def test_corrected_domestic_hot_water_setpoint_limits(trovis: Trovis557x) -> Non def test_outdoor_temperature_delay_uses_hardware_verified_scale( trovis: Trovis557x, ) -> None: - field = trovis.controller._register_fields["outdoor_temperature_delay"] + field = trovis.controller.declared_fields["outdoor_temperature_delay"] assert field.decode([30]) == pytest.approx(3.0) assert field.encode(3) == [30] @@ -65,7 +65,7 @@ def test_outdoor_temperature_delay_uses_hardware_verified_scale( def test_all_writable_numbers_have_complete_ranges(trovis: Trovis557x) -> None: for component in trovis.components: - for field in component._register_fields: + for field in component.declared_fields: metadata = component.require_metadata_for(field) if metadata.value_kind != "number" or not metadata.writable: continue @@ -88,7 +88,7 @@ def test_all_writable_temporal_values_have_complete_ranges( } for component in trovis.components: - for field in component._register_fields: + for field in component.declared_fields: metadata = component.require_metadata_for(field) expected_type = expected_types.get(metadata.value_kind) if expected_type is None or not metadata.writable: diff --git a/tests/test_operational_datapoints.py b/tests/test_operational_datapoints.py index 8ed8ea0..d0dd4f1 100755 --- a/tests/test_operational_datapoints.py +++ b/tests/test_operational_datapoints.py @@ -25,25 +25,25 @@ def test_heating_circuit_stride_patterns() -> None: rk2 = device.rk2 rk3 = device.rk3 - assert rk1._address(rk1._bit_fields["valve_closing"]) == 61 - assert rk2._address(rk2._bit_fields["valve_closing"]) == 63 - assert rk3._address(rk3._bit_fields["valve_closing"]) == 65 + assert rk1._address(rk1.declared_fields["valve_closing"]) == 61 + assert rk2._address(rk2.declared_fields["valve_closing"]) == 63 + assert rk3._address(rk3.declared_fields["valve_closing"]) == 65 - assert rk1._address(rk1._register_fields["fixed_setpoint_day"]) == 1041 - assert rk2._address(rk2._register_fields["fixed_setpoint_day"]) == 1241 - assert rk3._address(rk3._register_fields["fixed_setpoint_day"]) == 1441 + assert rk1._address(rk1.declared_fields["fixed_setpoint_day"]) == 1041 + assert rk2._address(rk2.declared_fields["fixed_setpoint_day"]) == 1241 + assert rk3._address(rk3.declared_fields["fixed_setpoint_day"]) == 1441 - assert rk1._address(rk1._bit_fields["room_setpoint_control_autonomous"]) == 121 - assert rk2._address(rk2._bit_fields["room_setpoint_control_autonomous"]) == 122 - assert rk3._address(rk3._bit_fields["room_setpoint_control_autonomous"]) == 123 + assert rk1._address(rk1.declared_fields["room_setpoint_control_autonomous"]) == 121 + assert rk2._address(rk2.declared_fields["room_setpoint_control_autonomous"]) == 122 + assert rk3._address(rk3.declared_fields["room_setpoint_control_autonomous"]) == 123 def test_domestic_hot_water_special_setpoint_is_distinct_from_active_setpoint() -> None: device = Trovis557x(unit=None) # type: ignore[arg-type] rk4 = device.rk4 - assert rk4._address(rk4._register_fields["setpoint_active"]) == 1807 - assert rk4._address(rk4._register_fields["special_setpoint"]) == 1808 + assert rk4._address(rk4.declared_fields["setpoint_active"]) == 1807 + assert rk4._address(rk4.declared_fields["special_setpoint"]) == 1808 assert rk4.ebene_coils["special_setpoint"] == (112, 0) @@ -71,11 +71,11 @@ def test_legacy_gap_registers_and_intermediate_heating_points() -> None: controller = device.controller rk4 = device.rk4 - assert controller._address(controller._register_fields["special_functions"]) == 4 + assert controller._address(controller.declared_fields["special_functions"]) == 4 overrun = rk4.require_metadata_for("storage_tank_charging_pump_lag_factor") assert ( - rk4._address(rk4._register_fields["storage_tank_charging_pump_lag_factor"]) + rk4._address(rk4.declared_fields["storage_tank_charging_pump_lag_factor"]) == 1804 ) assert overrun.writable is True @@ -84,28 +84,31 @@ def test_legacy_gap_registers_and_intermediate_heating_points() -> None: assert overrun.number.max_value == pytest.approx(10.0) assert overrun.number.step == pytest.approx(0.1) - assert rk4._address(rk4._bit_fields["intermediate_heating_function_enabled"]) == 406 - assert rk4._address(rk4._bit_fields["intermediate_heating_operation"]) == 1830 + assert ( + rk4._address(rk4.declared_fields["intermediate_heating_function_enabled"]) + == 406 + ) + assert rk4._address(rk4.declared_fields["intermediate_heating_operation"]) == 1830 def test_additional_5578_sensor_addresses() -> None: sensors = Sensors(unit=None) # type: ignore[arg-type] - assert sensors._address(sensors._register_fields["af2"]) == 10 - assert sensors._address(sensors._register_fields["sf3"]) == 24 - assert sensors._address(sensors._register_fields["ae1"]) == 25 - assert sensors._address(sensors._register_fields["fg1"]) == 25 - assert sensors._address(sensors._register_fields["ae2"]) == 26 - assert sensors._address(sensors._register_fields["fg2"]) == 26 - assert sensors._address(sensors._register_fields["ae3"]) == 27 - assert sensors._address(sensors._register_fields["fg3"]) == 27 - assert sensors._address(sensors._register_fields["pulse_rate"]) == 28 - assert sensors._address(sensors._register_fields["analog_input_voltage"]) == 41 - assert sensors._address(sensors._register_fields["analog_input_current"]) == 41 - - assert "ae1_fg1" not in sensors._register_fields - assert "ae2_fg2" not in sensors._register_fields - assert "ae3_fg3" not in sensors._register_fields + assert sensors._address(sensors.declared_fields["af2"]) == 10 + assert sensors._address(sensors.declared_fields["sf3"]) == 24 + assert sensors._address(sensors.declared_fields["ae1"]) == 25 + assert sensors._address(sensors.declared_fields["fg1"]) == 25 + assert sensors._address(sensors.declared_fields["ae2"]) == 26 + assert sensors._address(sensors.declared_fields["fg2"]) == 26 + assert sensors._address(sensors.declared_fields["ae3"]) == 27 + assert sensors._address(sensors.declared_fields["fg3"]) == 27 + assert sensors._address(sensors.declared_fields["pulse_rate"]) == 28 + assert sensors._address(sensors.declared_fields["analog_input_voltage"]) == 41 + assert sensors._address(sensors.declared_fields["analog_input_current"]) == 41 + + assert "ae1_fg1" not in sensors.declared_fields + assert "ae2_fg2" not in sensors.declared_fields + assert "ae3_fg3" not in sensors.declared_fields for field in ("ae1", "fg1", "ae2", "fg2", "ae3", "fg3"): metadata = sensors.require_metadata_for(field) @@ -144,11 +147,11 @@ def test_controller_monitoring_metadata_and_timeout() -> None: timeout = controller.require_metadata_for("glt_timeout_active") assert timeout.writable is True - assert controller._address(controller._bit_fields["glt_timeout_active"]) == 158 + assert controller._address(controller.declared_fields["glt_timeout_active"]) == 158 assert ( controller._address( - controller._register_fields["summer_outdoor_temperature_average"] + controller.declared_fields["summer_outdoor_temperature_average"] ) == 42 ) @@ -158,15 +161,14 @@ def test_solar_circuit_uses_dedicated_register_and_coil_block() -> None: solar = SolarCircuit(unit=None) # type: ignore[arg-type] assert ( - solar._address(solar._register_fields["pump_on_temperature_difference"]) == 1809 + solar._address(solar.declared_fields["pump_on_temperature_difference"]) == 1809 ) assert ( - solar._address(solar._register_fields["pump_off_temperature_difference"]) - == 1810 + solar._address(solar.declared_fields["pump_off_temperature_difference"]) == 1810 ) - assert solar._address(solar._register_fields["maximum_storage_temperature"]) == 1811 - assert solar._address(solar._register_fields["operating_hours"]) == 1812 - assert solar._address(solar._bit_fields["pump_running"]) == 1807 + assert solar._address(solar.declared_fields["maximum_storage_temperature"]) == 1811 + assert solar._address(solar.declared_fields["operating_hours"]) == 1812 + assert solar._address(solar.declared_fields["pump_running"]) == 1807 pump_on = solar.require_metadata_for("pump_on_temperature_difference") assert pump_on.writable is True @@ -192,10 +194,10 @@ def test_solar_circuit_uses_dedicated_register_and_coil_block() -> None: def test_solar_datapoints_are_no_longer_owned_by_rk4() -> None: device = Trovis557x(unit=None) # type: ignore[arg-type] - assert "solar_operating_hours" not in device.rk4._register_fields - assert "solar_circuit_pump_running" not in device.rk4._bit_fields - assert "operating_hours" in device.solar._register_fields - assert "pump_running" in device.solar._bit_fields + assert "solar_operating_hours" not in device.rk4.declared_fields + assert "solar_circuit_pump_running" not in device.rk4.declared_fields + assert "operating_hours" in device.solar.declared_fields + assert "pump_running" in device.solar.declared_fields def test_buffer_tank_status_enum_matches_firmware_values() -> None: @@ -206,22 +208,22 @@ def test_buffer_tank_circuit_uses_rk1_extension_registers() -> None: buffer_tank = BufferTankCircuit(unit=None) # type: ignore[arg-type] assert ( - buffer_tank._address(buffer_tank._register_fields["minimum_charging_setpoint"]) + buffer_tank._address(buffer_tank.declared_fields["minimum_charging_setpoint"]) == 1099 ) assert ( - buffer_tank._address(buffer_tank._register_fields["charging_end_temperature"]) + buffer_tank._address(buffer_tank.declared_fields["charging_end_temperature"]) == 1100 ) assert ( - buffer_tank._address(buffer_tank._register_fields["charging_temperature_boost"]) + buffer_tank._address(buffer_tank.declared_fields["charging_temperature_boost"]) == 1101 ) assert ( - buffer_tank._address(buffer_tank._register_fields["charging_pump_lag_factor"]) + buffer_tank._address(buffer_tank.declared_fields["charging_pump_lag_factor"]) == 1102 ) - assert buffer_tank._address(buffer_tank._register_fields["status"]) == 1103 + assert buffer_tank._address(buffer_tank.declared_fields["status"]) == 1103 minimum = buffer_tank.require_metadata_for("minimum_charging_setpoint") assert minimum.writable is True @@ -254,6 +256,6 @@ def test_buffer_tank_circuit_uses_rk1_extension_registers() -> None: def test_buffer_tank_extension_does_not_duplicate_rk1_fields() -> None: device = Trovis557x(unit=None) # type: ignore[arg-type] - assert set(device.buffer_tank._register_fields).isdisjoint( - device.rk1._register_fields + assert set(device.buffer_tank.declared_fields).isdisjoint( + device.rk1.declared_fields ) diff --git a/tests/test_temporal_values.py b/tests/test_temporal_values.py index 52377a9..c238c66 100755 --- a/tests/test_temporal_values.py +++ b/tests/test_temporal_values.py @@ -50,17 +50,17 @@ def test_temporal_metadata(trovis: Trovis557x) -> None: def test_temporal_codecs(trovis: Trovis557x) -> None: - time_field = trovis.clock._register_fields["time"] + time_field = trovis.clock.declared_fields["time"] assert time_field.decode([1430]) == time(14, 30) assert time_field.decode([2360]) is None assert time_field.encode(time(8, 5)) == [805] - date_field = trovis.clock._register_fields["date"] + date_field = trovis.clock.declared_fields["date"] assert date_field.decode([2106, 2026]) == date(2026, 6, 21) assert date_field.decode([3102, 2025]) is None assert date_field.encode(date(2027, 1, 2)) == [201, 2027] - month_day_field = trovis.controller._register_fields["summer_start"] + month_day_field = trovis.controller.declared_fields["summer_start"] assert month_day_field.decode([1505]) == MonthDay(15, 5) assert month_day_field.decode([3102]) is None assert month_day_field.encode(MonthDay(29, 2)) == [2902] @@ -109,7 +109,7 @@ def test_time_rejects_subminute_precision( trovis: Trovis557x, value: time, ) -> None: - field = trovis.clock._register_fields["time"] + field = trovis.clock.declared_fields["time"] with pytest.raises(TrovisValueValidationError): field.encode(value) @@ -117,13 +117,13 @@ def test_time_rejects_subminute_precision( def test_disinfection_time_rejects_out_of_range_value( trovis: Trovis557x, ) -> None: - field = trovis.rk4._register_fields["disinfection_start"] + field = trovis.rk4.declared_fields["disinfection_start"] assert field.decode([2350]) is None with pytest.raises(TrovisValueValidationError): field.encode(time(23, 50)) def test_date_rejects_unsupported_year(trovis: Trovis557x) -> None: - field = trovis.clock._register_fields["date"] + field = trovis.clock.declared_fields["date"] with pytest.raises(TrovisValueValidationError): field.encode(date(2099, 1, 1)) From f155efeb18b1fc865d1309f3d8dcf489afcba1b0 Mon Sep 17 00:00:00 2001 From: Paulus Schoutsen Date: Wed, 5 Aug 2026 03:51:43 -0400 Subject: [PATCH 2/7] Address review: simpler connect, no README change - Move connect() out of the read try/except in query.py and drop the explanatory comment. A failed connect returns before there is anything to close, so the read path keeps its own try/finally. - Revert the README paragraph on connection lifetime. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01WYw6yc9R1sgG2qybAp6p5j --- README.md | 2 +- script/query.py | 15 ++++++--------- 2 files changed, 7 insertions(+), 10 deletions(-) diff --git a/README.md b/README.md index 7db97b4..47e3d95 100755 --- a/README.md +++ b/README.md @@ -17,7 +17,7 @@ The library `trovis-modbus`: - provides the auto-discovery and auto-configuration logic required to determine the controller model and its capabilities, the configured hydronic system, the sensor and input assignments, and the relevant controller functions, parameters, and configuration settings. - contains the controller-specific data model. It knows the available registers and coils, their data types and metadata, and the rules required for safe reads and writes. -- does _NOT_ create or own the Modbus transport. Applications utilizing the library provide a [`modbus_connection.ModbusUnit`](https://github.com/home-assistant-libs/modbus-connection) and may use any backend supported by `modbus-connection` (pymodbus, tmodbus, ...). Since `modbus-connection` 4.0 a connection manages its own link: it connects on demand and re-establishes itself after a drop, over the same unit handle. An application does not have to rebuild the `Trovis557x` object (nor, in Home Assistant, reload its config entry) when the link goes down. +- does _NOT_ create or own the Modbus transport. Applications utilizing the library provide a [`modbus_connection.ModbusUnit`](https://github.com/home-assistant-libs/modbus-connection) and may use any backend supported by `modbus-connection` (pymodbus, tmodbus, ...). ## Data provided by the library diff --git a/script/query.py b/script/query.py index 478738a..ebf79d1 100755 --- a/script/query.py +++ b/script/query.py @@ -158,18 +158,15 @@ def _print(device: Trovis557x) -> None: async def _run(args: argparse.Namespace) -> int: - # Constructing the connection performs no I/O; a request would establish the - # link on its own. This one-shot dump connects eagerly anyway, to tell a - # connect failure apart from a device that answers but refuses a read. connection = _connection(args) - counting = CountingUnit(connection.for_unit(args.unit)) try: - try: - await connection.connect() - except ModbusError as err: - print(f"Could not connect: {err}", file=sys.stderr) - return 1 + await connection.connect() + except ModbusError as err: + print(f"Could not connect: {err}", file=sys.stderr) + return 1 + counting = CountingUnit(connection.for_unit(args.unit)) + try: probe = await Trovis557x.async_probe(counting) device = Trovis557x( counting, From 9af511f6a3611d36608791ebe7feefe120099440 Mon Sep 17 00:00:00 2001 From: Paulus Schoutsen Date: Thu, 6 Aug 2026 03:55:36 -0400 Subject: [PATCH 3/7] Stop reaching into modbus-connection internals where it is avoidable Follow-up cleanup on the 4.0.0a3 bump, all of it possible with the library as it stands today: - _register_field_is_readable reimplemented Component._scale_address verbatim, down to the scale_in_block branch. Call the library's helper so the two cannot drift apart. - test_ranges_must_be_configured_before_read_layout still branched on hasattr(type(functions), "register_items"). That attribute went away in 4.0, so the branch was dead. - Eight tests reached through a component for the unit to assert on (trovis.rk1._unit). The trovis fixture is built from mock_modbus_unit, so a `unit` fixture hands them the same object by a public route. - test_full_update_never_reads_across_an_unreadable_gap wrapped the unit to capture block spans. async_read_raw() reports every address a pooled read touched, which is what the test actually asserts on. - test_consolidated_reads_decode_correctly wrapped the unit without ever reading the counts. _CountingUnit stays for test_full_update_consolidates_reads, which needs block boundaries the library does not expose. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01WYw6yc9R1sgG2qybAp6p5j --- src/trovis_modbus/data_model.py | 9 +--- tests/conftest.py | 10 ++++ tests/test_address_ranges_availability.py | 5 +- tests/test_device.py | 60 +++++++++++------------ tests/test_operating_mode_write.py | 8 +-- tests/test_temporal_values.py | 9 ++-- 6 files changed, 49 insertions(+), 52 deletions(-) diff --git a/src/trovis_modbus/data_model.py b/src/trovis_modbus/data_model.py index 00418ee..5735f97 100755 --- a/src/trovis_modbus/data_model.py +++ b/src/trovis_modbus/data_model.py @@ -820,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]: diff --git a/tests/conftest.py b/tests/conftest.py index 1befae2..23d91db 100755 --- a/tests/conftest.py +++ b/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 diff --git a/tests/test_address_ranges_availability.py b/tests/test_address_ranges_availability.py index d89348d..ccaa837 100755 --- a/tests/test_address_ranges_availability.py +++ b/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: diff --git a/tests/test_device.py b/tests/test_device.py index ee90bd8..f5a383c 100755 --- a/tests/test_device.py +++ b/tests/test_device.py @@ -181,34 +181,33 @@ async def test_full_update_consolidates_reads() -> None: assert all(count <= 50 for _, count in unit.coil_blocks) -async def test_full_update_never_reads_across_an_unreadable_gap() -> None: +async def test_full_update_never_reads_across_an_unreadable_gap( + unit: MockModbusUnit, +) -> None: """Every block stays inside the controller's readable ranges (no NAK risk).""" - inner = MockModbusConnection().for_unit(1) - unit = _CountingUnit(inner) - device = Trovis557x(unit) # type: ignore[arg-type] - await device.async_update() - - def readable(address: int) -> bool: - return any(low <= address <= high for low, high in REGISTER_RANGES) - - for start, count in unit.register_blocks: - assert all(readable(start + i) for i in range(count)), ( - f"block {start}..{start + count - 1} crosses an unreadable gap" - ) + device = Trovis557x(unit) + # async_read_raw reports every address the pooled read touched, so a block + # spanning a gap shows up as a read of an address no range declares. + raw = await device._group.async_read_raw() + read = set(raw["holding"]) + + unreadable = { + address + for address in read + if not any(low <= address <= high for low, high in REGISTER_RANGES) + } + assert not unreadable, f"read addresses outside every readable range: {unreadable}" # Addresses 7-8 (between ranges [0,6] and [9,40]) are never read — the low # registers split there instead of being merged into one 0..26 block. - read = {start + i for start, count in unit.register_blocks for i in range(count)} assert 7 not in read and 8 not in read -async def test_consolidated_reads_decode_correctly() -> None: +async def test_consolidated_reads_decode_correctly(unit: MockModbusUnit) -> None: """Adjacent physical sensor registers decode to the right sensor inputs.""" - inner = MockModbusConnection().for_unit(1) # Flow sensors VF1/VF2/VF3 are at adjacent addresses 12/13/14 — fetched in # one block, then decoded on the central Sensors component. - inner.holding.update({12: 300, 13: 310, 14: 320}) - unit = _CountingUnit(inner) - device = Trovis557x(unit) # type: ignore[arg-type] + unit.holding.update({12: 300, 13: 310, 14: 320}) + device = Trovis557x(unit) await device.async_update() @@ -276,10 +275,10 @@ async def test_write_rejects_readonly(trovis: Trovis557x) -> None: await trovis.rk1.write("flow_setpoint", 50.0) -async def test_mode_write_releases_override_coil(trovis: Trovis557x) -> None: +async def test_mode_write_releases_override_coil( + trovis: Trovis557x, unit: MockModbusUnit +) -> None: """Setting the mode first releases the Ebene coil (0 = remote control).""" - unit = trovis.rk1._unit - await trovis.async_enable_writing() await unit.write_coil(88, True) # start "autonomous" (controller-controlled) await trovis.rk1.set_mode(OperatingMode.DAY) @@ -289,18 +288,19 @@ async def test_mode_write_releases_override_coil(trovis: Trovis557x) -> None: assert (await unit.read_holding_registers(105, 1))[0] == int(OperatingMode.DAY) -async def test_circuit2_mode_uses_strided_override(trovis: Trovis557x) -> None: +async def test_circuit2_mode_uses_strided_override( + trovis: Trovis557x, unit: MockModbusUnit +) -> None: """Circuit 2's override coil follows the +2 stride (90, not 88).""" - unit = trovis.rk2._unit await unit.write_coil(90, True) await trovis.rk2.set_mode(OperatingMode.NIGHT) assert (await unit.read_coils(90, 1))[0] is False assert (await unit.read_holding_registers(107, 1))[0] == int(OperatingMode.NIGHT) -async def test_write_access_enable_disable(trovis: Trovis557x) -> None: - unit = trovis.controller._unit - +async def test_write_access_enable_disable( + trovis: Trovis557x, unit: MockModbusUnit +) -> None: assert await trovis.async_read_writing_enabled() is False await trovis.async_enable_writing() @@ -312,10 +312,10 @@ async def test_write_access_enable_disable(trovis: Trovis557x) -> None: assert (await unit.read_holding_registers(144, 1))[0] == 0 -async def test_write_refreshes_access_code(trovis: Trovis557x) -> None: +async def test_write_refreshes_access_code( + trovis: Trovis557x, unit: MockModbusUnit +) -> None: """The public write path refreshes HR40145 before writing the datapoint.""" - unit = trovis.controller._unit - assert await trovis.async_read_writing_enabled() is False await trovis.rk1.set_room_setpoint_day(21.5) diff --git a/tests/test_operating_mode_write.py b/tests/test_operating_mode_write.py index 1b0c397..c75bf0e 100755 --- a/tests/test_operating_mode_write.py +++ b/tests/test_operating_mode_write.py @@ -2,6 +2,8 @@ from __future__ import annotations +from modbus_connection.mock import MockModbusUnit + from trovis_modbus import OperatingMode, Trovis557x from trovis_modbus.enums import OPERATING_MODE_OPTIONS @@ -18,10 +20,9 @@ def test_circuits_share_operating_mode_metadata(trovis: Trovis557x) -> None: async def test_heating_automatic_restores_autark_without_register_write( - trovis: Trovis557x, + trovis: Trovis557x, unit: MockModbusUnit ) -> None: """Automatic mode is selected through the Ebene coil, not the mode HR.""" - unit = trovis.rk1._unit await unit.write_register(105, int(OperatingMode.DAY)) await unit.write_coil(88, False) @@ -32,10 +33,9 @@ async def test_heating_automatic_restores_autark_without_register_write( async def test_domestic_hot_water_automatic_uses_its_own_ebene_coil( - trovis: Trovis557x, + trovis: Trovis557x, unit: MockModbusUnit ) -> None: """Domestic hot water restores AUTARK through CL95 and leaves HR40112 intact.""" - unit = trovis.rk4._unit await unit.write_register(111, int(OperatingMode.NIGHT)) await unit.write_coil(94, False) diff --git a/tests/test_temporal_values.py b/tests/test_temporal_values.py index c238c66..111a1f0 100755 --- a/tests/test_temporal_values.py +++ b/tests/test_temporal_values.py @@ -5,6 +5,7 @@ from datetime import date, datetime, time import pytest +from modbus_connection.mock import MockModbusUnit from trovis_modbus import MonthDay, Trovis557x, TrovisValueValidationError @@ -66,9 +67,7 @@ def test_temporal_codecs(trovis: Trovis557x) -> None: assert month_day_field.encode(MonthDay(29, 2)) == [2902] -async def test_clock_native_writes(trovis: Trovis557x) -> None: - unit = trovis.clock._unit - +async def test_clock_native_writes(trovis: Trovis557x, unit: MockModbusUnit) -> None: await trovis.clock.set_time(time(8, 15)) assert (await unit.read_holding_registers(99, 1))[0] == 815 @@ -88,10 +87,8 @@ async def test_datetime_write(trovis: Trovis557x) -> None: async def test_recurring_date_and_disinfection_writes( - trovis: Trovis557x, + trovis: Trovis557x, unit: MockModbusUnit ) -> None: - unit = trovis.controller._unit - await trovis.controller.set_summer_start(MonthDay(1, 4)) assert (await unit.read_holding_registers(112, 1))[0] == 104 From a9b9263e5be9d48287c2b7cdc5d1de9b2b4c92e5 Mon Sep 17 00:00:00 2001 From: Paulus Schoutsen Date: Thu, 6 Aug 2026 07:30:17 -0400 Subject: [PATCH 4/7] Move to modbus-connection 4.0.0a4 MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit a4 is a small release for this library. Its headline items — sunspec.boolean, float-typed repeating_group counts, the scan() lookup helper — are SunSpec, which TROVIS is not, and the newly exported field classes (NumberField, FloatField, RawField, StringField) are ones nothing here constructs. The model-layer changes are type widening and exports, with no behaviour change. What does reach us: - The default request timeout rose from 3s to 10s, which script/query.py inherits. Better suited to an RTU-over-TCP gateway than the old 3s. - Import PymodbusConnection under its own name rather than aliasing the backend's ModbusConnection, now that the backend names are canonical. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01WYw6yc9R1sgG2qybAp6p5j --- pyproject.toml | 4 ++-- script/query.py | 2 +- 2 files changed, 3 insertions(+), 3 deletions(-) diff --git a/pyproject.toml b/pyproject.toml index 8c40e49..ecb3383 100755 --- a/pyproject.toml +++ b/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>=4.0.0a3"] +dependencies = ["modbus-connection>=4.0.0a4"] [project.optional-dependencies] # Concrete backend for the script/query.py CLI (pip install trovis-modbus[cli]). -cli = ["modbus-connection[pymodbus]>=4.0.0a3"] +cli = ["modbus-connection[pymodbus]>=4.0.0a4"] [project.urls] Homepage = "https://github.com/Tom-Bom-badil/trovis-modbus" diff --git a/script/query.py b/script/query.py index ebf79d1..7a9bd2b 100755 --- a/script/query.py +++ b/script/query.py @@ -90,7 +90,7 @@ def _parse_args(argv: list[str] | None = None) -> argparse.Namespace: 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 ModbusConnection as PymodbusConnection + from modbus_connection.pymodbus import PymodbusConnection if args.transport == "serial": return PymodbusConnection( From 278b15fd9f2854649a25406449fb54694bcf0ad4 Mon Sep 17 00:00:00 2001 From: Paulus Schoutsen Date: Thu, 6 Aug 2026 20:10:30 -0400 Subject: [PATCH 5/7] Require the stable modbus-connection 4.0.0 The pin drops the prerelease suffix and goes back to the repo's usual >=MAJOR.MINOR form, which also stops it opting in to prereleases. Nothing new to adopt since 4.0.0a4: notify=False on Component.async_update and the _verify_read hook it enables are for callers that fire their own listeners or check a model header after a read, and this library does neither. NumberField's per-field warning dedup applies to the enum() fields here with no change on our side. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01WYw6yc9R1sgG2qybAp6p5j --- pyproject.toml | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/pyproject.toml b/pyproject.toml index ecb3383..8c3cf0f 100755 --- a/pyproject.toml +++ b/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>=4.0.0a4"] +dependencies = ["modbus-connection>=4.0"] [project.optional-dependencies] # Concrete backend for the script/query.py CLI (pip install trovis-modbus[cli]). -cli = ["modbus-connection[pymodbus]>=4.0.0a4"] +cli = ["modbus-connection[pymodbus]>=4.0"] [project.urls] Homepage = "https://github.com/Tom-Bom-badil/trovis-modbus" From b55f7513bd2dadfe6e1e511e1cf4f05ac692b604 Mon Sep 17 00:00:00 2001 From: Paulus Schoutsen Date: Fri, 7 Aug 2026 03:43:41 -0400 Subject: [PATCH 6/7] Bump modbus-connection to 4.1 and drop the hand-rolled read counter MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit MockModbusUnit now logs every dispatched read as a ReadEvent(register_type, address, count), which is what _CountingUnit existed to do. The release notes name this case: several libraries built on modbus-connection carried a near-identical wrapper. Removing it makes the two tests that used it stronger, not just shorter: - The unreadable-gap test went back to asserting per block rather than per address, and no longer reaches through Trovis557x._group for async_read_raw. It also covers coil blocks now, which it never did — 18 of the 35 blocks a full update issues are coils, and COIL_RANGES has gaps of its own. - Both tests take the trovis fixture instead of rebuilding a device with the same stores, and neither needs a type: ignore for a unit that is not a ModbusUnit. 4.1.0's other additions do not apply: TROVIS models every boolean as a real coil, so the new generic boolean() register field has no candidate here, and there are no flags() fields for the CLI helper's new rendering to reach. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01WYw6yc9R1sgG2qybAp6p5j --- pyproject.toml | 4 +-- tests/test_device.py | 85 ++++++++++++++++---------------------------- 2 files changed, 32 insertions(+), 57 deletions(-) diff --git a/pyproject.toml b/pyproject.toml index 8c3cf0f..9bf758b 100755 --- a/pyproject.toml +++ b/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>=4.0"] +dependencies = ["modbus-connection>=4.1"] [project.optional-dependencies] # Concrete backend for the script/query.py CLI (pip install trovis-modbus[cli]). -cli = ["modbus-connection[pymodbus]>=4.0"] +cli = ["modbus-connection[pymodbus]>=4.1"] [project.urls] Homepage = "https://github.com/Tom-Bom-badil/trovis-modbus" diff --git a/tests/test_device.py b/tests/test_device.py index f5a383c..f5feba2 100755 --- a/tests/test_device.py +++ b/tests/test_device.py @@ -17,39 +17,14 @@ Trovis557x, Weekday, ) -from trovis_modbus.configurations.address_ranges import REGISTER_RANGES +from trovis_modbus.configurations.address_ranges import ( + COIL_RANGES, + REGISTER_RANGES, +) from .conftest import COILS, HOLDING -class _CountingUnit: - """Wraps a ModbusUnit and records read calls; delegates everything else.""" - - def __init__(self, inner: MockModbusUnit) -> None: - self._inner = inner - self.register_blocks: list[tuple[int, int]] = [] - self.coil_blocks: list[tuple[int, int]] = [] - - @property - def register_reads(self) -> int: - return len(self.register_blocks) - - @property - def coil_reads(self) -> int: - return len(self.coil_blocks) - - async def read_holding_registers(self, address: int, count: int) -> list[int]: - self.register_blocks.append((address, count)) - return await self._inner.read_holding_registers(address, count) - - async def read_coils(self, address: int, count: int) -> list[bool]: - self.coil_blocks.append((address, count)) - return await self._inner.read_coils(address, count) - - def __getattr__(self, name: str) -> object: - return getattr(self._inner, name) - - async def test_device_info(trovis: Trovis557x) -> None: await trovis.async_update() info = trovis.info @@ -159,46 +134,46 @@ async def test_independent_component_update(trovis: Trovis557x) -> None: assert trovis.rk1.flow_setpoint is None # not updated yet -async def test_full_update_consolidates_reads() -> None: +async def test_full_update_consolidates_reads( + trovis: Trovis557x, unit: MockModbusUnit +) -> None: """A full device update pools all sub-systems into a few block reads.""" - inner = MockModbusConnection().for_unit(1) - inner.holding.update(HOLDING) - inner.coils.update(COILS) - unit = _CountingUnit(inner) - device = Trovis557x(unit) # type: ignore[arg-type] - - field_count = sum(len(c.readable_field_names) for c in device.components) - await device.async_update() + field_count = sum(len(c.readable_field_names) for c in trovis.components) + await trovis.async_update() # Many fields collapse into a small number of range-aware block reads — far # fewer than the field count, and well under a naive per-field strategy. - total_reads = unit.register_reads + unit.coil_reads - assert total_reads < field_count // 4 + assert len(unit.read_events) < field_count // 4 # The exact number of blocks may change when manufacturer ranges or the # planner improve. The stable safety guarantee is the configured max_span. - assert all(count <= 50 for _, count in unit.register_blocks) - assert all(count <= 50 for _, count in unit.coil_blocks) + assert all(event.count <= 50 for event in unit.read_events) async def test_full_update_never_reads_across_an_unreadable_gap( - unit: MockModbusUnit, + trovis: Trovis557x, unit: MockModbusUnit ) -> None: """Every block stays inside the controller's readable ranges (no NAK risk).""" - device = Trovis557x(unit) - # async_read_raw reports every address the pooled read touched, so a block - # spanning a gap shows up as a read of an address no range declares. - raw = await device._group.async_read_raw() - read = set(raw["holding"]) - - unreadable = { - address - for address in read - if not any(low <= address <= high for low, high in REGISTER_RANGES) - } - assert not unreadable, f"read addresses outside every readable range: {unreadable}" + await trovis.async_update() + + ranges = {"holding": REGISTER_RANGES, "coil": COIL_RANGES} + for event in unit.read_events: + declared = ranges[event.register_type] + covered = range(event.address, event.address + event.count) + assert all( + any(low <= address <= high for low, high in declared) for address in covered + ), ( + f"{event.register_type} block {covered.start}..{covered.stop - 1} " + "crosses an unreadable gap" + ) # Addresses 7-8 (between ranges [0,6] and [9,40]) are never read — the low # registers split there instead of being merged into one 0..26 block. + read = { + event.address + i + for event in unit.read_events + if event.register_type == "holding" + for i in range(event.count) + } assert 7 not in read and 8 not in read From 08e1912da2fe04d820ac9625ed2427bda64a28b4 Mon Sep 17 00:00:00 2001 From: Paulus Schoutsen Date: Sat, 8 Aug 2026 15:21:41 -0400 Subject: [PATCH 7/7] Bump modbus-connection to 4.2 and pin the write sequences it can now see MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 4.2.0's mock upgrades close two coverage gaps that were previously untestable here. WriteEvent now carries the function code. Clock.write deliberately sends the year and the packed date as two FC06 writes, year first, instead of one FC16 across both words — the sequence the 55Pro uses. Asserting the resulting register values cannot tell the two apart, so deleting that override passed the whole suite and would have failed on hardware. Verified: with the override removed, 644 tests still pass and only the new one fails. fail_requests() models a device that answers nothing, without a test having to guess which address its caller reaches first. That covers the four TrovisWriteAccessError wrappers around HR40145, which had no tests at all. It is armed with GatewayTargetError — one of the new typed exception-response subclasses, and the realistic failure for a TROVIS behind an RTU-over-TCP bridge. Also cover disconnect(), 4.2.0's headline addition: the owner recycling a stuck link keeps the same unit handles and components, and does not fire the on_connection_lost callbacks. Not applicable: TROVIS declares no repeating_group, no flags(), and no multi-register numeric fields, and every boolean is a real coil. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01WYw6yc9R1sgG2qybAp6p5j --- pyproject.toml | 4 ++-- tests/test_device.py | 43 ++++++++++++++++++++++++++++++++++- tests/test_temporal_values.py | 23 ++++++++++++++++++- 3 files changed, 66 insertions(+), 4 deletions(-) diff --git a/pyproject.toml b/pyproject.toml index 9bf758b..fb01253 100755 --- a/pyproject.toml +++ b/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>=4.1"] +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]>=4.1"] +cli = ["modbus-connection[pymodbus]>=4.2"] [project.urls] Homepage = "https://github.com/Tom-Bom-badil/trovis-modbus" diff --git a/tests/test_device.py b/tests/test_device.py index f5feba2..fb4c1b9 100755 --- a/tests/test_device.py +++ b/tests/test_device.py @@ -5,7 +5,7 @@ from datetime import date, time import pytest -from modbus_connection import ClientClosedError +from modbus_connection import ClientClosedError, GatewayTargetError from modbus_connection.mock import MockModbusConnection, MockModbusUnit from trovis_modbus import ( @@ -15,6 +15,7 @@ SystemActivity, TemperatureRange, Trovis557x, + TrovisWriteAccessError, Weekday, ) from trovis_modbus.configurations.address_ranges import ( @@ -211,6 +212,25 @@ async def test_update_survives_a_dropped_connection() -> None: assert device.rk1.flow_setpoint == pytest.approx(55.0) +async def test_update_survives_a_recycled_connection() -> None: + """disconnect() recycles a stuck link; the device keeps its handles.""" + connection = MockModbusConnection() + unit = connection.for_unit(1) + unit.holding.update(HOLDING) + unit.coils.update(COILS) + device = Trovis557x(unit) + await device.async_update() + + lost: list[int] = [] + connection.on_connection_lost(lambda: lost.append(1)) + await connection.disconnect() + + await device.async_update() + assert device.rk1.flow_setpoint == pytest.approx(55.0) + # The owner tore the link down; nothing was lost. + assert lost == [] + + async def test_update_after_close_raises() -> None: """Closing is the owner's permanent end of the connection, not a drop.""" connection = MockModbusConnection() @@ -299,6 +319,27 @@ async def test_write_refreshes_access_code( assert (await unit.read_holding_registers(1002, 1))[0] == 215 +async def test_write_access_reports_a_device_that_answers_nothing( + trovis: Trovis557x, unit: MockModbusUnit +) -> None: + """Every HR40145 path reports a refusal as a TROVIS write-access failure.""" + # A gateway that answers for itself while the controller behind it does not + # — the usual failure of a TROVIS on an RTU-over-TCP bridge. + unit.fail_requests(GatewayTargetError()) + + with pytest.raises(TrovisWriteAccessError): + await trovis.async_read_writing_enabled() + with pytest.raises(TrovisWriteAccessError): + await trovis.async_enable_writing() + with pytest.raises(TrovisWriteAccessError): + await trovis.async_disable_writing() + # The public datapoint write refreshes the access code first, so it fails + # there rather than reaching the register it was asked to write. + with pytest.raises(TrovisWriteAccessError): + await trovis.rk1.set_room_setpoint_day(21.5) + assert trovis.writing_enabled is False + + async def test_5576_anlage_2_1_exposes_rk1_and_rk4( mock_modbus_unit: MockModbusUnit, ) -> None: diff --git a/tests/test_temporal_values.py b/tests/test_temporal_values.py index 111a1f0..c08be4e 100755 --- a/tests/test_temporal_values.py +++ b/tests/test_temporal_values.py @@ -5,7 +5,7 @@ from datetime import date, datetime, time import pytest -from modbus_connection.mock import MockModbusUnit +from modbus_connection.mock import MockModbusUnit, WriteEvent from trovis_modbus import MonthDay, Trovis557x, TrovisValueValidationError @@ -80,6 +80,27 @@ async def test_clock_native_writes(trovis: Trovis557x, unit: MockModbusUnit) -> assert trovis.clock.date == date(2027, 1, 2) +async def test_clock_date_write_sends_the_year_first_as_two_single_writes( + trovis: Trovis557x, unit: MockModbusUnit +) -> None: + """The 55Pro sequence: HR40102 then HR40101, never one FC16 across both. + + Asserting the resulting register values cannot see the difference — a single + FC16 leaves the same two words behind — so this pins the function codes and + their order instead. + """ + writes: list[WriteEvent] = [] + unit.on_write(writes.append) + + await trovis.clock.set_date(date(2027, 1, 2)) + + assert [ + (event.address, event.values, event.function_code) + for event in writes + if event.address in (100, 101) + ] == [(101, [2027], 0x06), (100, [201], 0x06)] + + async def test_datetime_write(trovis: Trovis557x) -> None: await trovis.clock.set_datetime(datetime(2028, 12, 31, 23, 45)) await trovis.clock.async_update()