diff --git a/pyproject.toml b/pyproject.toml index c969954..7d30b84 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "hatchling.build" [project] name = "python-qube-heatpump" -version = "1.14.0" +version = "1.15.0" authors = [ { name="MattieGit", email="6250046+MattieGit@users.noreply.github.com" }, ] diff --git a/src/python_qube_heatpump/client.py b/src/python_qube_heatpump/client.py index c598fd0..513ef66 100644 --- a/src/python_qube_heatpump/client.py +++ b/src/python_qube_heatpump/client.py @@ -2,6 +2,7 @@ from __future__ import annotations +import contextlib import logging import math import struct @@ -10,6 +11,7 @@ from typing import Any from pymodbus.client import AsyncModbusTcpClient +from pymodbus.exceptions import ConnectionException, ModbusIOException from . import const from .entities import BINARY_SENSORS, SENSORS, SWITCHES, EntityDef @@ -18,6 +20,22 @@ _LOGGER = logging.getLogger(__name__) +# Exceptions meaning the TCP link (or the device behind it) stopped answering, +# as opposed to a Modbus error reply to one request. pymodbus raises +# ConnectionException when it cannot connect and ModbusIOException when a +# request got no response after its retries. OSError covers socket errors +# and TimeoutError. +_LINK_ERRORS: tuple[type[Exception], ...] = ( + ConnectionException, + ModbusIOException, + OSError, +) + + +class _BlockReadError(Exception): + """A block read got a Modbus error reply or a short response.""" + + # Entities read by get_all_data() (used by the official HA core integration). # These mirror the register definitions in `const` exactly (address, scale, # offset) so batching cannot change any value get_all_data() returns. @@ -168,10 +186,29 @@ def _log_read_failure(self, target: str, exc: Exception) -> None: self._read_failures_warned.add(target) _LOGGER.warning("Exception reading %s: %s", target, exc) + def _log_read_ok(self, target: str) -> None: + """Re-arm the warn-once log for a target that reads fine again.""" + if target in self._read_failures_warned: + self._read_failures_warned.discard(target) + _LOGGER.info("Reading %s recovered", target) + + def _mark_disconnected(self) -> None: + """Drop the connection after a link failure so the next poll reconnects. + + pymodbus' connect() replaces the transport without closing the old + one, so the transport is closed here before any reconnect. + """ + self._connected = False + with contextlib.suppress(Exception): + self._client.close() + async def connect(self) -> bool: """Connect to the Modbus server.""" if not self._connected: self._connected = await self._client.connect() + if self._connected: + self._backoff_seconds = 0.0 + self._next_connect_at = 0.0 return self._connected @property @@ -210,7 +247,8 @@ async def get_all_data(self) -> QubeState | None: """Fetch all definition data and return a state object. This fetches core sensors for the official HA integration. - Returns None if not connected and reconnection fails. + Returns None if not connected and reconnection fails, or if the + connection was lost during this poll. """ await self._ensure_connected() if not self._connected: @@ -225,6 +263,8 @@ async def get_all_data(self) -> QubeState | None: # handful of block transactions instead of ~59 per-field reads. entities = [*_CORE_STATE_ENTITIES, *BINARY_SENSORS.values()] results = await self.read_entities_batched(entities) + if not self._connected: + return None for ent in _CORE_STATE_ENTITIES: setattr(state, ent.key, results.get(ent.key)) @@ -450,10 +490,13 @@ async def read_value(self, definition: tuple) -> float | None: if offset is not None: val += offset + self._log_read_ok(f"address {address}") return val except Exception as e: self._log_read_failure(f"address {address}", e) + if isinstance(e, _LINK_ERRORS): + self._mark_disconnected() return None @staticmethod @@ -542,51 +585,32 @@ async def read_entities_batched( Groups entities into a handful of Modbus block reads instead of one transaction per entity. If a block read fails, its entities - are read individually as a fallback. + are read individually as a fallback. If one of those single reads + hits a link failure (no connection, no response), the device is + treated as unreachable: the client is marked disconnected and the + remaining reads are skipped. Returns: - Dictionary mapping entity keys to their values (None on error). + Dictionary mapping every requested entity key to its value + (None on error). """ results: dict[str, Any] = {} - for input_type_str, start, count, members in self._plan_blocks(entities): + blocks = self._plan_blocks(entities) + for input_type_str, start, count, members in blocks: + target = f"block {input_type_str}@{start} (count {count})" try: - if input_type_str == "coil": - result = await self._client.read_coils( - start, count=count, device_id=self.unit - ) - elif input_type_str == "discrete_input": - result = await self._client.read_discrete_inputs( - start, count=count, device_id=self.unit - ) - elif input_type_str == "input": - result = await self._client.read_input_registers( - start, count=count, device_id=self.unit - ) - else: # holding - result = await self._client.read_holding_registers( - start, count=count, device_id=self.unit - ) - if result.isError(): - raise OSError(f"Modbus error response for block @{start}") - if input_type_str in ("coil", "discrete_input"): - if len(result.bits) < count: - raise OSError( - f"Short bit response for block {input_type_str}@{start} " - f"(got {len(result.bits)}, expected {count})" - ) - elif len(result.registers) < count: - raise OSError( - f"Short register response for block {input_type_str}@{start} " - f"(got {len(result.registers)}, expected {count})" - ) + result = await self._read_block(input_type_str, start, count) except Exception as exc: - self._log_read_failure( - f"block {input_type_str}@{start} (count {count})", exc - ) - for ent in members: - results[ent.key] = await self.read_entity(ent) + self._log_read_failure(target, exc) + if not await self._read_individually(members, results): + self._mark_disconnected() + for _, _, _, block_members in blocks: + for ent in block_members: + results.setdefault(ent.key, None) + return results continue + self._log_read_ok(target) for ent in members: offset = ent.address - start if input_type_str in ("coil", "discrete_input"): @@ -599,6 +623,63 @@ async def read_entities_batched( results[ent.key] = self._apply_scaling(ent, val) return results + async def _read_block(self, input_type_str: str, start: int, count: int) -> Any: + """Read one block and validate the response. + + Raises: + _BlockReadError: On a Modbus error reply or a short response. + Any of _LINK_ERRORS: When the device could not be reached. + """ + if input_type_str == "coil": + result = await self._client.read_coils( + start, count=count, device_id=self.unit + ) + elif input_type_str == "discrete_input": + result = await self._client.read_discrete_inputs( + start, count=count, device_id=self.unit + ) + elif input_type_str == "input": + result = await self._client.read_input_registers( + start, count=count, device_id=self.unit + ) + else: # holding + result = await self._client.read_holding_registers( + start, count=count, device_id=self.unit + ) + if result.isError(): + raise _BlockReadError(f"Modbus error response for block @{start}") + if input_type_str in ("coil", "discrete_input"): + if len(result.bits) < count: + raise _BlockReadError( + f"Short bit response for block {input_type_str}@{start} " + f"(got {len(result.bits)}, expected {count})" + ) + elif len(result.registers) < count: + raise _BlockReadError( + f"Short register response for block {input_type_str}@{start} " + f"(got {len(result.registers)}, expected {count})" + ) + return result + + async def _read_individually( + self, members: list[EntityDef], results: dict[str, Any] + ) -> bool: + """Read a failed block's entities one by one into ``results``. + + Returns False as soon as a read hits a link failure, so a dead + device costs one extra transaction instead of one per entity. + """ + for ent in members: + try: + results[ent.key] = await self._read_entity_or_raise(ent) + except _LINK_ERRORS as exc: + self._log_read_failure(f"entity {ent.key}", exc) + return False + except Exception as exc: + self._log_read_failure(f"entity {ent.key}", exc) + results[ent.key] = None + return True + async def read_entity(self, entity: EntityDef) -> Any: """Read a single entity value based on EntityDef. @@ -608,33 +689,41 @@ async def read_entity(self, entity: EntityDef) -> Any: Returns: The read value (float, int, or bool depending on entity type). """ + try: + return await self._read_entity_or_raise(entity) + except Exception as e: + self._log_read_failure(f"entity {entity.key}", e) + if isinstance(e, _LINK_ERRORS): + self._mark_disconnected() + return None + + async def _read_entity_or_raise(self, entity: EntityDef) -> Any: + """Read a single entity; return None on an error reply, raise otherwise.""" # Determine register count based on data type # Use string comparison to handle potential enum class differences data_type_str = entity.data_type.value if entity.data_type else None count = self._register_count(entity) - try: - # Read based on input type (use string comparison for safety) - input_type_str = entity.input_type.value if entity.input_type else None - - if input_type_str == "coil": - result = await self._client.read_coils( - entity.address, count=1, device_id=self.unit - ) - if result.isError(): - _LOGGER.warning("Error reading coil %s", entity.address) - return None - return bool(result.bits[0]) - - if input_type_str == "discrete_input": - result = await self._client.read_discrete_inputs( - entity.address, count=1, device_id=self.unit - ) - if result.isError(): - _LOGGER.warning("Error reading discrete input %s", entity.address) - return None - return bool(result.bits[0]) + # Read based on input type (use string comparison for safety) + input_type_str = entity.input_type.value if entity.input_type else None + if input_type_str == "coil": + result = await self._client.read_coils( + entity.address, count=1, device_id=self.unit + ) + if result.isError(): + _LOGGER.warning("Error reading coil %s", entity.address) + return None + value: Any = bool(result.bits[0]) + elif input_type_str == "discrete_input": + result = await self._client.read_discrete_inputs( + entity.address, count=1, device_id=self.unit + ) + if result.isError(): + _LOGGER.warning("Error reading discrete input %s", entity.address) + return None + value = bool(result.bits[0]) + else: if input_type_str == "input": result = await self._client.read_input_registers( entity.address, count=count, device_id=self.unit @@ -649,11 +738,10 @@ async def read_entity(self, entity: EntityDef) -> Any: return None val = self._decode_registers(data_type_str, result.registers) - return self._apply_scaling(entity, val) + value = self._apply_scaling(entity, val) - except Exception as e: - self._log_read_failure(f"entity {entity.key}", e) - return None + self._log_read_ok(f"entity {entity.key}") + return value async def read_sensor(self, key: str) -> float | int | None: """Read a sensor value by key. @@ -700,16 +788,18 @@ async def read_switch(self, key: str) -> bool | None: return None return await self.read_entity(entity) + async def _read_table(self, table: dict[str, EntityDef]) -> dict[str, Any]: + """Batch-read every entity in a table, keeping the table's key order.""" + results = await self.read_entities_batched(table.values()) + return {key: results.get(key) for key in table} + async def read_all_sensors(self) -> dict[str, Any]: """Read all sensor values. Returns: Dictionary mapping sensor keys to their values. """ - result: dict[str, Any] = {} - for key, entity in SENSORS.items(): - result[key] = await self.read_entity(entity) - return result + return await self._read_table(SENSORS) async def read_all_binary_sensors(self) -> dict[str, bool | None]: """Read all binary sensor values. @@ -717,10 +807,7 @@ async def read_all_binary_sensors(self) -> dict[str, bool | None]: Returns: Dictionary mapping binary sensor keys to their values. """ - result: dict[str, bool | None] = {} - for key, entity in BINARY_SENSORS.items(): - result[key] = await self.read_entity(entity) - return result + return await self._read_table(BINARY_SENSORS) async def read_all_switches(self) -> dict[str, bool | None]: """Read all switch states. @@ -728,10 +815,7 @@ async def read_all_switches(self) -> dict[str, bool | None]: Returns: Dictionary mapping switch keys to their states. """ - result: dict[str, bool | None] = {} - for key, entity in SWITCHES.items(): - result[key] = await self.read_entity(entity) - return result + return await self._read_table(SWITCHES) async def write_switch(self, key: str, value: bool) -> bool: """Write a switch state by key. @@ -762,6 +846,8 @@ async def write_switch(self, key: str, value: bool) -> bool: return True except Exception as e: _LOGGER.error("Exception writing switch %s: %s", key, e) + if isinstance(e, _LINK_ERRORS): + self._mark_disconnected() return False # SG Ready mode API @@ -782,8 +868,11 @@ async def get_sg_ready_mode(self) -> str | None: Returns: Mode string ("off", "block", "plus", "max"), or None on error. """ - bit_a = await self.read_switch("bms_sgready_a") - bit_b = await self.read_switch("bms_sgready_b") + results = await self.read_entities_batched( + [SWITCHES["bms_sgready_a"], SWITCHES["bms_sgready_b"]] + ) + bit_a = results.get("bms_sgready_a") + bit_b = results.get("bms_sgready_b") if bit_a is None or bit_b is None: return None return self._SGREADY_BITS_TO_MODE.get((bool(bit_a), bool(bit_b))) @@ -791,16 +880,41 @@ async def get_sg_ready_mode(self) -> str | None: async def set_sg_ready_mode(self, mode: str) -> bool: """Set the SG Ready mode. + Both coils are written in one request so the heat pump never sees + a half-applied mode. If the device rejects the multi-coil write, + the coils are written one after the other instead. + Args: mode: One of "off", "block", "plus", "max". Returns: - True if both writes succeeded, False otherwise. + True if both coils were written, False otherwise. """ bits = self._SGREADY_MODE_TO_BITS.get(mode) if bits is None: _LOGGER.warning("Unknown SG Ready mode: %s", mode) return False + + coil_a = SWITCHES["bms_sgready_a"] + coil_b = SWITCHES["bms_sgready_b"] + if coil_b.address == coil_a.address + 1: + try: + result = await self._client.write_coils( + coil_a.address, list(bits), device_id=self.unit + ) + except Exception as e: + _LOGGER.error("Exception writing SG Ready mode %s: %s", mode, e) + if isinstance(e, _LINK_ERRORS): + self._mark_disconnected() + return False + if not result.isError(): + return True + _LOGGER.debug( + "Multi-coil write rejected for SG Ready mode %s; " + "falling back to single writes", + mode, + ) + success_a = await self.write_switch("bms_sgready_a", bits[0]) success_b = await self.write_switch("bms_sgready_b", bits[1]) return success_a and success_b @@ -828,6 +942,24 @@ async def write_setpoint(self, key: str, value: float) -> bool: _LOGGER.warning("Sensor %s is not a holding register", key) return False + if not isinstance(value, int | float) or not math.isfinite(value): + _LOGGER.warning( + "Refusing to write %r to %s: not a finite number", value, key + ) + return False + + if (entity.min_value is not None and value < entity.min_value) or ( + entity.max_value is not None and value > entity.max_value + ): + _LOGGER.warning( + "Refusing to write %s to %s: outside allowed range %s..%s", + value, + key, + entity.min_value, + entity.max_value, + ) + return False + try: # Reverse scale/offset if needed write_value = value @@ -846,15 +978,25 @@ async def write_setpoint(self, key: str, value: float) -> bool: result = await self._client.write_registers( entity.address, regs, device_id=self.unit ) - elif entity.data_type == DataType.INT16: - if write_value < 0: - write_value = int(write_value) + 65536 - result = await self._client.write_register( - entity.address, int(write_value), device_id=self.unit + elif entity.data_type in (DataType.INT16, DataType.UINT16): + raw = round(write_value) + low, high = ( + (-32768, 32767) + if entity.data_type == DataType.INT16 + else (0, 65535) ) - elif entity.data_type == DataType.UINT16: + if not low <= raw <= high: + _LOGGER.warning( + "Refusing to write %s to %s: does not fit in %s", + value, + key, + entity.data_type.value, + ) + return False + if raw < 0: + raw += 65536 result = await self._client.write_register( - entity.address, int(write_value), device_id=self.unit + entity.address, raw, device_id=self.unit ) else: _LOGGER.warning( @@ -869,4 +1011,6 @@ async def write_setpoint(self, key: str, value: float) -> bool: except Exception as e: _LOGGER.error("Exception writing setpoint %s: %s", key, e) + if isinstance(e, _LINK_ERRORS): + self._mark_disconnected() return False diff --git a/src/python_qube_heatpump/entities/base.py b/src/python_qube_heatpump/entities/base.py index 0fbacef..64dcad7 100644 --- a/src/python_qube_heatpump/entities/base.py +++ b/src/python_qube_heatpump/entities/base.py @@ -79,3 +79,10 @@ class EntityDef: # Value precision precision: int | None = None """Number of decimal places to round to. None for no rounding.""" + + # Write limits + min_value: float | None = None + """Lowest value write_setpoint() accepts, in the entity's unit.""" + + max_value: float | None = None + """Highest value write_setpoint() accepts, in the entity's unit.""" diff --git a/src/python_qube_heatpump/entities/sensors.py b/src/python_qube_heatpump/entities/sensors.py index 716d30d..12a6190 100644 --- a/src/python_qube_heatpump/entities/sensors.py +++ b/src/python_qube_heatpump/entities/sensors.py @@ -85,6 +85,8 @@ platform=Platform.SENSOR, unit="°C", writable=True, # User-override setpoint, bypasses heating curve + min_value=20.0, + max_value=65.0, ), EntityDef( key="usr_pid_coolsetp", @@ -95,6 +97,8 @@ platform=Platform.SENSOR, unit="°C", writable=True, # User-override setpoint, bypasses cooling curve + min_value=7.0, + max_value=25.0, ), EntityDef( key="regulation_buffersetp_max", @@ -114,6 +118,8 @@ platform=Platform.SENSOR, unit="°C", writable=True, # This setpoint can be written via Modbus + min_value=40.0, + max_value=65.0, ), # Input register sensors (read-only) EntityDef( @@ -460,6 +466,8 @@ platform=Platform.SENSOR, unit="°C", writable=True, + min_value=40.0, + max_value=65.0, ), ) diff --git a/src/python_qube_heatpump/network.py b/src/python_qube_heatpump/network.py index 3e3ae89..7632e34 100644 --- a/src/python_qube_heatpump/network.py +++ b/src/python_qube_heatpump/network.py @@ -54,7 +54,8 @@ async def async_get_mac_address(host: str, port: int = MODBUS_PORT) -> str | Non Returns: MAC address as lowercase colon-separated string, or None. """ - ip = _resolve_ip(host) + # Both lookups block (DNS, file I/O), so keep them off the event loop + ip = await asyncio.to_thread(_resolve_ip, host) if ip is None: _LOGGER.debug("Could not resolve host %s to IP", host) return None @@ -71,4 +72,4 @@ async def async_get_mac_address(host: str, port: int = MODBUS_PORT) -> str | Non _LOGGER.debug("Could not connect to %s:%s to populate ARP", ip, port) return None - return _read_arp_table(ip) + return await asyncio.to_thread(_read_arp_table, ip) diff --git a/tests/conftest.py b/tests/conftest.py index a3ef9a1..beff330 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -1,5 +1,5 @@ import pytest -from unittest.mock import AsyncMock +from unittest.mock import AsyncMock, MagicMock @pytest.fixture @@ -8,7 +8,8 @@ def mock_modbus_client(mocker): mock_client = mocker.patch("python_qube_heatpump.client.AsyncModbusTcpClient") mock_instance = mock_client.return_value mock_instance.connect = AsyncMock(return_value=True) - mock_instance.close = AsyncMock() + # pymodbus close() is synchronous + mock_instance.close = MagicMock() mock_instance.read_holding_registers = AsyncMock() mock_instance.read_input_registers = AsyncMock() mock_instance.write_register = AsyncMock() diff --git a/tests/test_client.py b/tests/test_client.py index 45491da..fb73a92 100644 --- a/tests/test_client.py +++ b/tests/test_client.py @@ -456,16 +456,12 @@ async def test_get_sg_ready_mode(mock_modbus_client): mock_instance = mock_modbus_client.return_value client._client = mock_instance - # Mock coil reads: A=False, B=True -> "plus" - mock_resp_a = MagicMock() - mock_resp_a.isError.return_value = False - mock_resp_a.bits = [False] - - mock_resp_b = MagicMock() - mock_resp_b.isError.return_value = False - mock_resp_b.bits = [True] + # Both coils come back in one block read: A=False, B=True -> "plus" + mock_resp = MagicMock() + mock_resp.isError.return_value = False + mock_resp.bits = [False, True] - mock_instance.read_coils = AsyncMock(side_effect=[mock_resp_a, mock_resp_b]) + mock_instance.read_coils = AsyncMock(return_value=mock_resp) result = await client.get_sg_ready_mode() assert result == "plus" @@ -485,13 +481,10 @@ async def test_get_sg_ready_mode_all_modes(mock_modbus_client): ((True, True), "max"), ] for (bit_a, bit_b), expected_mode in cases: - resp_a = MagicMock() - resp_a.isError.return_value = False - resp_a.bits = [bit_a] - resp_b = MagicMock() - resp_b.isError.return_value = False - resp_b.bits = [bit_b] - mock_instance.read_coils = AsyncMock(side_effect=[resp_a, resp_b]) + resp = MagicMock() + resp.isError.return_value = False + resp.bits = [bit_a, bit_b] + mock_instance.read_coils = AsyncMock(return_value=resp) result = await client.get_sg_ready_mode() assert result == expected_mode, ( @@ -516,23 +509,21 @@ async def test_get_sg_ready_mode_read_error(mock_modbus_client): @pytest.mark.asyncio async def test_set_sg_ready_mode(mock_modbus_client): - """Test setting SG Ready mode writes both coils.""" + """Test setting SG Ready mode writes both coils in one request.""" client = QubeClient("1.2.3.4", 502) mock_instance = mock_modbus_client.return_value client._client = mock_instance mock_resp = MagicMock() mock_resp.isError.return_value = False - mock_instance.write_coil = AsyncMock(return_value=mock_resp) + mock_instance.write_coils = AsyncMock(return_value=mock_resp) result = await client.set_sg_ready_mode("plus") assert result is True - # "plus" = (False, True) -> A=False, B=True - calls = mock_instance.write_coil.call_args_list - assert len(calls) == 2 - assert calls[0].args == (65, False) # bms_sgready_a address - assert calls[1].args == (66, True) # bms_sgready_b address + # "plus" = (False, True) -> A=False (coil 65), B=True (coil 66) + mock_instance.write_coils.assert_awaited_once() + assert mock_instance.write_coils.call_args.args == (65, [False, True]) @pytest.mark.asyncio diff --git a/tests/test_client_read_all.py b/tests/test_client_read_all.py new file mode 100644 index 0000000..b9b7e11 --- /dev/null +++ b/tests/test_client_read_all.py @@ -0,0 +1,83 @@ +"""read_all_* helpers must match per-entity reads while batching requests.""" + +from unittest.mock import AsyncMock, MagicMock + +import pytest + +from python_qube_heatpump import QubeClient +from python_qube_heatpump.entities import BINARY_SENSORS, SENSORS, SWITCHES + + +def _fake_device(mock_instance): + """Serve a fixed, address-dependent register and bit map.""" + + def registers(address, count=1, **kwargs): + resp = MagicMock() + resp.isError.return_value = False + # Varied words so float32/int16/uint32 decoding is exercised + resp.registers = [ + (a * 7919 + 0x4100) & 0xFFFF for a in range(address, address + count) + ] + return resp + + def bits(address, count=1, **kwargs): + resp = MagicMock() + resp.isError.return_value = False + resp.bits = [ + bool((a * 2654435761) & 4) for a in range(address, address + count) + ] + return resp + + mock_instance.read_input_registers = AsyncMock(side_effect=registers) + mock_instance.read_holding_registers = AsyncMock(side_effect=registers) + mock_instance.read_coils = AsyncMock(side_effect=bits) + mock_instance.read_discrete_inputs = AsyncMock(side_effect=bits) + + +def _total_reads(mock_instance) -> int: + return sum( + getattr(mock_instance, name).call_count + for name in ( + "read_coils", + "read_discrete_inputs", + "read_input_registers", + "read_holding_registers", + ) + ) + + +def _same(a, b) -> bool: + if isinstance(a, float) and isinstance(b, float) and a != a and b != b: + return True # both NaN + return a == b and type(a) is type(b) + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + ("method", "table"), + [ + ("read_all_sensors", SENSORS), + ("read_all_binary_sensors", BINARY_SENSORS), + ("read_all_switches", SWITCHES), + ], +) +async def test_read_all_matches_single_reads_with_fewer_requests( + mock_modbus_client, method, table +): + client = QubeClient("1.2.3.4", 502) + mock_instance = mock_modbus_client.return_value + client._client = mock_instance + _fake_device(mock_instance) + + expected = {key: await client.read_entity(ent) for key, ent in table.items()} + single_reads = _total_reads(mock_instance) + + mock_instance.reset_mock() + _fake_device(mock_instance) + result = await getattr(client, method)() + + assert list(result) == list(table) + mismatched = [k for k in table if not _same(result[k], expected[k])] + assert mismatched == [] + assert _total_reads(mock_instance) < single_reads + assert _total_reads(mock_instance) == len(client._plan_blocks(table.values())) diff --git a/tests/test_client_resilience.py b/tests/test_client_resilience.py new file mode 100644 index 0000000..d888a59 --- /dev/null +++ b/tests/test_client_resilience.py @@ -0,0 +1,252 @@ +"""Tests for QubeClient behaviour when the Modbus link fails.""" + +import logging +from unittest.mock import AsyncMock, MagicMock + +import pytest +from pymodbus.exceptions import ConnectionException, ModbusIOException + +from python_qube_heatpump import QubeClient +from python_qube_heatpump.entities import BINARY_SENSORS, SWITCHES + + +def _ok_registers(address, count=1, **kwargs): + resp = MagicMock() + resp.isError.return_value = False + resp.registers = [0] * count + return resp + + +def _ok_bits(address, count=1, **kwargs): + resp = MagicMock() + resp.isError.return_value = False + resp.bits = [False] * count + return resp + + +def _error_reply(*args, **kwargs): + resp = MagicMock() + resp.isError.return_value = True + return resp + + +def _all_reads(mock_instance, side_effect): + """Point every read function at the same side effect.""" + for name in ( + "read_coils", + "read_discrete_inputs", + "read_input_registers", + "read_holding_registers", + ): + setattr(mock_instance, name, AsyncMock(side_effect=side_effect)) + + +def _total_reads(mock_instance) -> int: + return sum( + getattr(mock_instance, name).call_count + for name in ( + "read_coils", + "read_discrete_inputs", + "read_input_registers", + "read_holding_registers", + ) + ) + + +def _healthy(mock_instance): + mock_instance.read_coils = AsyncMock(side_effect=_ok_bits) + mock_instance.read_discrete_inputs = AsyncMock(side_effect=_ok_bits) + mock_instance.read_input_registers = AsyncMock(side_effect=_ok_registers) + mock_instance.read_holding_registers = AsyncMock(side_effect=_ok_registers) + + +@pytest.fixture +def connected_client(mock_modbus_client): + client = QubeClient("1.2.3.4", 502) + mock_instance = mock_modbus_client.return_value + client._client = mock_instance + client._connected = True + return client, mock_instance + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + "exc", + [ + ConnectionException("down"), + ModbusIOException("No response received after 3 retries"), + TimeoutError(), + ConnectionResetError(), + ], +) +async def test_link_error_marks_client_disconnected(connected_client, exc): + """A link-level failure clears is_connected and closes the transport.""" + client, mock_instance = connected_client + _all_reads(mock_instance, exc) + + await client.read_entities_batched(SWITCHES.values()) + + assert client.is_connected is False + mock_instance.close.assert_called() + + +@pytest.mark.asyncio +async def test_get_all_data_stops_after_first_link_error(connected_client): + """A dead link costs two transactions per poll, not one per entity. + + The failed block read is followed by one single-entity read that acts as + a probe (a device may time out on a large block but answer small reads); + when that fails too, the poll stops. + """ + client, mock_instance = connected_client + _all_reads(mock_instance, ModbusIOException("no response")) + + state = await client.get_all_data() + + assert state is None + assert _total_reads(mock_instance) == 2 + + +@pytest.mark.asyncio +async def test_batched_read_returns_every_key_after_link_error(connected_client): + """Callers still get a value (None) for every requested entity.""" + client, mock_instance = connected_client + _all_reads(mock_instance, ConnectionException("down")) + + results = await client.read_entities_batched(BINARY_SENSORS.values()) + + assert set(results) == set(BINARY_SENSORS) + assert all(value is None for value in results.values()) + + +@pytest.mark.asyncio +async def test_block_timeout_with_working_single_reads_stays_connected( + connected_client, +): + """A block that times out while single reads work is not a dead link.""" + client, mock_instance = connected_client + mock_instance.read_discrete_inputs = AsyncMock( + side_effect=[ModbusIOException("no response"), _ok_bits(0), _ok_bits(1)] + ) + + entities = [BINARY_SENSORS["dout_srcpmp_val"], BINARY_SENSORS["dout_usrpmp_val"]] + results = await client.read_entities_batched(entities) + + assert results == {e.key: False for e in entities} + assert client.is_connected is True + + +@pytest.mark.asyncio +async def test_error_reply_still_falls_back_to_single_reads(connected_client): + """A Modbus error reply is not a link failure: fall back per entity.""" + client, mock_instance = connected_client + _all_reads(mock_instance, _error_reply) + + entities = [BINARY_SENSORS["dout_srcpmp_val"], BINARY_SENSORS["dout_usrpmp_val"]] + await client.read_entities_batched(entities) + + # one block read plus one single read per entity + assert mock_instance.read_discrete_inputs.call_count == 3 + assert client.is_connected is True + + +@pytest.mark.asyncio +async def test_single_read_fallback_stops_at_link_error(connected_client): + """If the link dies during the per-entity fallback, the rest is skipped.""" + client, mock_instance = connected_client + mock_instance.read_discrete_inputs = AsyncMock( + side_effect=[_error_reply(), ConnectionException("down")] + [_ok_bits(0)] * 10 + ) + + entities = [ + BINARY_SENSORS["dout_srcpmp_val"], + BINARY_SENSORS["dout_usrpmp_val"], + BINARY_SENSORS["dout_fourwayvlv_val"], + ] + results = await client.read_entities_batched(entities) + + assert mock_instance.read_discrete_inputs.call_count == 2 + assert results == {e.key: None for e in entities} + assert client.is_connected is False + + +@pytest.mark.asyncio +async def test_get_all_data_reconnects_on_next_poll(connected_client): + """After a lost link the next poll reconnects and returns data again.""" + client, mock_instance = connected_client + _all_reads(mock_instance, ConnectionException("down")) + assert await client.get_all_data() is None + assert client.is_connected is False + + mock_instance.connect = AsyncMock(return_value=True) + _healthy(mock_instance) + state = await client.get_all_data() + + assert state is not None + mock_instance.connect.assert_awaited_once() + assert client.is_connected is True + + +@pytest.mark.asyncio +async def test_read_entity_link_error_marks_disconnected(connected_client): + """Single-entity reads detect a lost link too.""" + client, mock_instance = connected_client + _all_reads(mock_instance, ConnectionException("down")) + + assert await client.read_entity(BINARY_SENSORS["dout_srcpmp_val"]) is None + assert client.is_connected is False + + +@pytest.mark.asyncio +async def test_write_link_error_marks_disconnected(connected_client): + """Writes that hit a lost link return False and clear is_connected.""" + client, mock_instance = connected_client + mock_instance.write_coil = AsyncMock(side_effect=ConnectionException("down")) + + assert await client.write_switch("bms_summerwinter", True) is False + assert client.is_connected is False + + +@pytest.mark.asyncio +async def test_failure_warns_again_after_recovery(connected_client, caplog): + """Warn-once resets when the target reads fine again.""" + client, mock_instance = connected_client + entity = BINARY_SENSORS["dout_srcpmp_val"] + mock_instance.read_discrete_inputs = AsyncMock( + side_effect=[ + _error_reply(), + _error_reply(), + _ok_bits(0), + _error_reply(), + _error_reply(), + ] + ) + with caplog.at_level(logging.DEBUG, logger="python_qube_heatpump.client"): + await client.read_entities_batched([entity]) # block fails, single fails + await client.read_entities_batched([entity]) # recovers + await client.read_entities_batched([entity]) # block fails again + + warnings = [ + r + for r in caplog.records + if r.levelno == logging.WARNING and "block" in r.getMessage() + ] + assert len(warnings) == 2 + + +@pytest.mark.asyncio +async def test_link_failure_warns_again_after_recovery(connected_client, caplog): + """A second outage days later is logged at WARNING again.""" + client, mock_instance = connected_client + entity = BINARY_SENSORS["dout_srcpmp_val"] + mock_instance.read_discrete_inputs = AsyncMock( + side_effect=[OSError("down"), _ok_bits(0), OSError("down again")] + ) + + with caplog.at_level(logging.DEBUG, logger="python_qube_heatpump.client"): + await client.read_entity(entity) + await client.read_entity(entity) + await client.read_entity(entity) + + warnings = [r for r in caplog.records if r.levelno == logging.WARNING] + assert len(warnings) == 2 diff --git a/tests/test_client_writes.py b/tests/test_client_writes.py new file mode 100644 index 0000000..348669b --- /dev/null +++ b/tests/test_client_writes.py @@ -0,0 +1,197 @@ +"""Tests for write validation and SG Ready writes in QubeClient.""" + +import math +import struct +from unittest.mock import AsyncMock, MagicMock + +import pytest +from pymodbus.exceptions import ConnectionException + +from python_qube_heatpump import QubeClient +from python_qube_heatpump.entities import SENSORS +from python_qube_heatpump.entities.base import ( + DataType, + EntityDef, + InputType, + Platform, +) + + +def _ok(): + resp = MagicMock() + resp.isError.return_value = False + return resp + + +def _error_reply(): + resp = MagicMock() + resp.isError.return_value = True + return resp + + +def _float_from_regs(regs): + return struct.unpack(">f", struct.pack(">I", (regs[0] << 16) | regs[1]))[0] + + +@pytest.fixture +def client(mock_modbus_client): + qube = QubeClient("1.2.3.4", 502) + mock_instance = mock_modbus_client.return_value + mock_instance.write_registers = AsyncMock(return_value=_ok()) + mock_instance.write_register = AsyncMock(return_value=_ok()) + mock_instance.write_coil = AsyncMock(return_value=_ok()) + mock_instance.write_coils = AsyncMock(return_value=_ok()) + qube._client = mock_instance + qube._connected = True + return qube, mock_instance + + +# Ranges the HA core and HACS integrations expose for these setpoints +@pytest.mark.parametrize( + ("key", "low", "high"), + [ + ("setpoint_dhw", 40.0, 65.0), + ("tapw_timeprogram_dhwsetp_nolinq", 40.0, 65.0), + ("usr_pid_heatsetp", 20.0, 65.0), + ("usr_pid_coolsetp", 7.0, 25.0), + ], +) +def test_writable_setpoints_have_safe_ranges(key, low, high): + entity = SENSORS[key] + assert entity.min_value == low + assert entity.max_value == high + + +def test_every_writable_sensor_has_a_range(): + for entity in SENSORS.values(): + if entity.writable: + assert entity.min_value is not None, entity.key + assert entity.max_value is not None, entity.key + + +@pytest.mark.asyncio +@pytest.mark.parametrize("value", [40.0, 52.5, 65.0]) +async def test_write_setpoint_accepts_values_in_range(client, value): + qube, mock_instance = client + assert await qube.write_setpoint("setpoint_dhw", value) is True + regs = mock_instance.write_registers.call_args.args[1] + assert _float_from_regs(regs) == pytest.approx(value) + + +@pytest.mark.asyncio +@pytest.mark.parametrize("value", [39.9, 65.1, 500.0, -10.0]) +async def test_write_setpoint_rejects_out_of_range(client, value): + qube, mock_instance = client + assert await qube.write_setpoint("setpoint_dhw", value) is False + mock_instance.write_registers.assert_not_called() + + +@pytest.mark.asyncio +@pytest.mark.parametrize("value", [math.nan, math.inf, -math.inf]) +async def test_write_setpoint_rejects_non_finite(client, value): + qube, mock_instance = client + assert await qube.write_setpoint("usr_pid_heatsetp", value) is False + mock_instance.write_registers.assert_not_called() + + +def _int_entity(data_type, **kwargs): + return EntityDef( + key="test_int_setpoint", + name="Test", + address=500, + input_type=InputType.HOLDING_REGISTER, + data_type=data_type, + platform=Platform.SENSOR, + writable=True, + **kwargs, + ) + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + ("data_type", "value", "written"), + [ + (DataType.UINT16, 21.6, 22), + (DataType.UINT16, 21.4, 21), + (DataType.INT16, -2.6, 65536 - 3), + ], +) +async def test_write_setpoint_rounds_16bit_values( + client, monkeypatch, data_type, value, written +): + qube, mock_instance = client + monkeypatch.setitem(SENSORS, "test_int_setpoint", _int_entity(data_type)) + assert await qube.write_setpoint("test_int_setpoint", value) is True + assert mock_instance.write_register.call_args.args[1] == written + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + ("data_type", "value"), + [ + (DataType.UINT16, -1), + (DataType.UINT16, 65536), + (DataType.INT16, 32768), + (DataType.INT16, -32769), + ], +) +async def test_write_setpoint_rejects_16bit_overflow( + client, monkeypatch, data_type, value +): + qube, mock_instance = client + monkeypatch.setitem(SENSORS, "test_int_setpoint", _int_entity(data_type)) + assert await qube.write_setpoint("test_int_setpoint", value) is False + mock_instance.write_register.assert_not_called() + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + ("mode", "bits"), + [ + ("off", [False, False]), + ("block", [True, False]), + ("plus", [False, True]), + ("max", [True, True]), + ], +) +async def test_set_sg_ready_mode_writes_both_coils_at_once(client, mode, bits): + qube, mock_instance = client + assert await qube.set_sg_ready_mode(mode) is True + mock_instance.write_coils.assert_awaited_once() + call = mock_instance.write_coils.call_args + assert call.args == (65, bits) + mock_instance.write_coil.assert_not_called() + + +@pytest.mark.asyncio +async def test_set_sg_ready_mode_falls_back_when_multi_write_unsupported(client): + """A device that rejects FC15 still gets the mode via two single writes.""" + qube, mock_instance = client + mock_instance.write_coils = AsyncMock(return_value=_error_reply()) + + assert await qube.set_sg_ready_mode("plus") is True + calls = mock_instance.write_coil.call_args_list + assert [c.args for c in calls] == [(65, False), (66, True)] + + +@pytest.mark.asyncio +async def test_set_sg_ready_mode_link_error_does_not_fall_back(client): + qube, mock_instance = client + mock_instance.write_coils = AsyncMock(side_effect=ConnectionException("down")) + + assert await qube.set_sg_ready_mode("max") is False + mock_instance.write_coil.assert_not_called() + assert qube.is_connected is False + + +@pytest.mark.asyncio +async def test_get_sg_ready_mode_reads_both_coils_in_one_request(client): + qube, mock_instance = client + resp = _ok() + resp.bits = [True, False, False, False, False, False, False, False] + mock_instance.read_coils = AsyncMock(return_value=resp) + + assert await qube.get_sg_ready_mode() == "block" + mock_instance.read_coils.assert_awaited_once() + assert mock_instance.read_coils.call_args.args[0] == 65 + assert mock_instance.read_coils.call_args.kwargs["count"] == 2 diff --git a/tests/test_network.py b/tests/test_network.py index d27ece2..69f0fae 100644 --- a/tests/test_network.py +++ b/tests/test_network.py @@ -118,3 +118,34 @@ async def test_get_mac_address_with_ip_directly(): result = await async_get_mac_address("192.168.5.208") assert result == "00:0a:5c:94:83:15" + + +@pytest.mark.asyncio +async def test_get_mac_address_does_blocking_work_off_the_event_loop(): + """DNS lookup and the ARP file read must not block the event loop.""" + import threading + + loop_thread = threading.current_thread() + seen: dict[str, threading.Thread] = {} + + def fake_resolve(host): + seen["resolve"] = threading.current_thread() + return "192.168.5.208" + + def fake_arp(ip): + seen["arp"] = threading.current_thread() + return "00:0a:5c:94:83:15" + + with ( + patch( + "python_qube_heatpump.network.asyncio.open_connection", + return_value=(AsyncMock(), AsyncMock()), + ), + patch("python_qube_heatpump.network._resolve_ip", side_effect=fake_resolve), + patch("python_qube_heatpump.network._read_arp_table", side_effect=fake_arp), + ): + result = await async_get_mac_address("qube.local") + + assert result == "00:0a:5c:94:83:15" + assert seen["resolve"] is not loop_thread + assert seen["arp"] is not loop_thread