commit b6ac0e3dd42073246614a537f3cadd505e5e5b7c parent 4d9d7f8fdb3b02d98f29e40ce023cf2b9fe5226f Author: Stéphan Kochen <git@stephank.nl> Date: Tue, 29 Sep 2026 21:42:55 +0200 Align protocol impl with official app
diff --git a/custom_components/inlite/lib/inlite_ble/hub.py b/custom_components/inlite/lib/inlite_ble/hub.py index 04c41587dd89922742ecd74ff89cd3abeac2c75d..b8ad50442b7d914418d568126e6e713e592c5d19 100644 --- a/custom_components/inlite/lib/inlite_ble/hub.py +++ b/custom_components/inlite/lib/inlite_ble/hub.py @@ -4,6 +4,7 @@ from __future__ import annotations import asyncio import logging +from collections import deque from collections.abc import Callable, Iterable from datetime import datetime from typing import Any @@ -44,11 +45,12 @@ _LOGGER = logging.getLogger(__name__) WRITE_DELAY = 0.06 # 60ms between BLE writes (matches app timing) -ACK_TIMEOUT = 2.0 # seconds to wait for hub ACK -STREAM_TIMEOUT = 3.0 # seconds to wait for STREAM response +ACK_TIMEOUT = 0.6 # Official client retries a segment after 600 ms. +TX_STREAM_MAX_RETRIES = 17 # Matches the official client's send budget. +STREAM_TIMEOUT = 15.0 # The official app retains an RX stream for 15 seconds. +STREAM_REACK_INTERVAL = 5.0 # Re-ACK an incomplete response after this delay. BLE_PACKET_PART_SIZE = 78 # Matches the official app's GATT packet splitter. STREAM_DATA_SIZE = 62 # Maximum official-app payload per PKT_BLOCK_DATA. -STREAM_RETRY_LIMIT = 3 # Bound retransmits or out-of-order response packets. SMART_HUB_75_PRODUCT_ID = 0x1D @@ -121,10 +123,11 @@ self._crypto = CsrMeshCrypto(passphrase, network_key) self._ble_address = ble_address self._client: BleakClient | None = None self._loop: asyncio.AbstractEventLoop | None = None - # Control packets are queued so a response FLUSH arriving immediately - # after a request-completion ACK cannot be lost by clearing an Event. - self._ack_queue: asyncio.Queue[bytes] = asyncio.Queue() - self._stream_queue: asyncio.Queue[bytes] = asyncio.Queue() + # ACK, FLUSH, and STREAM packets share one queue. Their ordering is + # significant: splitting them into separate queues lets a later STREAM + # overtake a FLUSH/ACK and desynchronise the block-stream state machine. + self._packet_queue: asyncio.Queue[tuple[int, bytes]] = asyncio.Queue() + self._deferred_packets: deque[tuple[int, bytes]] = deque() self._last_stream_data = b"" self._stream_buffer = bytearray() self._stream_invalid = False @@ -276,7 +279,7 @@ loop = self._loop if loop is None or loop.is_closed(): return - if pkt_type in (PKT_BLOCK_ACK, PKT_BLOCK_FLUSH): + if pkt_type in (PKT_BLOCK_ACK, PKT_BLOCK_FLUSH, PKT_BLOCK_STREAM): # Control packets for a command are unicast. The hub uses ACKs # while receiving our request, then FLUSH packets for its response. # Filter before queuing so traffic from another mesh node cannot @@ -285,16 +288,9 @@ if ( decrypted["src_id"] == self._device_id and decrypted["dest_id"] == self._crypto.controller_address ): - loop.call_soon_threadsafe(self._ack_queue.put_nowait, payload) - elif pkt_type == PKT_BLOCK_STREAM: - if ( - decrypted["src_id"] == self._device_id - and decrypted["dest_id"] == self._crypto.controller_address - ): - # ACK each stream segment from the command coroutine. This - # preserves packet order and avoids writing from the BLE - # notification callback. - loop.call_soon_threadsafe(self._stream_queue.put_nowait, payload) + loop.call_soon_threadsafe( + self._packet_queue.put_nowait, (pkt_type, payload) + ) elif pkt_type == PKT_BLOCK_DATA_BLK: if len(payload) >= 3 and payload[1:3] == b"\x0c\x00": self._discovered_device_ids.add(decrypted["src_id"]) @@ -405,10 +401,30 @@ """Encrypt and write a mesh packet.""" packet = self._crypto.encrypt_packet(dest_id, pkt_type, data) await self._write(packet) + async def _wait_packet( + self, packet_types: tuple[int, ...], timeout: float + ) -> tuple[int, bytes]: + """Return the next packet of a requested type without losing others.""" + for packet in self._deferred_packets: + if packet[0] in packet_types: + self._deferred_packets.remove(packet) + return packet + loop = asyncio.get_running_loop() + deadline = loop.time() + timeout + while True: + remaining = deadline - loop.time() + if remaining <= 0: + raise asyncio.TimeoutError + packet = await asyncio.wait_for(self._packet_queue.get(), remaining) + if packet[0] in packet_types: + return packet + self._deferred_packets.append(packet) + async def _wait_ack(self, timeout: float = ACK_TIMEOUT) -> bytes: - """Return the next queued control packet from this hub.""" + """Return the next ACK, preserving an early response packet.""" try: - return await asyncio.wait_for(self._ack_queue.get(), timeout) + _, payload = await self._wait_packet((PKT_BLOCK_ACK,), timeout) + return payload except asyncio.TimeoutError: _LOGGER.warning("ACK timeout after %.1fs", timeout) return b"" @@ -440,79 +456,117 @@ """ dest = self._device_id self._stream_buffer.clear() self._stream_invalid = False - while not self._ack_queue.empty(): - self._ack_queue.get_nowait() - while not self._stream_queue.empty(): - self._stream_queue.get_nowait() + while not self._packet_queue.empty(): + self._packet_queue.get_nowait() + self._deferred_packets.clear() # ``build_block_data_payload`` retains its offset prefix for callers # and tests. The stream transport owns that prefix, however, so strip # its initial zero offset before transmitting numbered segments. # Association is already an offset-free vendor payload. stream_payload = block_data[2:] if block_data.startswith(b"\x00\x00") else block_data - await self._write_mesh(dest, PKT_BLOCK_FLUSH, build_flush_payload(0)) - await self._wait_ack() + async def send_and_wait_for_ack( + packet_type: int, payload: bytes, expected_min: int, expected_max: int + ) -> int | None: + """Send one stream phase until its cumulative ACK advances.""" + for attempt in range(TX_STREAM_MAX_RETRIES): + await self._write_mesh(dest, packet_type, payload) + ack = await self._wait_ack() + if len(ack) >= 2: + acked_bytes = ack[0] | (ack[1] << 8) + if expected_min <= acked_bytes <= expected_max: + return acked_bytes + _LOGGER.debug( + "Stream packet type 0x%02X attempt %d/%d was not acknowledged " + "in range %d..%d", + packet_type, attempt + 1, TX_STREAM_MAX_RETRIES, + expected_min, expected_max, + ) + return None + + if await send_and_wait_for_ack( + PKT_BLOCK_FLUSH, build_flush_payload(0), 0, 0 + ) is None: + _LOGGER.warning("Block stream: start FLUSH was not acknowledged") + return False offset = 0 while offset < len(stream_payload): segment = stream_payload[offset : offset + STREAM_DATA_SIZE] - await self._write_mesh( - dest, + acked_bytes = await send_and_wait_for_ack( PKT_BLOCK_DATA, offset.to_bytes(2, "little") + segment, + offset + 1, + offset + len(segment), ) - ack = await self._wait_ack() - if len(ack) >= 2: - acked_bytes = ack[0] | (ack[1] << 8) - # Acknowledgements carry the total stream offset. Avoid a - # malformed/stale ACK causing a non-progressing loop. - if offset < acked_bytes <= offset + len(segment): - offset = acked_bytes - continue - _LOGGER.warning( - "Unexpected DATA ACK offset %d (expected %d..%d)", - acked_bytes, - offset + 1, - offset + len(segment), - ) - offset += len(segment) + if acked_bytes is None: + _LOGGER.warning("Block stream: DATA was not acknowledged") + return False + offset = acked_bytes # Step 3: Flush (end) - await self._write_mesh(dest, PKT_BLOCK_FLUSH, build_flush_payload(offset)) - ack = await self._wait_ack() - - # Check for completion marker (0xef suffix) - success = len(ack) >= 3 and ack[-1] == 0xEF - if success: - _LOGGER.info("Block stream sent successfully") - else: + completed = await send_and_wait_for_ack( + PKT_BLOCK_FLUSH, build_flush_payload(offset), offset, offset + ) + if completed is None: _LOGGER.warning("Block stream: no completion ACK") - - return success + return False + _LOGGER.info("Block stream sent successfully") + return True async def _receive_response_stream(self) -> bytes | None: """Drain the hub's FLUSH → STREAM* → FLUSH response exchange.""" dest = self._device_id - retries = 0 - - # The next queued control packet is the hub's response FLUSH(0). - await self._wait_ack(timeout=STREAM_TIMEOUT) - await asyncio.sleep(WRITE_DELAY) - await self._write_mesh(dest, PKT_BLOCK_ACK, build_ack_payload(0)) - + # A response begins with FLUSH(0). Keep an incomplete response alive, + # like the official client, rather than treating one missed interval as + # a failed command. try: + _, start_flush = await self._wait_packet( + (PKT_BLOCK_FLUSH,), STREAM_TIMEOUT + ) + if len(start_flush) >= 2 and (start_flush[0] | (start_flush[1] << 8)) != 0: + _LOGGER.warning("Response stream started at non-zero offset") + return None + await asyncio.sleep(WRITE_DELAY) + await self._write_mesh(dest, PKT_BLOCK_ACK, build_ack_payload(0)) + + started = asyncio.get_running_loop().time() + last_activity = started while True: - response_type, payload = await self._wait_response_part() + now = asyncio.get_running_loop().time() + remaining = STREAM_TIMEOUT - (now - started) + if remaining <= 0: + _LOGGER.warning("Timed out waiting for hub response stream") + return None + try: + response_type, payload = await self._wait_response_part( + min(STREAM_REACK_INTERVAL, remaining) + ) + except asyncio.TimeoutError: + # The target may have missed our last ACK. Re-ACK the + # contiguous prefix and retain the receive state. + if self._stream_buffer and ( + asyncio.get_running_loop().time() - last_activity + >= STREAM_REACK_INTERVAL + ): + _LOGGER.debug( + "Re-ACKing incomplete STREAM response at offset %d", + len(self._stream_buffer), + ) + await self._write_mesh( + dest, PKT_BLOCK_ACK, + build_ack_payload(len(self._stream_buffer)), + ) + last_activity = asyncio.get_running_loop().time() + continue + continue + + last_activity = asyncio.get_running_loop().time() if response_type == PKT_BLOCK_STREAM: offset = payload[0] | (payload[1] << 8) if len(payload) >= 2 else None if offset != len(self._stream_buffer): - # The hub retransmits a segment when its ACK was lost. - # Match the official app: retain the contiguous data and - # re-ACK the offset we have already received. - retries += 1 - if retries > STREAM_RETRY_LIMIT: - _LOGGER.warning("Discarding malformed STREAM response") - return None + # Re-ACK the contiguous prefix; the hub will retransmit + # a duplicate or the missing segment. _LOGGER.debug( "Out-of-order STREAM segment at offset %s (expected %d); re-ACKing", offset, @@ -525,9 +579,7 @@ continue if not self._accept_stream_segment(payload): _LOGGER.warning("Discarding malformed STREAM response") return None - retries = 0 - # Android acknowledges every segment by its cumulative stream - # offset; wait for the next segment or final FLUSH afterwards. + # The official app acknowledges every contiguous segment. await self._write_mesh( dest, PKT_BLOCK_ACK, build_ack_payload(len(self._stream_buffer)) ) @@ -539,10 +591,6 @@ if len(payload) >= 2 else len(self._stream_buffer) ) if flush_count != len(self._stream_buffer): - retries += 1 - if retries > STREAM_RETRY_LIMIT: - _LOGGER.warning("Discarding malformed STREAM response") - return None _LOGGER.debug( "STREAM end FLUSH at offset %d (expected %d); re-ACKing", flush_count, @@ -560,26 +608,12 @@ dest, PKT_BLOCK_ACK, build_ack_payload(flush_count, end=True) ) return bytes(self._stream_buffer) except asyncio.TimeoutError: + _LOGGER.warning("Timed out waiting for hub response stream") return None - async def _wait_response_part(self) -> tuple[int, bytes]: + async def _wait_response_part(self, timeout: float = STREAM_TIMEOUT) -> tuple[int, bytes]: """Return whichever arrives next: a stream segment or final FLUSH.""" - ack_task = asyncio.create_task(self._ack_queue.get()) - stream_task = asyncio.create_task(self._stream_queue.get()) - done, pending = await asyncio.wait( - (ack_task, stream_task), timeout=STREAM_TIMEOUT, - return_when=asyncio.FIRST_COMPLETED, - ) - for task in pending: - task.cancel() - if pending: - await asyncio.gather(*pending, return_exceptions=True) - if not done: - _LOGGER.warning("Timed out waiting for hub response stream") - raise asyncio.TimeoutError - if stream_task in done: - return PKT_BLOCK_STREAM, stream_task.result() - return PKT_BLOCK_FLUSH, ack_task.result() + return await self._wait_packet((PKT_BLOCK_STREAM, PKT_BLOCK_FLUSH), timeout) async def _send_acknowledged_command( self, opcode: int, command_data: bytes @@ -611,6 +645,10 @@ Returns: Dict mapping output_id → ZoneState for each zone. """ + # Parse STREAM response: + # [offset(2), cmd_type=0x02, opcode(2), vendorId, productId, firmware, status, + # numZones, then per zone(7 bytes): outputId, outputType, capMask, outputMode, + # dtd1, dtd2, outputState] stream = await self._send_acknowledged_command( OPCODE_GET_INFO_DEVICES, b"" ) @@ -618,10 +656,6 @@ if stream is None: _LOGGER.warning("Failed to query zone states") return self._zone_states - # Parse STREAM response: - # [offset(2), cmd_type=0x02, opcode(2), vendorId, productId, firmware, status, - # numZones, then per zone(7 bytes): outputId, outputType, capMask, outputMode, - # dtd1, dtd2, outputState] if len(stream) < 10: _LOGGER.warning("STREAM too short: %d bytes", len(stream)) return self._zone_states @@ -629,7 +663,6 @@ offset = stream[0] | (stream[1] << 8) cmd_type = stream[2] opcode = stream[3] | (stream[4] << 8) - if offset != 0 or cmd_type != 0x02 or opcode != OPCODE_GET_INFO_DEVICES: _LOGGER.warning("Unexpected opcode in STREAM: 0x%04X", opcode) return self._zone_states @@ -641,8 +674,12 @@ # status = stream[8] num_zones = stream[9] expected_length = 10 + num_zones * 7 if len(stream) != expected_length: - _LOGGER.warning("Malformed GET_INFO_DEVICES response: %d bytes, expected %d", len(stream), expected_length) + _LOGGER.warning( + "Malformed GET_INFO_DEVICES response: %d bytes, expected %d", + len(stream), expected_length, + ) return self._zone_states + zone_records: list[ZoneState] = [] zone_start = 10 diff --git a/tests/test_hub.py b/tests/test_hub.py index 8beb70785f8bffa2e34dc04e9597dfc5006664bd..af09b8b73b662d6211485bba6929a8831525e86a 100644 --- a/tests/test_hub.py +++ b/tests/test_hub.py @@ -7,6 +7,7 @@ from inlite_ble.protocol import ( CHAR_CONTINUATION_UUID, CHAR_WRITE_UUID, OPCODE_IDENTIFY, + PKT_BLOCK_ACK, PKT_BLOCK_DATA_BLK, PKT_BLOCK_FLUSH, PKT_BLOCK_STREAM, @@ -110,10 +111,29 @@ "data": b"\x00\x00", } hub.handle_notification(CHAR_WRITE_UUID, bytearray(b"packet")) await asyncio.sleep(0) - return await hub._wait_ack() + _, payload = await hub._wait_packet((PKT_BLOCK_FLUSH,), 0.1) + return payload assert asyncio.run(run()) == b"\x00\x00" + def test_response_packets_preserve_cross_type_arrival_order(self) -> None: + async def run() -> list[tuple[int, bytes]]: + hub = InliteHub(device_id=0x1234, passphrase="test") + hub._packet_queue.put_nowait((PKT_BLOCK_ACK, b"\x05\x00")) + hub._packet_queue.put_nowait((PKT_BLOCK_FLUSH, b"\x00\x00")) + hub._packet_queue.put_nowait((PKT_BLOCK_STREAM, b"\x00\x00abc")) + return [ + await hub._wait_packet((PKT_BLOCK_ACK,), 0.1), + await hub._wait_packet((PKT_BLOCK_FLUSH,), 0.1), + await hub._wait_packet((PKT_BLOCK_STREAM,), 0.1), + ] + + assert asyncio.run(run()) == [ + (PKT_BLOCK_ACK, b"\x05\x00"), + (PKT_BLOCK_FLUSH, b"\x00\x00"), + (PKT_BLOCK_STREAM, b"\x00\x00abc"), + ] + def test_identify_is_fire_and_forget(self) -> None: async def run() -> list[tuple[int, int, bytes]]: hub = InliteHub(device_id=0x1234, passphrase="test") @@ -138,7 +158,7 @@ async def write_mesh(_dest: int, packet_type: int, data: bytes) -> None: writes.append((packet_type, data)) hub._write_mesh = write_mesh # type: ignore[method-assign] - hub._ack_queue.put_nowait(b"\x00\x00") + hub._packet_queue.put_nowait((PKT_BLOCK_FLUSH, b"\x00\x00")) parts = iter( [ @@ -147,7 +167,7 @@ (PKT_BLOCK_FLUSH, b"\x03\x00"), ] ) - async def wait_response_part() -> tuple[int, bytes]: + async def wait_response_part(_timeout: float = 0) -> tuple[int, bytes]: return next(parts) hub._wait_response_part = wait_response_part # type: ignore[method-assign] @@ -167,7 +187,7 @@ async def write_mesh(_dest: int, packet_type: int, data: bytes) -> None: writes.append((packet_type, data)) hub._write_mesh = write_mesh # type: ignore[method-assign] - hub._ack_queue.put_nowait(b"\x00\x00") + hub._packet_queue.put_nowait((PKT_BLOCK_FLUSH, b"\x00\x00")) parts = iter( [ (PKT_BLOCK_STREAM, b"\x00\x00abc"), @@ -176,7 +196,7 @@ (PKT_BLOCK_FLUSH, b"\x05\x00"), ] ) - async def wait_response_part() -> tuple[int, bytes]: + async def wait_response_part(_timeout: float = 0) -> tuple[int, bytes]: return next(parts) hub._wait_response_part = wait_response_part # type: ignore[method-assign] @@ -200,7 +220,7 @@ async def write_mesh(_dest: int, packet_type: int, data: bytes) -> None: writes.append((packet_type, data)) hub._write_mesh = write_mesh # type: ignore[method-assign] - hub._ack_queue.put_nowait(b"\x00\x00") + hub._packet_queue.put_nowait((PKT_BLOCK_FLUSH, b"\x00\x00")) parts = iter( [ (PKT_BLOCK_STREAM, b"\x00\x00abc"), @@ -209,7 +229,7 @@ (PKT_BLOCK_FLUSH, b"\x03\x00"), ] ) - async def wait_response_part() -> tuple[int, bytes]: + async def wait_response_part(_timeout: float = 0) -> tuple[int, bytes]: return next(parts) hub._wait_response_part = wait_response_part # type: ignore[method-assign]