diff --git a/CHANGELOG.md b/CHANGELOG.md index 6b93ae8..de428d0 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,31 @@ # Changelog +## Unreleased + +### Features + +* partial-region refresh over the PIPE_WRITE sliding window (0x0080 flags bit1 + `PIPE_FLAG_PARTIAL` + 12-byte LE geometry; ACK flags bit1 confirms). Small mono + partial updates now get the pipe's throughput/robustness and the e-paper partial + waveform, with the legacy 0x76 path retained as fallback. New START NACK codes + 0x05 (etag mismatch) / 0x06 (partial unsupported) / 0x07 (rect invalid) drive the + fallback ladder. No new capability bit — eligibility is inferred from the existing + `supports_pipe_write` + `partial_update_support` signals. Works unchanged under an + encrypted session. +* PIPE_WRITE (full and partial) is now hard-gated on the device config advertising + `transmission_modes` bit 0x10 (`supports_pipe_write`): a device whose config lacks + the bit never receives a 0x0080 probe and stays on the legacy path. When the gate + passes, the 0x0080 negotiation remains authoritative for transfer parameters. + Previously the bit was advisory and the probe alone decided. +* With the config gate in place, the 0x0080 START wait is no longer a 2 s discovery + probe: `TIMEOUT_PIPE_PROBE` is replaced by `TIMEOUT_PIPE_START` (30 s, a normal + command timeout sized for ESP32's response-queue flush landing after slow color + panel bring-up). Silence still falls back to legacy (stale config bit), cached per + connection. +* Hardening: the sliding-window sender and END_ACK waiter now bound pathological + ACK streams that never make progress (previously loops without a progress + guarantee, reachable only with buggy/hostile firmware). + ## [7.11.2](https://github.com/OpenDisplay/py-opendisplay/compare/v7.11.1...v7.11.2) (2026-07-06) diff --git a/FINDINGS.md b/FINDINGS.md index c5f20c9..2a0f625 100644 --- a/FINDINGS.md +++ b/FINDINGS.md @@ -259,6 +259,21 @@ of **8** regardless of bpp (`display_service.cpp:1534`). Python: Fix: restrict partial to `ColorScheme.MONO`, align to 8 pixels, and treat all pre-refresh 0x76/0x71 NACKs as fallback-to-full. +**Update (pipe-partial):** partial-region refresh now rides the sliding window when the +device advertises `supports_pipe_write` and `max_queue_size > 1`. `_maybe_upload_partial` +negotiates an extended `0x0080` START (flags bit1 `PIPE_FLAG_PARTIAL` + 12-byte LE +`[old_etag][x][y][w][h]` geometry, `total_size = plane_size*2`); the device confirms with +ACK flags bit1. New START NACK codes gate the fallback ladder: `0x05 ETAG_MISMATCH` → skip +0x76, go full (device already cleared its etag); `0x06 PARTIAL_UNSUPPORTED`/`0x07 +RECT_INVALID` → go full (0x76 would fail identically; 0x06 caches a per-connection +negative); `0x02` on a partial request (after one uncompressed-still-partial retry) or an +ACK without bit1 → disable pipe-partial for the connection and fall back to legacy 0x76; +silence/garble → 0x76. Partial transfers never auto-complete (firmware waits for the +explicit `0x0082` END which alone carries the refresh selector `2` + new_etag), so the +sender uses the same explicit-END contract compressed transfers use. Encryption parity +holds: the 24-byte plaintext START fits one CCM frame, data frames size to 212 B @ frame +244, and the 0x76 fallback rung still caps at `ENCRYPTED_CHUNK_SIZE`. + ### M2. 🟠 [FW-interplay] Etag never committed on uncompressed full uploads — partial mode never engages, and a stale-etag hazard exists Uncompressed uploads always finish via firmware auto-END at the exact byte count diff --git a/src/opendisplay/__init__.py b/src/opendisplay/__init__.py index e0085d2..2c52cb0 100644 --- a/src/opendisplay/__init__.py +++ b/src/opendisplay/__init__.py @@ -22,6 +22,7 @@ OTAError, OTANotSupportedError, ProtocolError, + RefreshTimeoutError, TruncatedConfigError, ) from .landing import LANDING_URL_PREFIX, build_landing_payload, build_landing_url @@ -101,6 +102,7 @@ "BLEConnectionError", "BLETimeoutError", "ProtocolError", + "RefreshTimeoutError", "ConfigParseError", "TruncatedConfigError", "InvalidResponseError", diff --git a/src/opendisplay/device.py b/src/opendisplay/device.py index 1d400e2..0918984 100644 --- a/src/opendisplay/device.py +++ b/src/opendisplay/device.py @@ -10,7 +10,7 @@ import time from collections.abc import AsyncIterator, Awaitable, Callable from contextlib import asynccontextmanager -from typing import TYPE_CHECKING, TypeVar +from typing import TYPE_CHECKING, TypeVar, cast from epaper_dithering import ColorScheme, DitherMode, dither_image from PIL import Image @@ -36,6 +36,7 @@ zlib_window_bits, ) from .exceptions import ( + AuthenticationError, AuthenticationFailedError, AuthenticationRequiredError, AuthenticationSessionExistsError, @@ -44,6 +45,7 @@ IntegrityCheckError, InvalidResponseError, ProtocolError, + RefreshTimeoutError, TruncatedConfigError, ) from .landing import build_landing_url @@ -55,6 +57,7 @@ from .models.led_flash import LedFlashConfig from .partial import ( PARTIAL_FLAG_COMPRESSED, + PartialRegion, PartialState, _generate_etag, build_partial_logical_stream, @@ -64,10 +67,17 @@ ) from .protocol import ( CHUNK_SIZE, + DEFAULT_MAX_FRAME, ENCRYPTED_CHUNK_SIZE, MAX_COMPRESSED_SIZE, + MAX_PTO, MAX_START_PAYLOAD, + PIPE_FLAG_PARTIAL, + PIPE_FRAME_OVERHEAD, + TIMEOUT_PIPE_START, CommandCode, + PipeParams, + PipePartialRequest, build_authenticate_step1, build_authenticate_step2, build_buzzer_activate_command, @@ -80,16 +90,32 @@ build_direct_write_start_uncompressed, build_enter_dfu_command, build_led_activate_command, + build_pipe_write_data_command, + build_pipe_write_end_command, + build_pipe_write_start_command, build_read_config_command, build_read_fw_version_command, build_reboot_command, build_write_config_command, + classify_pipe_frame, parse_config_response, parse_firmware_version, + parse_pipe_data_ack, + parse_pipe_data_nack, + parse_pipe_start_response, serialize_config, + unpack_ack_ranges, validate_ack_response, ) from .protocol.responses import ( + PIPE_FRAME_ACK, + PIPE_FRAME_END_ACK, + PIPE_FRAME_END_NACK, + PIPE_FRAME_NACK, + PIPE_START_NACK_COMPRESSION, + PIPE_START_NACK_ETAG_MISMATCH, + PIPE_START_NACK_PARTIAL_UNSUPPORTED, + PIPE_START_NACK_RECT_INVALID, check_response_type, is_compressed_failure_frame, parse_authenticate_challenge, @@ -104,6 +130,24 @@ _LOGGER = logging.getLogger(__name__) + +class _PipePartialEtagMismatch(Exception): + """Internal: the device NACKed a pipe-partial START with 0x05 (etag mismatch). + + The device has already cleared its displayed_etag, so a 0x76 retry would + mismatch again — the caller clears PartialState and goes straight to full. + """ + + +class _PipePartialRejected(Exception): + """Internal: the device NACKed a pipe-partial START with 0x06/0x07. + + 0x06 = partial unsupported (bpp/driver), 0x07 = rect invalid. The 0x76 + fallback runs identical checks and would fail identically, so the caller + clears PartialState and goes straight to full. + """ + + _INDEX_TO_ROTATION: dict[int, Rotation] = { 0: Rotation.ROTATE_0, 1: Rotation.ROTATE_90, @@ -348,7 +392,7 @@ async def wrapper(self: OpenDisplayDevice, *args: object, **kwargs: object) -> _ return wrapper -class OpenDisplayDevice: +class OpenDisplayDevice: # pylint: disable=too-many-instance-attributes """OpenDisplay BLE e-paper device. Main API for communicating with OpenDisplay BLE tags. @@ -383,6 +427,23 @@ class OpenDisplayDevice: TIMEOUT_COMPRESSED_END_ACK = 90.0 # Compressed END: decompression + full SPI write to IC (~60s on Spectra/ACeP) TIMEOUT_REFRESH = 90.0 # Display refresh (firmware spec: up to 60s) + # PIPE_WRITE per-path progress timeouts (Part 1 §1.4): a compressed chunk lands + # fast; an uncompressed chunk can block bbepWriteData on SPI for the Spectra/ACeP + # ~60s SPI-block budget, so 90s preserves it. + TIMEOUT_PIPE_DATA_COMPRESSED = 5.0 + TIMEOUT_PIPE_DATA_UNCOMPRESSED = 90.0 + # Compressed tail-flush: firmware ACKs only every N_eff accepted frames, so a + # tail of < N_eff unacked frames never earns a cadence ACK on its own. Rather + # than stalling chunk_timeout (5 s) waiting for one, block briefly and then + # dup-probe (resend the oldest unacked chunk) — the duplicate elicits an + # immediate ACK from firmware. Never applied to the uncompressed path, whose + # 90 s budget covers legitimate SPI stalls. + TIMEOUT_PIPE_TAIL_FLUSH = 0.5 + + # Version gate sentinel: None ⇒ version gating disabled, the 0x0080 probe is + # authoritative. Pin a (major, minor) tuple once a firmware release ships PIPE_WRITE. + PIPE_MIN_FW: tuple[int, int] | None = None + def __init__( self, mac_address: str | None = None, @@ -396,6 +457,8 @@ def __init__( use_services_cache: bool = True, use_measured_palettes: bool = True, encryption_key: bytes | None = None, + blocks_per_ack: int = 8, + max_queue_size: int = 16, ): """Initialize OpenDisplay device. @@ -411,6 +474,11 @@ def __init__( use_services_cache: Enable GATT service caching for faster reconnections (default: True) use_measured_palettes: Use measured color palettes when available (default: True) encryption_key: 16-byte AES-128 master key for encrypted devices (optional). + blocks_per_ack: Requested PIPE_WRITE ACK cadence N (blocks per ack), 1..32 + (default: 8). Negotiated down to the device maximum. + max_queue_size: Requested PIPE_WRITE window W (tokens in flight), 1..32 + (default: 16). ``max_queue_size <= 1`` disables sliding-window fast + transfer entirely — legacy stop-and-wait only, no 0x0080 probe. Raises: ValueError: If neither or both mac_address and device_name provided @@ -450,6 +518,17 @@ def __init__( self._command_lock = asyncio.Lock() self._lock_owner: asyncio.Task[object] | None = None + # Sliding-window (PIPE_WRITE) tuning + per-connection capability cache. + self._blocks_per_ack = blocks_per_ack + self._max_queue_size = max_queue_size + self._pipe_params: PipeParams | None = None # active transfer only + self._pipe_probed: bool = False # capability determined this connection + self._pipe_supported: bool = False # probe result (valid iff _pipe_probed) + # Pipe-partial support is inferred, then confirmed by the 0x0080 ACK flags + # bit1: None = unknown, True = confirmed, False = rejected this connection + # (older pipe-capable firmware NACKs the partial flag with 0x02). + self._pipe_partial_supported: bool | None = None + async def __aenter__(self) -> OpenDisplayDevice: """Connect and optionally interrogate device.""" @@ -555,9 +634,30 @@ def _clear_session(self) -> None: self._auth_time = None def _on_ble_disconnect(self) -> None: - """Handle an unexpected BLE drop: forget the (now-dead) session.""" + """Handle an unexpected BLE drop: forget the (now-dead) session and pipe state.""" _LOGGER.debug("Link to %s dropped; clearing session state", self.mac_address) self._clear_session() + # All pipe negotiation/capability state is per-connection (Part 1 §1.1). + self._pipe_probed = False + self._pipe_supported = False + self._pipe_partial_supported = None + self._pipe_params = None + + def _encrypt_frame(self, data: bytes) -> bytes: + """Encrypt one plaintext command frame under the active session. + + Advances the nonce counter by one so every transmission — including a + PIPE_WRITE retransmission — carries a fresh, higher nonce. Returns ``data`` + unchanged when no session is active. Does NOT re-authenticate (callers + handle re-auth once before a stream, never mid-stream). + """ + if self._session_key is not None and self._session_id is not None: + cmd = data[:2] + payload = data[2:] + frame = encrypt_command(self._session_key, self._session_id, self._nonce_counter, cmd, payload) + self._nonce_counter += 1 + return frame + return data async def _write(self, data: bytes, response: bool = True) -> None: """Write a command, encrypting it if an active session exists. @@ -570,14 +670,20 @@ async def _write(self, data: bytes, response: bool = True) -> None: """ if self._session_key is not None and self._session_id is not None: await self._reauthenticate_if_needed() - cmd = data[:2] - payload = data[2:] - encrypted = encrypt_command(self._session_key, self._session_id, self._nonce_counter, cmd, payload) - self._nonce_counter += 1 - await self._conn.write_command(encrypted, response=response) + await self._conn.write_command(self._encrypt_frame(data), response=response) else: await self._conn.write_command(data, response=response) + async def _write_pipe_frame(self, data: bytes, *, response: bool) -> None: + """Encrypt (no re-auth) and write a live PIPE_WRITE stream frame. + + Passes ``drain_stale=False`` so queued sliding-window ACKs are preserved. + Used for every 0x0081 DATA frame (response=False) and the 0x0082 END + (response=True). Re-authentication is intentionally skipped: it runs once + before 0x0080 and never mid-stream (Part 1 §1.6). + """ + await self._conn.write_command(self._encrypt_frame(data), response=response, drain_stale=False) + async def _reauthenticate_if_needed(self) -> None: """Re-authenticate proactively at 90% of session_timeout_seconds.""" if self._encryption_key is None or self._auth_time is None: @@ -1696,6 +1802,71 @@ async def _maybe_upload_partial( new_etag = _generate_etag() _LOGGER.debug("Partial upload: old_etag=0x%08x new_etag=0x%08x", state.etag, new_etag) + # Pipe-first: ride the sliding window when the device advertises PIPE_WRITE + # and pipe-partial has not been disabled this connection. partial_update_support + # != NONE is already guaranteed (compute_partial_region gated on it). Falls + # back to the legacy 0x76 flow below on any None result. + if ( + display.supports_pipe_write + and self._max_queue_size > 1 + and not (self._pipe_probed and not self._pipe_supported) + and self._pipe_partial_supported is not False + ): + total_size = len(logical_stream) + try: + params = await self._negotiate_pipe_partial(use_compression, total_size, state.etag, region) + except _PipePartialEtagMismatch: + # Device already cleared its etag — a 0x76 retry would mismatch + # again, so skip straight to a full upload. + _LOGGER.info("pipe-partial etag mismatch (0x05); clearing state, falling back to full") + state.etag = 0 + state.last_image = None + return "fallback_full" + except _PipePartialRejected as exc: + # 0x76 runs identical bpp/rect checks and would fail identically. + _LOGGER.info("pipe-partial rejected (%s); clearing state, falling back to full", exc) + state.etag = 0 + state.last_image = None + return "fallback_full" + if params is not None: + # Negotiation may have downgraded compressed→uncompressed (NACK 0x02). + payload = compressed_stream if params.compressed else logical_stream + try: + await self._run_pipe_upload(payload, params, RefreshMode.PARTIAL, progress_callback, new_etag) + except RefreshTimeoutError: + # 0x74 after END_ACK: firmware already cleared its etag on + # the failed refresh; re-raise (parity with the 0x76 path). + # Note BLETimeoutError on the post-DATA reads deliberately + # propagates too (not caught below): the END may already have + # committed on the device, so falling back to full and + # re-baselining state on an unknown outcome would be unsafe. + raise + except (AuthenticationError, IntegrityCheckError): + # An auth/session failure or a decrypt-integrity rejection is NOT + # a clean, safe-to-retry protocol abort. Auth errors MUST surface + # so the caller (e.g. Home Assistant) can trigger reauth instead + # of silently retrying; an integrity failure signals an out-of-sync + # encrypted channel that a full upload over the same session would + # likely hit again. Masking either as a full-upload fallback would + # hide it, so re-raise. Only the genuine protocol NACKs below are + # safe to recover from with a full upload. + raise + except ProtocolError as exc: + # Mid-stream NACK / MAX_RETX / END NACK before any refresh — the + # transfer aborted cleanly, so a full upload is safe. + _LOGGER.info("pipe-partial upload failed (%s); clearing state, falling back to full", exc) + state.etag = 0 + state.last_image = None + return "fallback_full" + # Success — commit partial state (mirrors the 0x76 success path). + state.etag = new_etag + state.last_image = region.new_palette + state.width = region.width + state.height = region.height + state.bytes_per_pixel = 1 + return "success" + # params is None → fall through to the legacy 0x76 flow unchanged. + # Start partial upload (0x76), stream remaining 0x71 chunks, and finish # with partial refresh. max_start = ENCRYPTED_CHUNK_SIZE if self._session_key is not None else MAX_START_PAYLOAD @@ -1743,7 +1914,7 @@ async def _maybe_upload_partial( response = await self._read(self.TIMEOUT_REFRESH) command, _ = check_response_type(response) if command == CommandCode.DIRECT_WRITE_REFRESH_TIMEOUT: - raise ProtocolError("Display refresh timed out (device sent 0x74)") + raise RefreshTimeoutError("Display refresh timed out (device sent 0x74)") if command != CommandCode.DIRECT_WRITE_REFRESH_COMPLETE: raise ProtocolError(f"Unexpected response waiting for refresh: {command.name} (0x{command:04x})") @@ -1780,6 +1951,37 @@ async def _execute_upload( Raises: ProtocolError: If upload fails """ + # 0. Sliding-window (PIPE_WRITE) attempt. Gated on the device config + # advertising pipe support (transmission_modes bit 0x10) — a device whose + # config lacks the bit never sees a 0x0080 probe. Also skipped when + # disabled (max_queue_size <= 1) or when this connection already probed + # negative. The 0x0080 negotiation below remains authoritative for the + # transfer parameters when the gate passes. + display_cfg = self._config.displays[0] if (self._config and self._config.displays) else None + pipe_eligible = ( + bool(display_cfg and display_cfg.supports_pipe_write) + and self._max_queue_size > 1 + and not (self._pipe_probed and not self._pipe_supported) + ) + if pipe_eligible: + total_size = len(image_data) + params = await self._negotiate_pipe(use_compression, total_size) + self._pipe_probed = True + if params is not None: + self._pipe_supported = True + # Negotiation may have downgraded compressed→uncompressed (NACK 0x02). + if params.compressed and compressed_data is not None: + payload = compressed_data + else: + payload = image_data + if params.compressed: + # Contract: use_compression implies compressed_data; guard anyway. + params = PipeParams(params.window, params.ack_every, params.max_frame, params.selective, False) + return await self._run_pipe_upload(payload, params, refresh_mode, progress_callback, new_etag) + self._pipe_supported = False + _LOGGER.info("PIPE_WRITE unavailable on %s; using legacy direct-write flow", self.mac_address) + # Fall through to the untouched legacy flow below. + # 1. Send START command (different for each protocol) if use_compression: if uncompressed_size is None or compressed_data is None: @@ -1852,7 +2054,7 @@ async def _execute_upload( response = await self._read(self.TIMEOUT_REFRESH) command, _ = check_response_type(response) if command == CommandCode.DIRECT_WRITE_REFRESH_TIMEOUT: - raise ProtocolError("Display refresh timed out (device sent 0x74)") + raise RefreshTimeoutError("Display refresh timed out (device sent 0x74)") if command != CommandCode.DIRECT_WRITE_REFRESH_COMPLETE: raise ProtocolError(f"Unexpected response waiting for refresh: {command.name} (0x{command:04x})") _LOGGER.info("Display refresh complete") @@ -1924,6 +2126,472 @@ async def _send_data_chunks( _LOGGER.debug("All data chunks sent (%d chunks total)", chunks_sent) return False + # ─── PIPE_WRITE (sliding-window) upload ────────────────────────────────── + + async def _negotiate_pipe( + self, compressed: bool, total_size: int, _retry_uncompressed: bool = True + ) -> PipeParams | None: + """Probe + negotiate a sliding-window transfer via 0x0080. + + Sends PIPE_WRITE_START and waits ``TIMEOUT_PIPE_START`` for the response. + Attempts are config-gated (transmission_modes bit 0x10), so this is a plain + command timeout rather than a discovery probe. Silence (stale config bit on + pipe-less firmware) or an unrecoverable NACK returns + None → the caller falls back to the legacy 0x70 flow. A NACK err 0x02 + (compression unsupported) on a compressed request retries 0x0080 once + uncompressed before giving up. + + Returns: + PipeParams (effective, post-min-rule) on success, else None. + """ + req_frame = DEFAULT_MAX_FRAME # HA GATT ceiling; also our client_max_frame + # The 0x0080 is the single pre-stream write; _write re-authenticates here + # (once, never again mid-stream). + await self._write( + build_pipe_write_start_command( + compressed, self._max_queue_size, self._blocks_per_ack, req_frame, total_size + ) + ) + try: + resp = await self._read(TIMEOUT_PIPE_START) + except BLETimeoutError: + _LOGGER.debug("No 0x0080 response within %.1fs; firmware lacks PIPE_WRITE", TIMEOUT_PIPE_START) + return None + + try: + ok, payload = parse_pipe_start_response(resp) + except InvalidResponseError as err: + _LOGGER.debug("Garbled 0x0080 response (%s); falling back to legacy", err) + return None + + if not ok: + err_code = cast(int, payload) + if err_code == PIPE_START_NACK_COMPRESSION and compressed and _retry_uncompressed: + _LOGGER.info("Device rejected compressed PIPE_WRITE (err 0x02); retrying uncompressed") + return await self._negotiate_pipe(False, total_size, _retry_uncompressed=False) + _LOGGER.info("PIPE_WRITE START NACK (err 0x%02x); falling back to legacy", err_code) + return None + + ver, dev_max_window, dev_max_ack_every, dev_max_frame, flags = cast("tuple[int, int, int, int, int]", payload) + # Min-rule (Part 1 §1.1) — computed identically to firmware. + w_eff = max(1, min(self._max_queue_size, dev_max_window, 32)) + n_eff = max(1, min(self._blocks_per_ack, dev_max_ack_every, w_eff)) + frame_eff = min(req_frame, dev_max_frame) + selective = bool(flags & 0x01) + params = PipeParams(w_eff, n_eff, frame_eff, selective, compressed) + _LOGGER.info( + "PIPE_WRITE negotiated: W=%d N=%d frame=%d selective=%s compressed=%s (dev max %d/%d/%d, ver %d)", + w_eff, + n_eff, + frame_eff, + selective, + compressed, + dev_max_window, + dev_max_ack_every, + dev_max_frame, + ver, + ) + return params + + async def _negotiate_pipe_partial( + self, + compressed: bool, + total_size: int, + old_etag: int, + region: PartialRegion, + _retry_uncompressed: bool = True, + ) -> PipeParams | None: + """Probe + negotiate a partial-region sliding-window transfer via 0x0080. + + Sends an extended PIPE_WRITE_START (flags bit1 + 12-byte geometry) and + interprets the response per Part 1 §1.2. A partial 0x0080 is a valid pipe + probe, so this also updates ``_pipe_probed`` / ``_pipe_supported``. + + Returns: + PipeParams(partial=True) on an ACK confirming partial (flags bit1), + or None when the caller should fall back to the legacy 0x76 flow + (silence / garble / other NACK / partial-flag rejected / bit1 clear). + + Raises: + _PipePartialEtagMismatch: NACK 0x05 — caller skips 0x76, goes full. + _PipePartialRejected: NACK 0x06/0x07 — caller skips 0x76, goes full. + """ + req_frame = DEFAULT_MAX_FRAME # HA GATT ceiling; also our client_max_frame + partial = PipePartialRequest(old_etag=old_etag, x=region.rx, y=region.ry, w=region.rw, h=region.rh) + # The 0x0080 is the single pre-stream write; _write re-authenticates here + # (once, never again mid-stream). + await self._write( + build_pipe_write_start_command( + compressed, + self._max_queue_size, + self._blocks_per_ack, + req_frame, + total_size, + partial=partial, + ) + ) + try: + resp = await self._read(TIMEOUT_PIPE_START) + except BLETimeoutError: + _LOGGER.debug("No 0x0080 partial response within %.1fs; firmware lacks PIPE_WRITE", TIMEOUT_PIPE_START) + self._pipe_probed = True + self._pipe_supported = False + return None + + try: + ok, payload = parse_pipe_start_response(resp) + except InvalidResponseError as err: + _LOGGER.debug("Garbled 0x0080 partial response (%s); falling back to legacy", err) + self._pipe_probed = True + self._pipe_supported = False + return None + + if not ok: + err_code = cast(int, payload) + # The device answered 0x0080, so pipe write itself is supported. + self._pipe_probed = True + self._pipe_supported = True + if err_code == PIPE_START_NACK_COMPRESSION and compressed and _retry_uncompressed: + _LOGGER.info("Device rejected compressed pipe-partial (err 0x02); retrying uncompressed still-partial") + return await self._negotiate_pipe_partial( + False, total_size, old_etag, region, _retry_uncompressed=False + ) + if err_code == PIPE_START_NACK_COMPRESSION: + # A second 0x02 (or 0x02 on an already-uncompressed request) means + # the partial flag bit itself is unknown — older pipe-capable + # firmware. Cache the negative so we never re-send a partial 0x0080 + # this connection. + _LOGGER.info("Device rejected the pipe-partial flag (err 0x02); disabling pipe-partial this connection") + self._pipe_partial_supported = False + return None + if err_code == PIPE_START_NACK_ETAG_MISMATCH: + raise _PipePartialEtagMismatch("Device rejected pipe-partial START: displayed etag mismatch (0x05)") + if err_code in (PIPE_START_NACK_PARTIAL_UNSUPPORTED, PIPE_START_NACK_RECT_INVALID): + if err_code == PIPE_START_NACK_PARTIAL_UNSUPPORTED: + # bpp/driver can't do partial at all — never retry this connection. + self._pipe_partial_supported = False + raise _PipePartialRejected(f"Device rejected pipe-partial START (err 0x{err_code:02x})") + _LOGGER.info("pipe-partial START NACK (err 0x%02x); falling back to legacy", err_code) + return None + + ver, dev_max_window, dev_max_ack_every, dev_max_frame, flags = cast("tuple[int, int, int, int, int]", payload) + # A valid ACK is a valid pipe probe. + self._pipe_probed = True + self._pipe_supported = True + if not flags & PIPE_FLAG_PARTIAL: + # Requested partial but the device did not confirm bit1 → older + # pipe-capable firmware. Abandon the pipe attempt; the subsequent 0x76 + # START resets the orphaned firmware session (Part 1 §1.2). + _LOGGER.info("Device ACKed 0x0080 without the partial bit; pipe-partial unsupported this connection") + self._pipe_partial_supported = False + return None + self._pipe_partial_supported = True + # Min-rule (Part 1 §1.1) — identical to _negotiate_pipe. + w_eff = max(1, min(self._max_queue_size, dev_max_window, 32)) + n_eff = max(1, min(self._blocks_per_ack, dev_max_ack_every, w_eff)) + frame_eff = min(req_frame, dev_max_frame) + selective = bool(flags & 0x01) + params = PipeParams(w_eff, n_eff, frame_eff, selective, compressed, partial=True) + _LOGGER.info( + "pipe-partial negotiated: W=%d N=%d frame=%d selective=%s compressed=%s (dev max %d/%d/%d, ver %d)", + w_eff, + n_eff, + frame_eff, + selective, + compressed, + dev_max_window, + dev_max_ack_every, + dev_max_frame, + ver, + ) + return params + + def _pipe_data_size(self, frame_eff: int) -> int: + """Chunk data capacity for a pipe frame at ``frame_eff`` bytes. + + Encrypted: frame_eff - CCM envelope (31) - seq (1) = 212 @ 244. + Plaintext: frame_eff - PIPE_FRAME_OVERHEAD (cmd 2 + seq 1) = 241 @ 244. + """ + if self._session_key is not None: + return frame_eff - 31 - 1 + return frame_eff - PIPE_FRAME_OVERHEAD + + async def _run_pipe_upload( + self, + payload: bytes, + params: PipeParams, + refresh_mode: RefreshMode, + progress_callback: Callable[[int, int], None] | None, + new_etag: int | None, + ) -> bool: + """Split, stream, END, and await refresh for a sliding-window transfer. + + Returns True if ``new_etag`` was committed via END-with-etag, False if the + firmware auto-completed the upload (no etag committed). + """ + size = self._pipe_data_size(params.max_frame) + if size < 1: + raise ProtocolError(f"Negotiated pipe frame {params.max_frame} too small for a data byte") + # Always keep at least one frame so the receiver's total check + END + # handshake run even for an empty payload (mirrors legacy START/END). + chunks = [payload[i : i + size] for i in range(0, len(payload), size)] or [b""] + if params.partial and new_etag is None: + # An etag-less partial END would make firmware commit displayed_etag=0 + # on a successful refresh, silently desyncing the client PartialState. + raise ProtocolError("PIPE_WRITE partial transfer requires a new_etag for the END commit") + self._pipe_params = params + chunk_timeout = self.TIMEOUT_PIPE_DATA_COMPRESSED if params.compressed else self.TIMEOUT_PIPE_DATA_UNCOMPRESSED + + try: + auto_completed = await self._send_pipe_chunks(chunks, params, chunk_timeout, progress_callback) + # Uncompressed full-frame transfers ALWAYS auto-complete (firmware + # resets pipe state and sends an unsolicited END_ACK once total_size is + # reached), so the explicit END path below is skipped there. Compressed + # and partial transfers use the explicit END (partial firmware never + # auto-completes — Part 1 §1.5). Sending an END after auto-complete + # would be NACKed and desync etag accounting. + if params.partial and auto_completed: + # Partial firmware must never auto-complete; an unsolicited END_ACK + # here is a contract violation (would have refreshed FULL, not + # REFRESH_PARTIAL, with no committed etag). + raise ProtocolError("PIPE_WRITE partial transfer auto-completed unexpectedly (unsolicited END_ACK)") + if not auto_completed: + await self._await_pipe_end_ack(chunks, refresh_mode, new_etag, params) + + # Shared refresh wait — identical to the legacy _execute_upload tail for + # both the auto-complete and explicit-END paths. + _LOGGER.debug("Display refresh started, waiting for completion...") + response = await self._read(self.TIMEOUT_REFRESH) + command, _ = check_response_type(response) + if command == CommandCode.DIRECT_WRITE_REFRESH_TIMEOUT: + raise RefreshTimeoutError("Display refresh timed out (device sent 0x74)") + if command != CommandCode.DIRECT_WRITE_REFRESH_COMPLETE: + raise ProtocolError(f"Unexpected response waiting for refresh: {command.name} (0x{command:04x})") + _LOGGER.info("Display refresh complete (pipe)") + finally: + self._pipe_params = None + + return not auto_completed and new_etag is not None + + async def _send_pipe_chunks( # pylint: disable=too-many-locals,too-many-branches,too-many-statements + self, + chunks: list[bytes], + params: PipeParams, + chunk_timeout: float, + progress_callback: Callable[[int, int], None] | None = None, + ) -> bool: + """QUIC-style selective-repeat sender for PIPE_WRITE data frames. + + Keeps up to ``params.window`` frames in flight (span-based tokens), refunds + tokens on ACK, retransmits only missing chunks (selective repeat) — or + rewinds when the receiver does not buffer out-of-order (bit0 clear). + + Completion contract is set by ``explicit_end = params.compressed or + params.partial``: + - explicit-END path (compressed OR partial): returns once every chunk is + acked; the caller sends the explicit 0x0082 END. Compressed firmware can + only verify totals at zlib flush; partial firmware never auto-completes + (Part 1 §1.5) because only the 0x0082 carries the refresh mode + new_etag. + A tail of < N_eff unacked frames never earns a cadence ACK, so the tail + wait uses TIMEOUT_PIPE_TAIL_FLUSH and a dup-probe instead of stalling + on chunk_timeout. + - auto-complete path (uncompressed full-frame only): the client sends + exactly total_size bytes, so firmware ALWAYS auto-completes — flush-ACK, + then an unsolicited {0x00,0x82}, resetting its pipe state before any + explicit END could arrive. This sender therefore keeps reading past the + flush-ACK until that END_ACK and returns True (mirroring legacy + _send_data_chunks' 0x72-in-place-of-0x71 handling); the caller must NOT + send END. + + Returns: + True if the device auto-completed (unsolicited END_ACK — only the + uncompressed full-frame path), False on normal explicit-END completion + (compressed or partial; caller sends END). + + Raises: + ProtocolError: On a fatal NACK, MAX_RETX/MAX_PTO exhaustion, a missing + auto-complete END_ACK, or an unexpected frame. + """ + n = len(chunks) + window = params.window + # Partial transfers never auto-complete (Part 1 §1.5), so they use the + # same explicit-END completion contract compressed transfers use today. + explicit_end = params.compressed or params.partial + max_retx = 3 * window + acked: set[int] = set() + window_base = 0 # lowest unacked + next_to_send = 0 + pending_retx: dict[int, int] = {} # missing idx → ACKs seen since last (re)transmit + retx_count = 0 + pto_count = 0 + stall_acks = 0 # consecutive ACKs that neither advance window_base nor expose a hole + bytes_total = sum(len(c) for c in chunks) + bytes_sent_hw = 0 + + async def _send(idx: int) -> None: + nonlocal bytes_sent_hw + await self._write_pipe_frame(build_pipe_write_data_command(idx % 256, chunks[idx]), response=False) + + while True: + # 1. Send new chunks while span-tokens are available. + while next_to_send < n and (next_to_send - window_base) < window: + await _send(next_to_send) + bytes_sent_hw += len(chunks[next_to_send]) + if progress_callback is not None: + progress_callback(min(bytes_sent_hw, bytes_total), bytes_total) + next_to_send += 1 + + # Explicit-END (compressed or partial): all acked → done, caller sends + # the explicit END. Uncompressed full-frame: keep reading for the + # unsolicited auto-complete END_ACK. + if window_base >= n and explicit_end: + break + + # 2. Block for an ACK (credit exhausted or tail pending). Compressed + # tail (< N_eff unacked frames, no holes) will never earn a cadence + # ACK, so wait only briefly before dup-probing — never send END with + # unacked chunks (a genuinely lost tail chunk must be repaired by + # retransmit, not surface as a fatal END NACK at zlib flush). + tail_flush = ( + explicit_end + and next_to_send >= n + and 0 < n - window_base < params.ack_every + and not (acked and max(acked) >= window_base) # no known holes + ) + read_timeout = self.TIMEOUT_PIPE_TAIL_FLUSH if tail_flush else chunk_timeout + try: + resp = await self._read(read_timeout) + except BLETimeoutError: + if window_base >= n: + # Uncompressed with everything acked: firmware owes the + # unsolicited END_ACK; there is nothing left to probe. + raise ProtocolError("PIPE_WRITE aborted: auto-complete END_ACK never arrived") from None + pto_count += 1 + if pto_count >= MAX_PTO: + raise ProtocolError(f"PIPE_WRITE aborted: no ACK progress after {MAX_PTO} PTO probes") from None + # PTO / tail-flush dup-probe: resend the oldest unacked chunk + # (fresh nonce); a duplicate elicits an immediate ACK. + await _send(window_base) + retx_count += 1 + if retx_count > max_retx: + raise ProtocolError(f"PIPE_WRITE aborted: MAX_RETX ({max_retx}) exceeded (PTO)") from None + continue + + kind = classify_pipe_frame(resp) + if kind == PIPE_FRAME_NACK: + err, _hs, _mask = parse_pipe_data_nack(resp) + raise ProtocolError(f"PIPE_WRITE data NACK err=0x{err:02x} (fatal)") + if kind == PIPE_FRAME_END_ACK: + # Unsolicited auto-complete: the receiver confirms it holds the full + # image (accepted total reached total_size), so it is authoritative + # and terminal — even if a final local ACK was lost. Mirrors legacy + # 0x72 auto-finish; the client sends no explicit END. + if window_base < n: + _LOGGER.debug("PIPE auto-complete END_ACK with %d/%d chunks locally acked", len(acked), n) + return True + if kind != PIPE_FRAME_ACK: + raise ProtocolError(f"Unexpected pipe frame during send: {resp[:8].hex()}") + + # 3. Process the ACK — refund tokens over the contiguous acked prefix. + pto_count = 0 + highest_seen, ack_mask = parse_pipe_data_ack(resp) + acked |= unpack_ack_ranges(highest_seen, ack_mask, window_base) + prev_base = window_base + while window_base in acked: + pending_retx.pop(window_base, None) + window_base += 1 + # Progress bound: an ACK stream that never advances the cumulative + # point and never exposes a hole would otherwise reset pto_count each + # pass and loop forever (only reachable with pathological firmware — + # real firmware ACKs in response to frames, not unsolicited). + if window_base > prev_base: + stall_acks = 0 + if window_base >= n: + # Loop top: explicit-end (compressed or partial) breaks (caller + # sends END); uncompressed full-frame keeps reading for the + # unsolicited auto-complete END_ACK. + stall_acks += 1 + if stall_acks > max_retx: + raise ProtocolError(f"PIPE_WRITE aborted: {stall_acks} ACKs without auto-complete END_ACK") + continue + + # 4. Loss handling — holes below the highest received are definite losses. + highest_recv = max(acked) if acked else window_base - 1 + missing = [i for i in range(window_base, min(highest_recv, next_to_send)) if i not in acked] + if not missing: + stall_acks += 1 + if stall_acks > max_retx: + raise ProtocolError(f"PIPE_WRITE aborted: {stall_acks} consecutive ACKs without progress") + continue + + if params.selective: + for m in missing: # oldest first + if m not in pending_retx: + do_retx = True # newly detected + else: + pending_retx[m] += 1 # a new ACK still shows it missing + do_retx = pending_retx[m] >= 1 # one implicit RTT of spacing + if do_retx: + await _send(m) + pending_retx[m] = 0 + retx_count += 1 + if retx_count > max_retx: + raise ProtocolError(f"PIPE_WRITE aborted: MAX_RETX ({max_retx}) exceeded") + else: + # bit0 clear: rewind-style recovery (resend from window_base). + next_to_send = window_base + pending_retx.clear() + retx_count += 1 + if retx_count > max_retx: + raise ProtocolError(f"PIPE_WRITE aborted: MAX_RETX ({max_retx}) exceeded (rewind)") + + return False + + async def _await_pipe_end_ack( + self, + chunks: list[bytes], + refresh_mode: RefreshMode, + new_etag: int | None, + params: PipeParams, + ) -> None: + """Send 0x0082 END and wait for the END_ACK (explicit-END transfers). + + Reached by compressed transfers and ALL partial transfers (partial never + auto-completes — the END alone carries the refresh selector + new_etag). + Uncompressed full-frame transfers never reach here: firmware always + auto-completes them (see _send_pipe_chunks), and an END sent after that + reset would be NACKed. Called only after ``_send_pipe_chunks`` has seen + every chunk acked, so the transfer is complete; a trailing tail-flush + PIPE_ACK may precede the END_ACK in the queue and is skipped. An + END_NACK or data NACK aborts (the caller's existing retry-from-scratch + recovers). + """ + del chunks # completeness already guaranteed by the sender loop + end_cmd = build_pipe_write_end_command(refresh_mode.value, new_etag) + await self._write_pipe_frame(end_cmd, response=True) + + end_timeout = self.TIMEOUT_COMPRESSED_END_ACK if params.compressed else self.TIMEOUT_UNCOMPRESSED_END_ACK + stray_acks = 0 + while True: + resp = await self._read(end_timeout) + kind = classify_pipe_frame(resp) + if kind == PIPE_FRAME_END_ACK: + return + if kind == PIPE_FRAME_END_NACK: + raise ProtocolError("PIPE_WRITE END NACK (byte-total mismatch or incomplete transfer)") + if kind == PIPE_FRAME_NACK: + err, _hs, _mask = parse_pipe_data_nack(resp) + raise ProtocolError(f"PIPE_WRITE data NACK during END: err=0x{err:02x}") + if kind == PIPE_FRAME_ACK: + # Tail-flush ACK preceding END_ACK — ignore, keep reading. Bounded: + # firmware sends at most one flush ACK per END; an unending ACK + # stream would otherwise renew end_timeout forever. + stray_acks += 1 + if stray_acks > 32: + raise ProtocolError(f"PIPE_WRITE aborted: {stray_acks} ACK frames while awaiting END_ACK") + continue + raise ProtocolError(f"Unexpected frame awaiting END_ACK: {resp[:8].hex()}") + def _extract_capabilities_from_config(self) -> DeviceCapabilities: """Extract DeviceCapabilities from GlobalConfig. diff --git a/src/opendisplay/exceptions.py b/src/opendisplay/exceptions.py index 41aed14..773d12a 100644 --- a/src/opendisplay/exceptions.py +++ b/src/opendisplay/exceptions.py @@ -27,6 +27,18 @@ class ProtocolError(OpenDisplayError): pass +class RefreshTimeoutError(ProtocolError): + """Display refresh timed out (device sent 0x74) after the image was delivered. + + Distinct from other ProtocolErrors because it fires *after* the transfer + completed: the device already cleared its displayed etag, so callers must + not treat it as a cleanly-aborted transfer (e.g. the pipe-partial path + re-raises instead of falling back to a full upload). + """ + + pass + + class ConfigParseError(ProtocolError): """Failed to parse device configuration.""" diff --git a/src/opendisplay/models/config.py b/src/opendisplay/models/config.py index 3ddcc82..e801e7b 100644 --- a/src/opendisplay/models/config.py +++ b/src/opendisplay/models/config.py @@ -339,6 +339,20 @@ def supports_direct_write(self) -> bool: """Check if display supports direct write mode - bufferless (TRANSMISSION_MODE_DIRECT_WRITE).""" return bool(self.transmission_modes & 0x08) + @property + def supports_pipe_write(self) -> bool: + """Check if display advertises sliding-window PIPE_WRITE (bit 0x10). + + Hard pre-flight gate: uploads only attempt a 0x0080 probe when this bit + is set (full pipe in ``_execute_upload``; pipe-partial additionally in + ``_maybe_upload_partial``). Firmware echoes the STORED TLV, so a device + flashed with pipe-capable firmware but an older config will not set the + bit and stays on the legacy path until its config is updated. When the + gate passes, the 0x0080 negotiation remains authoritative for the + transfer parameters. + """ + return bool(self.transmission_modes & 0x10) + @property def no_boot_text(self) -> bool: """Check if display should suppress boot text (TRANSMISSION_MODE_NO_BOOT_TEXT).""" diff --git a/src/opendisplay/protocol/__init__.py b/src/opendisplay/protocol/__init__.py index eeb2299..fe9bcec 100644 --- a/src/opendisplay/protocol/__init__.py +++ b/src/opendisplay/protocol/__init__.py @@ -2,13 +2,20 @@ from .commands import ( CHUNK_SIZE, + DEFAULT_MAX_FRAME, ENCRYPTED_CHUNK_SIZE, MANUFACTURER_ID, MAX_COMPRESSED_SIZE, + MAX_PTO, MAX_START_PAYLOAD, - PIPELINE_CHUNKS, + PIPE_FLAG_COMPRESSED, + PIPE_FLAG_PARTIAL, + PIPE_FRAME_OVERHEAD, + PIPE_VERSION, SERVICE_UUID, + TIMEOUT_PIPE_START, CommandCode, + PipePartialRequest, build_authenticate_step1, build_authenticate_step2, build_buzzer_activate_command, @@ -21,6 +28,9 @@ build_direct_write_start_uncompressed, build_enter_dfu_command, build_led_activate_command, + build_pipe_write_data_command, + build_pipe_write_end_command, + build_pipe_write_start_command, build_read_config_command, build_read_fw_version_command, build_reboot_command, @@ -32,7 +42,13 @@ serialize_config, ) from .responses import ( + PipeParams, + classify_pipe_frame, parse_firmware_version, + parse_pipe_data_ack, + parse_pipe_data_nack, + parse_pipe_start_response, + unpack_ack_ranges, validate_ack_response, ) @@ -44,9 +60,16 @@ "MANUFACTURER_ID", "CHUNK_SIZE", "ENCRYPTED_CHUNK_SIZE", - "PIPELINE_CHUNKS", "MAX_COMPRESSED_SIZE", "MAX_START_PAYLOAD", + "DEFAULT_MAX_FRAME", + "MAX_PTO", + "PIPE_FLAG_COMPRESSED", + "PIPE_FLAG_PARTIAL", + "PIPE_FRAME_OVERHEAD", + "PIPE_VERSION", + "PipePartialRequest", + "TIMEOUT_PIPE_START", "build_read_config_command", "build_read_fw_version_command", "build_enter_dfu_command", @@ -59,6 +82,9 @@ "build_direct_write_data_command", "build_direct_write_end_command", "build_direct_write_end_with_etag", + "build_pipe_write_start_command", + "build_pipe_write_data_command", + "build_pipe_write_end_command", "build_buzzer_activate_command", "build_led_activate_command", "parse_config_response", @@ -66,4 +92,10 @@ "calculate_config_crc", "validate_ack_response", "parse_firmware_version", + "PipeParams", + "classify_pipe_frame", + "parse_pipe_start_response", + "parse_pipe_data_ack", + "parse_pipe_data_nack", + "unpack_ack_ranges", ] diff --git a/src/opendisplay/protocol/commands.py b/src/opendisplay/protocol/commands.py index 81afefb..e877463 100644 --- a/src/opendisplay/protocol/commands.py +++ b/src/opendisplay/protocol/commands.py @@ -3,6 +3,7 @@ from __future__ import annotations import struct +from dataclasses import dataclass from enum import IntEnum from ..models.buzzer_activate import BuzzerActivateConfig @@ -38,6 +39,11 @@ class CommandCode(IntEnum): ENTER_DFU = 0x0051 # Trigger DFU bootloader mode (nRF only) DEEP_SLEEP = 0x0052 # Enter deep sleep now (ESP32 timer-wake / Silabs EM4; nRF unsupported) + # Sliding-window image transfer (PIPE_WRITE, firmware 2.x+) + PIPE_WRITE_START = 0x0080 # Start + negotiate a sliding-window transfer + PIPE_WRITE_DATA = 0x0081 # Windowed data frame (seq + chunk); also device→host ACK/NACK opcode + PIPE_WRITE_END = 0x0082 # End sliding-window transfer and trigger display refresh + # Protocol constants SERVICE_UUID = "00002446-0000-1000-8000-00805F9B34FB" @@ -49,14 +55,27 @@ class CommandCode(IntEnum): ENCRYPTED_CHUNK_SIZE = 154 # Maximum data bytes per chunk when session is active # Encrypted packet: cmd(2)+nonce(16)+len(1)+data(154)+tag(12) = 185 bytes CONFIG_CHUNK_SIZE = 200 # Maximum config chunk size (verified from firmware) -PIPELINE_CHUNKS = 1 # One 0x71 write in flight at a time; ACK awaited after each chunk -# NOTE: 0x71 data chunks use BLE Write Without Response (no ATT confirmation), but the -# application-layer per-chunk ACK is still awaited, so only one write is ever outstanding. # Upload protocol constants MAX_COMPRESSED_SIZE = 50 * 1024 # Standard firmware buffer (nRF, ~50KB) MAX_START_PAYLOAD = 200 # Maximum bytes in START command (prevents MTU issues) +# Sliding-window PIPE_WRITE (0x0080-0x0082) constants +PIPE_VERSION = 1 # Protocol version carried in the 0x0080 request/response +PIPE_FLAG_COMPRESSED = 0x01 # 0x0080 flags bit0: streamed bytes are zlib-compressed +PIPE_FLAG_PARTIAL = 0x02 # 0x0080 flags bit1: transfer is a partial-region refresh +PIPE_FRAME_OVERHEAD = 3 # Plaintext 0x0081 header: cmd(2) + seq(1) +DEFAULT_MAX_FRAME = 244 # HA native GATT write ceiling (client_max_frame request) +# Seconds to wait for the 0x0080 START response. Pipe attempts are gated on the +# config advertising transmission_modes bit 0x10, so this is a normal command +# timeout, not a fast discovery probe: a gated-in device WILL answer, but on +# ESP32 the ACK is queued and only flushed after panel bring-up returns, which +# can take up to the color-panel busy cap (~30 s worst case). Silence still +# falls back to legacy — it now means a stale config bit (pipe-less firmware), +# paid at most once per connection thanks to the probe cache. +TIMEOUT_PIPE_START = 30.0 +MAX_PTO = 3 # Consecutive silent probe timeouts before aborting a pipe transfer + def build_read_config_command() -> bytes: """Build command to read device TLV configuration. @@ -271,6 +290,157 @@ def build_direct_write_end_with_etag(refresh_mode: int, new_etag: int) -> bytes: return cmd + refresh_mode.to_bytes(1, byteorder="big") + new_etag.to_bytes(4, byteorder="big") +@dataclass(frozen=True) +class PipePartialRequest: + """Partial-region geometry appended to a PIPE_WRITE_START (0x0080) request. + + All fields ride the wire little-endian, matching the rest of the pipe header + (NOTE: the legacy 0x76 partial START packs the same geometry big-endian — that + byte order is deliberately NOT copied here). + + Attributes: + old_etag: uint32 currently displayed etag (nonzero; must equal the device's + displayed_etag or the device NACKs 0x05). + x: Rectangle origin x (uint16, must be a multiple of 8). + y: Rectangle origin y (uint16). + w: Rectangle width (uint16, nonzero, must be a multiple of 8). + h: Rectangle height (uint16, nonzero). + """ + + old_etag: int + x: int + y: int + w: int + h: int + + +def build_pipe_write_start_command( + compressed: bool, + window: int, + ack_every: int, + max_frame: int, + total_size: int, + *, + partial: PipePartialRequest | None = None, +) -> bytes: + """Build a PIPE_WRITE_START (0x0080) start + negotiation command. + + One round trip carries both the transfer parameters and the negotiation + request; the device replies with its own maxima (see parse_pipe_start_response). + + Wire (10-byte header, or 22-byte payload when ``partial`` is set): + [0x00][0x80][ver:1][flags:1][req_window:1][req_ack_every:1] + [client_max_frame:2 LE][total_size:4 LE] + --- appended iff flags bit1 (PIPE_FLAG_PARTIAL) is set --- + [old_etag:4 LE][x:2 LE][y:2 LE][w:2 LE][h:2 LE] + - ver = PIPE_VERSION (1) + - flags bit0 = compressed (zlib, window_bits <= 9) + - flags bit1 = partial-region refresh (0x02); other bits reserved 0 + - req_window = requested "max queue size" W (tokens in flight), 1..32 + - req_ack_every = requested "blocks per ack" N, 1..32 + - client_max_frame = client MTU-derived frame ceiling (<= 244) + - total_size = decompressed panel byte total (partial: plane_size*2) + + The 12-byte partial extension is packed little-endian via ``struct.pack`` to + match the rest of the pipe header; the legacy 0x76 START packs the same fields + big-endian and that byte order is intentionally not reused. ``partial=None`` + yields the exact same bytes as before the extension was added. + + Unlike legacy compressed 0x70 START, this header is fixed-length and carries + NO inline data — all payload flows via 0x0081 DATA frames. + + Args: + compressed: Whether the streamed bytes are zlib-compressed. + window: Requested window / max queue size (tokens in flight). + ack_every: Requested ACK cadence (blocks per ack). + max_frame: Client max frame size in bytes. + total_size: Decompressed panel byte total. + partial: Optional partial-region geometry (keyword-only). When set, flags + bit1 is raised and the 12-byte geometry is appended. + + Returns: + Command bytes for 0x0080. + + Raises: + ValueError: If any field is outside its wire range. + """ + if not 0 <= window <= 0xFF: + raise ValueError(f"window out of uint8 range: {window}") + if not 0 <= ack_every <= 0xFF: + raise ValueError(f"ack_every out of uint8 range: {ack_every}") + if not 0 <= max_frame <= 0xFFFF: + raise ValueError(f"max_frame out of uint16 range: {max_frame}") + if not 0 <= total_size <= 0xFFFFFFFF: + raise ValueError(f"total_size out of uint32 range: {total_size}") + + cmd = CommandCode.PIPE_WRITE_START.to_bytes(2, byteorder="big") + flags = PIPE_FLAG_COMPRESSED if compressed else 0 + if partial is not None: + if not 1 <= partial.old_etag <= 0xFFFFFFFF: + raise ValueError(f"partial old_etag must be a nonzero uint32, got {partial.old_etag}") + for name, value in (("x", partial.x), ("y", partial.y), ("w", partial.w), ("h", partial.h)): + if not 0 <= value <= 0xFFFF: + raise ValueError(f"partial {name} out of uint16 range: {value}") + flags |= PIPE_FLAG_PARTIAL + header = bytes([PIPE_VERSION, flags, window, ack_every]) + packet = cmd + header + struct.pack(" bytes: + """Build a PIPE_WRITE_DATA (0x0081) windowed data frame. + + Wire (plaintext): [0x00][0x81][seq:1][data] + - seq = chunk index mod 256, reset to 0 by each 0x0080 + - data = chunk bytes (<= frame_eff - PIPE_FRAME_OVERHEAD) + + When a session is active the frame is encrypted downstream (device._encrypt_frame): + the standard CCM envelope wraps ``[seq][data]`` as the plaintext payload, so seq + remains the first authenticated plaintext byte. + + Args: + seq: Rolling sequence byte (0..255). + chunk: Chunk payload bytes. + + Returns: + Command bytes for 0x0081. + + Raises: + ValueError: If seq is outside 0..255. + """ + if not 0 <= seq <= 0xFF: + raise ValueError(f"seq out of uint8 range: {seq}") + return CommandCode.PIPE_WRITE_DATA.to_bytes(2, byteorder="big") + bytes([seq]) + chunk + + +def build_pipe_write_end_command(refresh_mode: int, new_etag: int | None = None) -> bytes: + """Build a PIPE_WRITE_END (0x0082) command. + + Wire: [0x00][0x82][refresh:1] (+ [new_etag:4 BE] when provided). + The etag tail mirrors build_direct_write_end_with_etag; presence is by length. + + Args: + refresh_mode: Display refresh mode. Full-frame transfers: 0=full, 1=fast. + Partial-negotiated transfers (PIPE_FLAG_PARTIAL): 0=full, 1=fast, + 2=partial; firmware defaults to partial when the byte is absent. + new_etag: Optional uint32 etag to commit on the device. + + Returns: + Command bytes for 0x0082. + + Raises: + ValueError: If new_etag is outside uint32 range. + """ + cmd = CommandCode.PIPE_WRITE_END.to_bytes(2, byteorder="big") + if new_etag is None: + return cmd + refresh_mode.to_bytes(1, byteorder="big") + if not 0 <= new_etag <= 0xFFFFFFFF: + raise ValueError(f"new_etag out of uint32 range: {new_etag}") + return cmd + refresh_mode.to_bytes(1, byteorder="big") + new_etag.to_bytes(4, byteorder="big") + + def build_led_activate_command( led_instance: int, flash_config: LedFlashConfig, diff --git a/src/opendisplay/protocol/responses.py b/src/opendisplay/protocol/responses.py index 3db5119..927ef42 100644 --- a/src/opendisplay/protocol/responses.py +++ b/src/opendisplay/protocol/responses.py @@ -3,6 +3,7 @@ from __future__ import annotations import struct +from dataclasses import dataclass from ..exceptions import ( AuthenticationFailedError, @@ -268,3 +269,190 @@ def parse_firmware_version(data: bytes) -> FirmwareVersion: "minor": minor, "sha": sha, } + + +# ─── PIPE_WRITE (0x0080-0x0082) sliding-window responses ────────────────────── + +# 0x0080 START NACK error codes (Part 1 §1.1) +PIPE_START_NACK_BAD_PARAMS = 0x01 # bad version / params +PIPE_START_NACK_COMPRESSION = 0x02 # compression unsupported (retry uncompressed) +PIPE_START_NACK_SIZE = 0x03 # total_size mismatch vs panel config +PIPE_START_NACK_BUSY = 0x04 # busy / bad state +PIPE_START_NACK_ETAG_MISMATCH = 0x05 # partial: old_etag == 0 or != device displayed_etag +PIPE_START_NACK_PARTIAL_UNSUPPORTED = 0x06 # partial: bpp != 1 or unsupported driver +PIPE_START_NACK_RECT_INVALID = 0x07 # partial: rect zero-size / OOB / misaligned + +# 0x0081 DATA NACK error codes (all fatal, Part 1 §1.3) +PIPE_DATA_NACK_DECOMPRESS = 0x02 +PIPE_DATA_NACK_WRITE = 0x03 +PIPE_DATA_NACK_STATE = 0x04 + +# Frame classifications returned by classify_pipe_frame(). +PIPE_FRAME_ACK = "PIPE_ACK" +PIPE_FRAME_NACK = "PIPE_NACK" +PIPE_FRAME_END_ACK = "END_ACK" +PIPE_FRAME_END_NACK = "END_NACK" +PIPE_FRAME_OTHER = "OTHER" + + +@dataclass(frozen=True) +class PipeParams: + """Effective sliding-window parameters after the min-rule negotiation. + + All values are the negotiated effective ones (min of client request and + device maximum), computed identically on both sides. + + Attributes: + window: W_eff — tokens in flight (1..32). + ack_every: N_eff — ACK cadence, clamped to <= window. + max_frame: frame_eff — effective frame size in bytes (<= 244). + selective: Response flags bit0 — receiver buffers out-of-order chunks + (selective repeat). When False the sender uses rewind-style recovery. + compressed: Whether this transfer streams zlib-compressed bytes. + partial: Whether this is a partial-region refresh (0x0080 flags bit1 + requested and confirmed by the device's ACK flags bit1). Partial + transfers never auto-complete, so the sender always uses the explicit + 0x0082 END completion contract. + """ + + window: int + ack_every: int + max_frame: int + selective: bool + compressed: bool + partial: bool = False + + +def parse_pipe_start_response(data: bytes) -> tuple[bool, object]: + """Parse a PIPE_WRITE_START (0x0080) response. + + Both forms tolerate trailing bytes (future fields). + + ACK (>= 8 B): [0x00][0x80][ver:1][dev_max_window:1][dev_max_ack_every:1] + [dev_max_frame:2 LE][flags:1] + NACK (4 B): [0xFF][0x80][err:1][0x00] + + Args: + data: Decrypted response bytes (``_read`` yields the ``00 80`` / ``FF 80`` + prefix on both plaintext and encrypted links). + + Returns: + (True, (ver, dev_max_window, dev_max_ack_every, dev_max_frame, flags)) on ACK, + or (False, err_code) on NACK. + + Raises: + InvalidResponseError: On a too-short/garbled ACK or an unexpected echo. + """ + if len(data) < 2: + raise InvalidResponseError(f"PIPE START response too short: {len(data)} bytes") + + if data[0] == 0xFF and data[1] == 0x80: + err = data[2] if len(data) >= 3 else 0 + return False, err + + if data[0] == 0x00 and data[1] == 0x80: + if len(data) < 8: + raise InvalidResponseError(f"PIPE START ACK too short: {len(data)} bytes (need >= 8)") + ver = data[2] + dev_max_window = data[3] + dev_max_ack_every = data[4] + dev_max_frame = int(struct.unpack(" tuple[int, int]: + """Parse a PIPE_WRITE_DATA ACK: [0x00][0x81][highest_seen:1][ack_mask:4 LE]. + + Trailing bytes tolerated. + + Returns: + (highest_seen, ack_mask) + + Raises: + InvalidResponseError: If the frame is not a >= 7-byte 0x0081 ACK. + """ + if len(data) < 7 or data[0] != 0x00 or data[1] != 0x81: + raise InvalidResponseError(f"Not a PIPE DATA ACK: {data[:8].hex()}") + highest_seen = data[2] + ack_mask = int(struct.unpack(" tuple[int, int, int]: + """Parse a PIPE_WRITE_DATA NACK: [0xFF][0x81][err:1][highest_seen:1][ack_mask:4 LE]. + + Trailing bytes tolerated. + + Returns: + (err, highest_seen, ack_mask) + + Raises: + InvalidResponseError: If the frame is not a >= 8-byte 0xFF81 NACK. + """ + if len(data) < 8 or data[0] != 0xFF or data[1] != 0x81: + raise InvalidResponseError(f"Not a PIPE DATA NACK: {data[:8].hex()}") + err = data[2] + highest_seen = data[3] + ack_mask = int(struct.unpack(" set[int]: + """Expand a QUIC-style ACK into absolute acked chunk indexes. + + ``highest_seen`` is a rolling mod-256 seq; it is resolved against the sender's + ``window_base`` (lowest unacked absolute index). Because the in-flight range is + <= 32 « 256, the resolution is unambiguous — including when the ACK is stale + (its highest_seen sits just below window_base after a superseding ACK already + advanced the window). + + ``ack_mask`` bit i (LSB first) marks chunk ``highest_seen - 1 - i`` as received. + + Args: + highest_seen: Highest received seq (mod 256), implicitly acked. + ack_mask: 32-bit selective-ack bitmask. + window_base: Sender's lowest unacked absolute chunk index. + + Returns: + Set of absolute chunk indexes the ACK reports as received. + """ + base_mod = window_base % 256 + delta = (highest_seen - base_mod) % 256 + if delta > 128: # highest_seen is behind window_base (stale / fully contiguous) + delta -= 256 + h_abs = window_base + delta + + acked: set[int] = set() + if h_abs >= 0: + acked.add(h_abs) + for i in range(32): + if ack_mask & (1 << i): + idx = h_abs - 1 - i + if idx >= 0: + acked.add(idx) + return acked + + +def classify_pipe_frame(data: bytes) -> str: + """Classify an inbound frame during a pipe transfer by length + opcode. + + Returns one of PIPE_FRAME_ACK / PIPE_FRAME_NACK / PIPE_FRAME_END_ACK / + PIPE_FRAME_END_NACK / PIPE_FRAME_OTHER. The more-specific (longer) ACK/NACK + forms are checked before the 2-byte END forms. + + Note: pipe reads MUST use this rather than ``check_response_type`` — the + device→host NACK opcode 0xFF81 is not a member of ``CommandCode`` and would + raise there. + """ + if len(data) >= 7 and data[0] == 0x00 and data[1] == 0x81: + return PIPE_FRAME_ACK + if len(data) >= 8 and data[0] == 0xFF and data[1] == 0x81: + return PIPE_FRAME_NACK + if len(data) >= 2 and data[0] == 0x00 and data[1] == 0x82: + return PIPE_FRAME_END_ACK + if len(data) >= 2 and data[0] == 0xFF and data[1] == 0x82: + return PIPE_FRAME_END_NACK + return PIPE_FRAME_OTHER diff --git a/src/opendisplay/transport/connection.py b/src/opendisplay/transport/connection.py index 7f4a3b0..f249d36 100644 --- a/src/opendisplay/transport/connection.py +++ b/src/opendisplay/transport/connection.py @@ -19,6 +19,11 @@ _LOGGER = logging.getLogger(__name__) +# Bounded number of clear-cache-and-rediscover retries on a stale-GATT-cache +# failure: 1 initial attempt + up to MAX_CACHE_RETRIES retries. Kept small so a +# genuinely broken link drops instead of looping. +MAX_CACHE_RETRIES = 2 + class BLEConnection: """Manages BLE connection to OpenDisplay device. @@ -91,33 +96,64 @@ async def connect(self) -> None: if self._client and self._client.is_connected: return # Already connected - try: - await self._attempt_connect(use_services_cache=self.use_services_cache) - except asyncio.TimeoutError as e: - raise BLETimeoutError(f"Connection timeout after {self.timeout}s") from e - except BLEConnectionError as e: - # A stale Bluetooth-proxy GATT cache can make the expected service - # appear missing — e.g. after the device was last seen in the DFU/OTA - # bootloader, the proxy keeps serving the bootloader's GATT for this - # MAC. Clear the proxy cache and retry once with fresh discovery - # before giving up. (Only for the service-missing case; a plain - # "device not found during scan" is re-raised unchanged.) - if SERVICE_UUID.lower() not in str(e).lower(): - raise - _LOGGER.debug( - "Connect to %s failed (%s); clearing GATT cache and retrying once", - self.mac_address, - e, - ) - await self._clear_cache_and_drop() + # A stale Bluetooth-proxy GATT cache can make a connection fail even though + # the device is present: the proxy serves a cached GATT layout whose handles + # no longer match the device (e.g. the expected service appears missing, or a + # CCCD write during notify setup hits an ATT "Invalid handle"). Recover by + # clearing the proxy cache and rediscovering with use_services_cache=False. + # This is bounded (MAX_CACHE_RETRIES) so a genuinely broken link drops the + # connection and raises instead of looping forever. + last_error: Exception | None = None + for attempt in range(MAX_CACHE_RETRIES + 1): + use_cache = self.use_services_cache and attempt == 0 try: - await self._attempt_connect(use_services_cache=False) - except asyncio.TimeoutError as e2: - raise BLETimeoutError(f"Connection timeout after {self.timeout}s") from e2 - except Exception as e2: - raise BLEConnectionError(f"Failed to connect: {e2}") from e2 - except Exception as e: - raise BLEConnectionError(f"Failed to connect: {e}") from e + await self._attempt_connect(use_services_cache=use_cache) + return + except asyncio.TimeoutError as e: + await self._clear_cache_and_drop() + raise BLETimeoutError(f"Connection timeout after {self.timeout}s") from e + except Exception as e: # noqa: BLE001 - classified below + last_error = e + if not self._is_stale_cache_error(e): + # Not a cache problem (e.g. device not found during scan) — drop + # the half-open link and fail fast; retrying won't help. + await self._clear_cache_and_drop() + raise BLEConnectionError(f"Failed to connect: {e}") from e + _LOGGER.debug( + "Connect to %s failed (%s); clearing GATT cache and retrying (attempt %d/%d)", + self.mac_address, + e, + attempt + 1, + MAX_CACHE_RETRIES + 1, + ) + # Clears the proxy cache AND disconnects, so the next iteration + # rediscovers from scratch and we never keep a stale connection. + await self._clear_cache_and_drop() + + # Bounded attempts exhausted: the connection was already dropped by + # _clear_cache_and_drop() above. + raise BLEConnectionError( + f"Failed to connect after {MAX_CACHE_RETRIES + 1} attempts (last error: {last_error})" + ) from last_error + + def _is_stale_cache_error(self, err: Exception) -> bool: + """Return True for GATT failures a proxy cache-clear + rediscovery can fix. + + Covers a missing expected service AND handle-level mismatches (e.g. an + ESPHome proxy serving a cached GATT layout whose CCCD handle no longer + exists on the device -> ATT "Invalid handle" during notify setup). + Deliberately EXCLUDES a plain "not found during scan", which is a device + presence problem, not a cache one, and must not trigger cache-clear retries. + """ + msg = str(err).lower() + if "not found during scan" in msg: + return False + return ( + SERVICE_UUID.lower() in msg + or "invalid handle" in msg + or "invalid attribute" in msg + or "attribute not found" in msg + ) async def _attempt_connect(self, *, use_services_cache: bool) -> None: """Resolve the device, establish a connection, and set up notifications.""" @@ -296,7 +332,7 @@ def drain_notifications(self) -> int: _LOGGER.warning("Discarded %d stale notification(s) before command", dropped) return dropped - async def write_command(self, data: bytes, response: bool = True) -> None: + async def write_command(self, data: bytes, response: bool = True, drain_stale: bool = True) -> None: """Write command to device. Args: @@ -307,6 +343,11 @@ async def write_command(self, data: bytes, response: bool = True) -> None: which are still flow-controlled by the application-layer ACK. Falls back to a Write Request if the characteristic does not advertise write-without-response. + drain_stale: If True (default), discard any queued notifications before + writing so this command's response reads from a clean queue. MUST be + False for every write during a live PIPE_WRITE stream (0x81 data + frames AND the 0x82 END): the sliding window keeps ACKs queued ahead + of the sender, and draining would eat them. Raises: BLEConnectionError: If not connected or write fails @@ -318,8 +359,10 @@ async def write_command(self, data: bytes, response: bool = True) -> None: raise BLEConnectionError("Notifications not set up") # Clear any stale/unsolicited frames so this command's response is read - # from a clean queue (see drain_notifications). - self.drain_notifications() + # from a clean queue (see drain_notifications). Skipped mid-pipe-stream + # where queued ACKs are expected and must be preserved. + if drain_stale: + self.drain_notifications() # Only skip the write confirmation when the caller opts out AND the # characteristic actually supports it; otherwise keep write-with-response. diff --git a/tests/unit/test_connection_clear_cache.py b/tests/unit/test_connection_clear_cache.py index b367a3c..f71a4c9 100644 --- a/tests/unit/test_connection_clear_cache.py +++ b/tests/unit/test_connection_clear_cache.py @@ -10,10 +10,15 @@ from unittest.mock import AsyncMock, MagicMock, patch import pytest +from bleak.exc import BleakError from opendisplay import OpenDisplayDevice from opendisplay.exceptions import BLEConnectionError -from opendisplay.transport.connection import BLEConnection +from opendisplay.transport.connection import MAX_CACHE_RETRIES, BLEConnection + +# A representative ESPHome-proxy stale-cache failure: the cached CCCD handle no +# longer exists on the device, so the notify-enable descriptor write is rejected. +_INVALID_HANDLE = "Bluetooth GATT Error address=CC:2D:73:41:2B:F1 handle=25 error=1 description=Invalid handle" def _connected(client: object) -> BLEConnection: @@ -111,6 +116,72 @@ async def fake_establish(*_a, **_k): await conn.connect() +@pytest.mark.asyncio +async def test_connect_clears_cache_and_retries_on_invalid_handle() -> None: + """An ATT 'Invalid handle' during notify setup → clear cache + rediscover fresh. + + This is the failure that previously bypassed recovery (a raw BleakError, not a + BLEConnectionError, and no SERVICE_UUID in the message) and looped forever. + """ + conn = BLEConnection("AA:BB:CC:DD:EE:FF", ble_device=MagicMock()) + conn._clear_cache_and_drop = AsyncMock() + + used_cache: list[bool] = [] + + async def fake_attempt(*, use_services_cache: bool) -> None: + used_cache.append(use_services_cache) + if len(used_cache) == 1: + raise BleakError(_INVALID_HANDLE) + # second attempt succeeds + + conn._attempt_connect = fake_attempt + await conn.connect() + + assert used_cache == [True, False] # first honors cache, retry rediscovers fresh + conn._clear_cache_and_drop.assert_awaited_once() # poisoned cache cleared before retry + + +@pytest.mark.asyncio +async def test_connect_bounded_then_drops_on_persistent_stale_cache() -> None: + """A stale-cache error that never clears is bounded: fixed attempts, then dropped.""" + conn = BLEConnection("AA:BB:CC:DD:EE:FF", ble_device=MagicMock()) + + attempts = 0 + + async def fake_attempt(*, use_services_cache: bool) -> None: + nonlocal attempts + attempts += 1 + raise BleakError(_INVALID_HANDLE) + + conn._attempt_connect = fake_attempt + with pytest.raises(BLEConnectionError, match="Failed to connect after"): + await conn.connect() + + assert attempts == MAX_CACHE_RETRIES + 1 # bounded, not infinite + assert conn._client is None # connection dropped, not left half-open + + +@pytest.mark.asyncio +async def test_connect_not_found_during_scan_fails_fast() -> None: + """A device-absent error is re-raised immediately: no cache clear, no retries.""" + conn = BLEConnection("AA:BB:CC:DD:EE:FF", ble_device=MagicMock()) + conn._clear_cache_and_drop = AsyncMock() + + attempts = 0 + + async def fake_attempt(*, use_services_cache: bool) -> None: + nonlocal attempts + attempts += 1 + raise BLEConnectionError("Device AA:BB:CC:DD:EE:FF not found during scan") + + conn._attempt_connect = fake_attempt + with pytest.raises(BLEConnectionError, match="Failed to connect"): + await conn.connect() + + assert attempts == 1 # fast-fail, no cache-clear retries + conn._clear_cache_and_drop.assert_awaited_once() # half-open link still dropped + + @pytest.mark.asyncio async def test_device_clear_gatt_cache_delegates_to_connection() -> None: """OpenDisplayDevice.clear_gatt_cache() forwards to the connection and returns it.""" diff --git a/tests/unit/test_device_partial.py b/tests/unit/test_device_partial.py index 8248eac..dca414f 100644 --- a/tests/unit/test_device_partial.py +++ b/tests/unit/test_device_partial.py @@ -3,18 +3,39 @@ from __future__ import annotations import asyncio +import struct +import pytest from epaper_dithering import ColorScheme from PIL import Image +from test_pipe_write import ( + END_ACK, + REFRESH_COMPLETE, + REFRESH_TIMEOUT, + ScriptedConn, + data_ack, + data_nack, + start_ack, + start_nack, +) +from test_pipe_write_sender import _make_session from opendisplay import OpenDisplayDevice +from opendisplay.crypto import decrypt_response +from opendisplay.encoding.compression import zlib_window_bits +from opendisplay.exceptions import BLETimeoutError, RefreshTimeoutError from opendisplay.models.capabilities import DeviceCapabilities from opendisplay.models.config import DisplayConfig, GlobalConfig, ManufacturerData, PowerOption, SystemConfig -from opendisplay.models.enums import RefreshMode +from opendisplay.models.enums import PartialUpdateSupport, RefreshMode from opendisplay.partial import ERR_ETAG_MISMATCH, PARTIAL_FLAG_COMPRESSED, PartialState -def _config(partial_update_support: int = 1, transmission_modes: int = 0x00) -> GlobalConfig: +def _config( + partial_update_support: int = 1, + transmission_modes: int = 0x00, + pixel_width: int = 16, + pixel_height: int = 8, +) -> GlobalConfig: return GlobalConfig( system=SystemConfig( ic_type=0, @@ -49,8 +70,8 @@ def _config(partial_update_support: int = 1, transmission_modes: int = 0x00) -> instance_number=0, display_technology=0, panel_ic_type=0, - pixel_width=16, - pixel_height=8, + pixel_width=pixel_width, + pixel_height=pixel_height, active_width_mm=10, active_height_mm=10, tag_type=0, @@ -325,3 +346,343 @@ def test_no_change_still_skips_transfer_on_full_frame_panels(): _image(), state, _config(partial_update_support=PartialUpdateSupport.FULL_FRAME), ColorScheme.MONO ) assert result == "no_change" + + +# ─── Pipe-partial (0x0080 flags bit1) end-to-end flow ──────────────────────── + + +def _pipe_device( + config: GlobalConfig, + *, + width: int, + height: int, + max_queue_size: int = 16, + blocks_per_ack: int = 8, +) -> OpenDisplayDevice: + return OpenDisplayDevice( + mac_address="AA:BB:CC:DD:EE:FF", + config=config, + capabilities=DeviceCapabilities(width=width, height=height, color_scheme=ColorScheme.MONO), + max_queue_size=max_queue_size, + blocks_per_ack=blocks_per_ack, + ) + + +def _mono(width: int, height: int, fill: int = 0) -> Image.Image: + return Image.new("P", (width, height), fill) + + +def test_pipe_partial_happy_path_uncompressed(): + device = _pipe_device(_config(transmission_modes=0x10), width=16, height=8) + new = _image(changed=True) + state = PartialState(etag=0x01020304, last_image=_image().tobytes(), width=16, height=8, bytes_per_pixel=1) + conn = ScriptedConn([start_ack(flags=0x03), data_ack({0}), END_ACK, REFRESH_COMPLETE]) + device._connection = conn + + asyncio.run(device.upload_prepared_image((b"\x00" * 16, None, new), refresh_mode=RefreshMode.PARTIAL, state=state)) + + starts = [w for w in conn.written if w[:2] == b"\x00\x80"] + assert len(starts) == 1 + assert starts[0][3] & 0x02 # partial flag + assert len(starts[0]) == 24 # extended packet + assert any(w[:2] == b"\x00\x81" for w in conn.written) # data rode the pipe + assert all(w[:2] != b"\x00\x76" for w in conn.written) # no legacy fallback + ends = [w for w in conn.written if w[:2] == b"\x00\x82"] + assert len(ends) == 1 + assert ends[0][2] == 2 # REFRESH_PARTIAL selector + committed_etag = int.from_bytes(ends[0][3:7], "big") + assert state.etag == committed_etag + assert state.etag != 0x01020304 # a fresh etag was committed + assert state.last_image == new.tobytes() + + +def test_pipe_partial_happy_path_compressed_zlib_window9(): + # 64x64 FULL_FRAME + streaming decompression (0x01) + pipe (0x10) → compression engages. + cfg = _config( + partial_update_support=PartialUpdateSupport.FULL_FRAME, + transmission_modes=0x11, + pixel_width=64, + pixel_height=64, + ) + device = _pipe_device(cfg, width=64, height=64) + new = _mono(64, 64) + new.putpixel((10, 10), 1) + state = PartialState(etag=0x0A0B0C0D, last_image=_mono(64, 64).tobytes(), width=64, height=64, bytes_per_pixel=1) + conn = ScriptedConn([start_ack(flags=0x03), data_ack({0}), END_ACK, REFRESH_COMPLETE]) + device._connection = conn + + asyncio.run( + device.upload_prepared_image((b"\x00" * (64 * 64), None, new), refresh_mode=RefreshMode.PARTIAL, state=state) + ) + + start = next(w for w in conn.written if w[:2] == b"\x00\x80") + assert start[3] == 0x03 # compressed + partial + # Reassemble the streamed bytes from the 0x0081 frames → confirm zlib window 9. + data_frames = sorted((w for w in conn.written if w[:2] == b"\x00\x81"), key=lambda w: w[2]) + payload = b"".join(w[3:] for w in data_frames) + assert zlib_window_bits(payload) == 9 + + +def test_pipe_partial_mid_stream_nack_falls_back_to_full(monkeypatch): + device = _pipe_device(_config(transmission_modes=0x10), width=16, height=8) + new = _image(changed=True) + state = PartialState(etag=0x01020304, last_image=_image().tobytes(), width=16, height=8, bytes_per_pixel=1) + conn = ScriptedConn([start_ack(flags=0x03), data_nack(0x03, 0, 0x0)]) + device._connection = conn + full_uploads = 0 + + async def execute_upload(image_data, refresh_mode, **kwargs) -> bool: + nonlocal full_uploads + full_uploads += 1 + return True + + monkeypatch.setattr(device, "_execute_upload", execute_upload) + + asyncio.run(device.upload_prepared_image((b"\x00" * 16, None, new), state=state)) + + assert any(w[:2] == b"\x00\x80" for w in conn.written) # pipe-partial was attempted + assert all(w[:2] != b"\x00\x82" for w in conn.written) # aborted before END + assert full_uploads == 1 # clean fallback to full + # The aborted transfer clears PartialState (plan 1.3/6) before the fallback; + # the successful full upload then re-baselines it with a FRESH etag. Either + # way the stale pre-NACK etag (now cleared on the device too) must be gone. + assert state.etag != 0x01020304 + + +def test_pipe_partial_refresh_timeout_reraises(): + device = _pipe_device(_config(transmission_modes=0x10), width=16, height=8) + new = _image(changed=True) + state = PartialState(etag=0x01020304, last_image=_image().tobytes(), width=16, height=8, bytes_per_pixel=1) + conn = ScriptedConn([start_ack(flags=0x03), data_ack({0}), END_ACK, REFRESH_TIMEOUT]) + device._connection = conn + + with pytest.raises(RefreshTimeoutError, match="refresh timed out"): + asyncio.run(device.upload_prepared_image((b"\x00" * 16, None, new), state=state)) + + +def test_pipe_partial_ack_no_bit1_falls_back_to_0x76(): + device = _pipe_device(_config(transmission_modes=0x10), width=16, height=8) + new = _image(changed=True) + state = PartialState(etag=0x01020304, last_image=_image().tobytes(), width=16, height=8, bytes_per_pixel=1) + # ACK without the partial bit → None → legacy 0x76 flow. + conn = ScriptedConn([start_ack(flags=0x01), b"\x00\x76", b"\x00\x72", b"\x00\x73"]) + device._connection = conn + + asyncio.run(device.upload_prepared_image((b"\x00" * 16, None, new), state=state)) + + assert any(w[:2] == b"\x00\x80" for w in conn.written) + assert any(w[:2] == b"\x00\x76" for w in conn.written) # fell back to legacy partial + assert device._pipe_partial_supported is False # cached negative + + +def test_pipe_partial_nack_05_skips_0x76_goes_full(monkeypatch): + device = _pipe_device(_config(transmission_modes=0x10), width=16, height=8) + new = _image(changed=True) + state = PartialState(etag=0x01020304, last_image=_image().tobytes(), width=16, height=8, bytes_per_pixel=1) + conn = ScriptedConn([start_nack(0x05)]) + device._connection = conn + full_uploads = 0 + + async def execute_upload(image_data, refresh_mode, **kwargs) -> bool: + nonlocal full_uploads + full_uploads += 1 + return True + + monkeypatch.setattr(device, "_execute_upload", execute_upload) + + asyncio.run(device.upload_prepared_image((b"\x00" * 16, None, new), state=state)) + + assert full_uploads == 1 + assert all(w[:2] != b"\x00\x76" for w in conn.written) # etag mismatch skips legacy partial + + +def test_pipe_partial_nack_06_negative_cache_skips_0x76(monkeypatch): + device = _pipe_device(_config(transmission_modes=0x10), width=16, height=8) + new = _image(changed=True) + state = PartialState(etag=0x01020304, last_image=_image().tobytes(), width=16, height=8, bytes_per_pixel=1) + conn = ScriptedConn([start_nack(0x06)]) + device._connection = conn + full_uploads = 0 + + async def execute_upload(image_data, refresh_mode, **kwargs) -> bool: + nonlocal full_uploads + full_uploads += 1 + return True + + monkeypatch.setattr(device, "_execute_upload", execute_upload) + + asyncio.run(device.upload_prepared_image((b"\x00" * 16, None, new), state=state)) + + assert full_uploads == 1 + assert all(w[:2] != b"\x00\x76" for w in conn.written) + assert device._pipe_partial_supported is False # 0x06 caches negative + + +def test_pipe_partial_compressed_02_retry_then_0x76(): + cfg = _config( + partial_update_support=PartialUpdateSupport.FULL_FRAME, + transmission_modes=0x11, + pixel_width=64, + pixel_height=64, + ) + device = _pipe_device(cfg, width=64, height=64) + new = _mono(64, 64) + new.putpixel((10, 10), 1) + state = PartialState(etag=0x0A0B0C0D, last_image=_mono(64, 64).tobytes(), width=64, height=64, bytes_per_pixel=1) + # compressed 0x02 → uncompressed-still-partial 0x02 → give up → legacy 0x76. + conn = ScriptedConn([start_nack(0x02), start_nack(0x02), b"\x00\x76", b"\x00\x72", b"\x00\x73"]) + device._connection = conn + + asyncio.run(device.upload_prepared_image((b"\x00" * (64 * 64), None, new), state=state)) + + starts = [w for w in conn.written if w[:2] == b"\x00\x80"] + assert len(starts) == 2 + assert starts[0][3] == 0x03 # compressed + partial + assert starts[1][3] == 0x02 # uncompressed, still partial + assert device._pipe_partial_supported is False + assert any(w[:2] == b"\x00\x76" for w in conn.written) + + +def test_pipe_partial_silence_falls_back_to_0x76(): + device = _pipe_device(_config(transmission_modes=0x10), width=16, height=8) + new = _image(changed=True) + state = PartialState(etag=0x01020304, last_image=_image().tobytes(), width=16, height=8, bytes_per_pixel=1) + conn = ScriptedConn([BLETimeoutError, b"\x00\x76", b"\x00\x72", b"\x00\x73"]) + device._connection = conn + + asyncio.run(device.upload_prepared_image((b"\x00" * 16, None, new), state=state)) + + assert any(w[:2] == b"\x00\x76" for w in conn.written) + assert device._pipe_supported is False + + +def test_pipe_partial_negative_cache_skips_second_probe(): + device = _pipe_device(_config(transmission_modes=0x10), width=16, height=8) + device._pipe_partial_supported = False # negative cached this connection + new = _image(changed=True) + state = PartialState(etag=0x01020304, last_image=_image().tobytes(), width=16, height=8, bytes_per_pixel=1) + conn = ScriptedConn([b"\x00\x76", b"\x00\x72", b"\x00\x73"]) + device._connection = conn + + asyncio.run(device.upload_prepared_image((b"\x00" * 16, None, new), state=state)) + + # No partial-flagged 0x0080 is ever written after the negative cache. + assert all(not (w[:2] == b"\x00\x80" and w[3] & 0x02) for w in conn.written) + assert conn.written[0][:2] == b"\x00\x76" + + +# ─── Pipe-partial gates ────────────────────────────────────────────────────── + + +def test_gate_no_pipe_write_bit_skips_0x0080(): + device = _pipe_device(_config(transmission_modes=0x00), width=16, height=8) + new = _image(changed=True) + state = PartialState(etag=0x01020304, last_image=_image().tobytes(), width=16, height=8, bytes_per_pixel=1) + conn = ScriptedConn([b"\x00\x76", b"\x00\x72", b"\x00\x73"]) + device._connection = conn + + asyncio.run(device.upload_prepared_image((b"\x00" * 16, None, new), state=state)) + + assert all(w[:2] != b"\x00\x80" for w in conn.written) # supports_pipe_write clear → no probe + assert conn.written[0][:2] == b"\x00\x76" + + +def test_gate_max_queue_size_one_skips_0x0080(): + device = _pipe_device(_config(transmission_modes=0x10), width=16, height=8, max_queue_size=1) + new = _image(changed=True) + state = PartialState(etag=0x01020304, last_image=_image().tobytes(), width=16, height=8, bytes_per_pixel=1) + conn = ScriptedConn([b"\x00\x76", b"\x00\x72", b"\x00\x73"]) + device._connection = conn + + asyncio.run(device.upload_prepared_image((b"\x00" * 16, None, new), state=state)) + + assert all(w[:2] != b"\x00\x80" for w in conn.written) # pipe disabled → no probe + assert conn.written[0][:2] == b"\x00\x76" + + +def test_gate_full_frame_whole_panel_rect_and_total_size(): + cfg = _config( + partial_update_support=PartialUpdateSupport.FULL_FRAME, + transmission_modes=0x10, + pixel_width=16, + pixel_height=8, + ) + device = _pipe_device(cfg, width=16, height=8) + new = _mono(16, 8) + new.putpixel((3, 3), 1) + state = PartialState(etag=0x01020304, last_image=_mono(16, 8).tobytes(), width=16, height=8, bytes_per_pixel=1) + conn = ScriptedConn([start_ack(flags=0x03), data_ack({0}), END_ACK, REFRESH_COMPLETE]) + device._connection = conn + + asyncio.run(device.upload_prepared_image((b"\x00" * 16, None, new), refresh_mode=RefreshMode.PARTIAL, state=state)) + + start = next(w for w in conn.written if w[:2] == b"\x00\x80") + total_size = struct.unpack(" OpenDisplayDevice: config = _make_config(transmission_modes=transmission_modes, width=width, height=height) caps = DeviceCapabilities(width=width, height=height, color_scheme=ColorScheme.MONO) - return OpenDisplayDevice(mac_address="AA:BB:CC:DD:EE:FF", config=config, capabilities=caps) + # These tests exercise the LEGACY 0x70/0x71/0x72 wire protocol; max_queue_size=1 + # disables the PIPE_WRITE probe so the scripted legacy responses line up. + return OpenDisplayDevice(mac_address="AA:BB:CC:DD:EE:FF", config=config, capabilities=caps, max_queue_size=1) # ─── _read(): encrypted session behaviour ──────────────────────────────────── @@ -558,7 +560,8 @@ def _make_gray4_device(transmission_modes: int = 0x02) -> OpenDisplayDevice: """Device reporting a 4-gray panel (uploads ship two split planes, plane0 ++ plane1).""" config = _make_config(transmission_modes=transmission_modes, width=8, height=8) caps = DeviceCapabilities(width=8, height=8, color_scheme=ColorScheme.GRAYSCALE_4) - return OpenDisplayDevice(mac_address="AA:BB:CC:DD:EE:FF", config=config, capabilities=caps) + # Legacy-protocol tests: disable the PIPE_WRITE probe (see _make_device). + return OpenDisplayDevice(mac_address="AA:BB:CC:DD:EE:FF", config=config, capabilities=caps, max_queue_size=1) @pytest.mark.asyncio diff --git a/tests/unit/test_pipe_write.py b/tests/unit/test_pipe_write.py new file mode 100644 index 0000000..411c24e --- /dev/null +++ b/tests/unit/test_pipe_write.py @@ -0,0 +1,763 @@ +"""Unit tests for the PIPE_WRITE (0x0080-0x0082) sliding-window protocol. + +Covers builders, response parsers, unpack_ack_ranges, classify_pipe_frame, and +the negotiation / probe-fallback layer. The sender loop (selective repeat, loss +recovery, PTO/RETX aborts, encryption, drain_stale) lives in +test_pipe_write_sender.py. +""" + +from __future__ import annotations + +import struct + +import pytest +from epaper_dithering import ColorScheme +from PIL import Image + +from opendisplay import OpenDisplayDevice +from opendisplay.device import _PipePartialEtagMismatch, _PipePartialRejected +from opendisplay.exceptions import BLETimeoutError, InvalidResponseError +from opendisplay.models.capabilities import DeviceCapabilities +from opendisplay.models.config import ( + DisplayConfig, + GlobalConfig, + ManufacturerData, + PowerOption, + SystemConfig, +) +from opendisplay.models.enums import RefreshMode +from opendisplay.partial import PartialRegion +from opendisplay.protocol import ( + DEFAULT_MAX_FRAME, + PIPE_FLAG_COMPRESSED, + PIPE_FLAG_PARTIAL, + PIPE_FRAME_OVERHEAD, + PIPE_VERSION, + CommandCode, + PipeParams, + PipePartialRequest, + build_pipe_write_data_command, + build_pipe_write_end_command, + build_pipe_write_start_command, + classify_pipe_frame, + parse_pipe_data_ack, + parse_pipe_data_nack, + parse_pipe_start_response, + unpack_ack_ranges, +) +from opendisplay.protocol.responses import ( + PIPE_FRAME_ACK, + PIPE_FRAME_END_ACK, + PIPE_FRAME_END_NACK, + PIPE_FRAME_NACK, + PIPE_FRAME_OTHER, + PIPE_START_NACK_ETAG_MISMATCH, + PIPE_START_NACK_PARTIAL_UNSUPPORTED, + PIPE_START_NACK_RECT_INVALID, +) + +# ─── Wire-frame helpers (shared with the sender tests via import) ───────────── + + +def start_ack(ver: int = 1, dw: int = 32, da: int = 32, df: int = 244, flags: int = 0x01) -> bytes: + """Build a 0x0080 START ACK frame.""" + return b"\x00\x80" + bytes([ver, dw, da]) + struct.pack(" bytes: + """Build a 0x0080 START NACK frame.""" + return b"\xff\x80" + bytes([err, 0x00]) + + +def data_ack(received: set[int]) -> bytes: + """Build a 0x0081 DATA ACK reflecting a set of received chunk indexes. + + Uses cumulative highest_seen = max(received) plus a selective 32-bit mask. + Only valid for indexes < 256 (unit tests never exceed that within one window). + """ + hs = max(received) + mask = 0 + for i in range(32): + idx = hs - 1 - i + if idx in received: + mask |= 1 << i + return b"\x00\x81" + bytes([hs % 256]) + struct.pack(" bytes: + return b"\x00\x81" + bytes([highest_seen % 256]) + struct.pack(" bytes: + return b"\xff\x81" + bytes([err, highest_seen % 256]) + struct.pack(" None: + self.written: list[bytes] = [] + self.write_responses: list[bool] = [] + self.drain_flags: list[bool] = [] + self._responses = list(responses) + self.timeouts: list[float] = [] + self.writes_at_read: list[int] = [] + + async def write_command(self, data: bytes, response: bool = True, drain_stale: bool = True) -> None: + self.written.append(data) + self.write_responses.append(response) + self.drain_flags.append(drain_stale) + + async def read_response(self, timeout: float) -> bytes: + self.timeouts.append(timeout) + self.writes_at_read.append(len(self.written)) + if not self._responses: + raise RuntimeError("ScriptedConn: no responses left") + item = self._responses.pop(0) + if isinstance(item, type) and issubclass(item, BaseException): + raise item("scripted") + if isinstance(item, BaseException): + raise item + return item + + +def make_config(transmission_modes: int = 0x12) -> GlobalConfig: + # Default 0x12 = zip (0x02) + pipe write (0x10): _execute_upload only probes + # 0x0080 when the config advertises the pipe bit, so upload-driving pipe + # tests need it. Pass an explicit value to test the gate itself. + return GlobalConfig( + system=SystemConfig(ic_type=0, communication_modes=0, device_flags=0, pwr_pin=0xFF, reserved=b"\x00" * 17), + manufacturer=ManufacturerData(manufacturer_id=0, board_type=0, board_revision=0, reserved=b"\x00" * 18), + power=PowerOption( + power_mode=0, + battery_capacity_mah=b"\x00\x00\x00", + sleep_timeout_ms=0, + tx_power=0, + sleep_flags=0, + battery_sense_pin=0xFF, + battery_sense_enable_pin=0xFF, + battery_sense_flags=0, + capacity_estimator=0, + voltage_scaling_factor=0, + deep_sleep_current_ua=0, + deep_sleep_time_seconds=0, + reserved=b"\x00" * 12, + ), + displays=[ + DisplayConfig( + instance_number=0, + display_technology=0, + panel_ic_type=0, + pixel_width=8, + pixel_height=8, + active_width_mm=10, + active_height_mm=10, + tag_type=0, + rotation=0, + reset_pin=0xFF, + busy_pin=0xFF, + dc_pin=0xFF, + cs_pin=0xFF, + data_pin=0, + partial_update_support=0, + color_scheme=ColorScheme.MONO.value, + transmission_modes=transmission_modes, + clk_pin=0, + reserved_pins=b"\x00" * 7, + full_update_mC=0, + reserved=b"\x00" * 13, + ) + ], + ) + + +def make_device( + responses: list, + *, + blocks_per_ack: int = 8, + max_queue_size: int = 16, + transmission_modes: int = 0x12, +) -> tuple[OpenDisplayDevice, ScriptedConn]: + dev = OpenDisplayDevice( + mac_address="AA:BB:CC:DD:EE:FF", + config=make_config(transmission_modes), + capabilities=DeviceCapabilities(width=8, height=8, color_scheme=ColorScheme.MONO), + blocks_per_ack=blocks_per_ack, + max_queue_size=max_queue_size, + ) + conn = ScriptedConn(responses) + dev._connection = conn + return dev, conn + + +def make_region(rx: int = 0, ry: int = 0, rw: int = 8, rh: int = 8) -> PartialRegion: + """Build a minimal PartialRegion for pipe-partial negotiation tests.""" + cfg = make_config(transmission_modes=0x10) + img = Image.new("P", (8, 8), 0) + return PartialRegion( + display=cfg.displays[0], + color_scheme=ColorScheme.MONO, + width=8, + height=8, + palette_image=img, + new_palette=img.tobytes(), + old_palette=img.tobytes(), + rx=rx, + ry=ry, + rw=rw, + rh=rh, + ) + + +# ─── Builders ──────────────────────────────────────────────────────────────── + + +def test_build_pipe_write_start_compressed() -> None: + cmd = build_pipe_write_start_command(True, 16, 8, 244, 0x11223344) + assert cmd[:2] == b"\x00\x80" + assert cmd[2] == PIPE_VERSION + assert cmd[3] == PIPE_FLAG_COMPRESSED + assert cmd[4] == 16 # req_window + assert cmd[5] == 8 # req_ack_every + assert cmd[6:8] == struct.pack(" None: + cmd = build_pipe_write_start_command(False, 1, 1, 244, 0) + assert cmd[3] == 0 + + +@pytest.mark.parametrize( + "kwargs", + [ + {"window": 256}, + {"ack_every": -1}, + {"max_frame": 0x10000}, + {"total_size": 0x1_0000_0000}, + ], +) +def test_build_pipe_write_start_range_validation(kwargs: dict) -> None: + base = {"compressed": False, "window": 8, "ack_every": 4, "max_frame": 244, "total_size": 0} + base.update(kwargs) + with pytest.raises(ValueError): + build_pipe_write_start_command(**base) # type: ignore[arg-type] + + +# ─── Partial START extension (0x0080 flags bit1) ───────────────────────────── + + +def test_build_pipe_write_start_partial_byte_layout() -> None: + partial = PipePartialRequest(old_etag=0x11223344, x=8, y=16, w=24, h=32) + cmd = build_pipe_write_start_command(False, 16, 8, 244, 0x30, partial=partial) + # 2 (opcode) + 10 (header) + 12 (extension) = 24 bytes. + assert len(cmd) == 24 + assert cmd[:2] == b"\x00\x80" + assert cmd[2] == PIPE_VERSION + assert cmd[3] == PIPE_FLAG_PARTIAL # flags bit1 only (uncompressed) + assert cmd[4] == 16 # req_window + assert cmd[5] == 8 # req_ack_every + assert cmd[6:8] == struct.pack(" None: + partial = PipePartialRequest(old_etag=1, x=0, y=0, w=8, h=8) + cmd = build_pipe_write_start_command(True, 4, 2, 244, 128, partial=partial) + assert cmd[3] == PIPE_FLAG_COMPRESSED | PIPE_FLAG_PARTIAL # 0x03 + + +def test_build_pipe_write_start_partial_none_is_byte_identical() -> None: + plain = build_pipe_write_start_command(True, 16, 8, 244, 0x11223344) + explicit_none = build_pipe_write_start_command(True, 16, 8, 244, 0x11223344, partial=None) + assert plain == explicit_none + assert len(plain) == 12 # unchanged 12-byte packet + + +@pytest.mark.parametrize( + "kwargs", + [ + {"old_etag": 0}, # zero etag rejected + {"old_etag": 0x1_0000_0000}, # > uint32 + {"x": 0x10000}, + {"y": -1}, + {"w": 0x10000}, + {"h": 0x10000}, + ], +) +def test_build_pipe_write_start_partial_range_validation(kwargs: dict) -> None: + fields = {"old_etag": 1, "x": 0, "y": 0, "w": 8, "h": 8} + fields.update(kwargs) + partial = PipePartialRequest(**fields) # type: ignore[arg-type] + with pytest.raises(ValueError): + build_pipe_write_start_command(False, 8, 4, 244, 128, partial=partial) + + +def test_build_pipe_write_data() -> None: + cmd = build_pipe_write_data_command(200, b"payload") + assert cmd == b"\x00\x81" + bytes([200]) + b"payload" + + +def test_build_pipe_write_data_seq_range() -> None: + with pytest.raises(ValueError): + build_pipe_write_data_command(256, b"x") + + +def test_build_pipe_write_end_no_etag() -> None: + assert build_pipe_write_end_command(0) == b"\x00\x82\x00" + + +def test_build_pipe_write_end_with_etag_mirrors_direct_write() -> None: + cmd = build_pipe_write_end_command(1, 0xDEADBEEF) + assert cmd == b"\x00\x82" + bytes([1]) + (0xDEADBEEF).to_bytes(4, "big") + + +def test_build_pipe_write_end_etag_range() -> None: + with pytest.raises(ValueError): + build_pipe_write_end_command(0, 0x1_0000_0000) + + +# ─── parse_pipe_start_response ─────────────────────────────────────────────── + + +def test_parse_start_ack() -> None: + ok, payload = parse_pipe_start_response(start_ack(1, 32, 16, 244, 0x01)) + assert ok is True + assert payload == (1, 32, 16, 244, 0x01) + + +def test_parse_start_ack_tolerates_trailing_bytes() -> None: + ok, payload = parse_pipe_start_response(start_ack() + b"\xaa\xbb") + assert ok is True + assert payload[3] == 244 # type: ignore[index] + + +def test_parse_start_ack_too_short_raises() -> None: + with pytest.raises(InvalidResponseError): + parse_pipe_start_response(b"\x00\x80\x01\x20") # only 4 bytes + + +@pytest.mark.parametrize("err", [0x01, 0x02, 0x03, 0x04]) +def test_parse_start_nack(err: int) -> None: + ok, payload = parse_pipe_start_response(start_nack(err)) + assert ok is False + assert payload == err + + +def test_parse_start_bad_echo_raises() -> None: + with pytest.raises(InvalidResponseError): + parse_pipe_start_response(b"\x00\x70\x00\x00\x00\x00\x00\x00") + + +@pytest.mark.parametrize("err", [0x05, 0x06, 0x07]) +def test_parse_start_nack_partial_codes(err: int) -> None: + ok, payload = parse_pipe_start_response(start_nack(err)) + assert ok is False + assert payload == err + + +def test_partial_nack_constants() -> None: + assert PIPE_START_NACK_ETAG_MISMATCH == 0x05 + assert PIPE_START_NACK_PARTIAL_UNSUPPORTED == 0x06 + assert PIPE_START_NACK_RECT_INVALID == 0x07 + + +def test_pipe_params_partial_defaults_false() -> None: + params = PipeParams(window=8, ack_every=4, max_frame=244, selective=True, compressed=False) + assert params.partial is False + # Explicit partial=True is accepted (frozen dataclass stays valid). + assert PipeParams(8, 4, 244, True, False, partial=True).partial is True + + +# ─── parse_pipe_data_ack / nack ────────────────────────────────────────────── + + +def test_parse_data_ack() -> None: + hs, mask = parse_pipe_data_ack(data_ack_raw(9, 0xDEADBEEF)) + assert hs == 9 + assert mask == 0xDEADBEEF + + +def test_parse_data_ack_trailing_bytes() -> None: + hs, mask = parse_pipe_data_ack(data_ack_raw(3, 0x7) + b"\xff") + assert (hs, mask) == (3, 0x7) + + +def test_parse_data_ack_wrong_shape_raises() -> None: + with pytest.raises(InvalidResponseError): + parse_pipe_data_ack(b"\x00\x81\x03") # too short + + +def test_parse_data_nack() -> None: + err, hs, mask = parse_pipe_data_nack(data_nack(0x03, 5, 0x1F)) + assert (err, hs, mask) == (0x03, 5, 0x1F) + + +# ─── unpack_ack_ranges ─────────────────────────────────────────────────────── + + +def test_unpack_contiguous() -> None: + # received {0,1,2}, window_base 0 + assert unpack_ack_ranges(2, 0b11, 0) == {0, 1, 2} + + +def test_unpack_with_hole() -> None: + # received {0,1,3,4}: hs=4 mask bits for 3,1,0 + mask = (1 << 0) | (1 << 2) | (1 << 3) # 3, 1, 0 + got = unpack_ack_ranges(4, mask, 0) + assert got == {0, 1, 3, 4} + assert 2 not in got + + +def test_unpack_mod256_rollover() -> None: + # window_base 250; receiver highest_seen wrapped to 4 (absolute 260) + # received absolute 250..260 contiguous → mask all 1s for 10 below hs + mask = (1 << 10) - 1 + got = unpack_ack_ranges(4, mask, 250) + assert got == set(range(250, 261)) + + +def test_unpack_stale_ack_below_window_base() -> None: + # window_base already advanced to 10; a stale ACK reports highest_seen 9. + got = unpack_ack_ranges(9, 0, 10) + assert got == {9} # resolves behind window_base, no spurious future index + + +# ─── classify_pipe_frame ───────────────────────────────────────────────────── + + +@pytest.mark.parametrize( + "frame,expected", + [ + (data_ack_raw(3, 0), PIPE_FRAME_ACK), + (data_nack(2, 3, 0), PIPE_FRAME_NACK), + (END_ACK, PIPE_FRAME_END_ACK), + (END_NACK, PIPE_FRAME_END_NACK), + (b"\x00\x81\x03", PIPE_FRAME_OTHER), # 3 bytes < 7 → not an ACK + (b"\xff\x81\x02\x03\x00\x00\x00", PIPE_FRAME_OTHER), # 7 bytes < 8 → not a NACK + (b"\x00\x73", PIPE_FRAME_OTHER), + ], +) +def test_classify_pipe_frame(frame: bytes, expected: str) -> None: + assert classify_pipe_frame(frame) == expected + + +# ─── Negotiation / probe ───────────────────────────────────────────────────── + + +@pytest.mark.asyncio +async def test_negotiate_min_rule() -> None: + dev, conn = make_device([start_ack(dw=32, da=32, df=244, flags=0x01)], blocks_per_ack=8, max_queue_size=16) + params = await dev._negotiate_pipe(compressed=True, total_size=100) + assert params == PipeParams(window=16, ack_every=8, max_frame=244, selective=True, compressed=True) + # 0x0080 sent with the requested params. The START wait is a normal command + # timeout (attempts are config-gated), sized for ESP32's ACK-after-bring-up + # flush on slow color panels — not the old 2 s discovery probe. + assert conn.written[0][:2] == b"\x00\x80" + assert conn.timeouts[0] == pytest.approx(30.0) + + +@pytest.mark.asyncio +async def test_negotiate_window_clamped_to_32() -> None: + dev, _ = make_device([start_ack(dw=64, da=64, df=244)], blocks_per_ack=64, max_queue_size=64) + params = await dev._negotiate_pipe(compressed=False, total_size=10) + assert params is not None + assert params.window == 32 # hard cap + assert params.ack_every == 32 # min(req, dev, W) + + +@pytest.mark.asyncio +async def test_negotiate_ack_every_clamped_to_window() -> None: + dev, _ = make_device([start_ack(dw=4, da=32, df=244)], blocks_per_ack=32, max_queue_size=16) + params = await dev._negotiate_pipe(compressed=False, total_size=10) + assert params is not None + assert params.window == 4 + assert params.ack_every == 4 # N clamped to W + + +@pytest.mark.asyncio +async def test_negotiate_floor_one() -> None: + dev, _ = make_device([start_ack(dw=0, da=0, df=244)], blocks_per_ack=1, max_queue_size=1) + # max_queue_size=1 disables pipe entirely, so call with a device that allows it: + dev2, _ = make_device([start_ack(dw=0, da=0, df=244)], blocks_per_ack=1, max_queue_size=2) + params = await dev2._negotiate_pipe(compressed=False, total_size=10) + assert params is not None + assert params.window == 1 + assert params.ack_every == 1 + + +@pytest.mark.asyncio +async def test_negotiate_frame_min() -> None: + dev, _ = make_device([start_ack(dw=32, da=32, df=180)], max_queue_size=16) + params = await dev._negotiate_pipe(compressed=False, total_size=10) + assert params is not None + assert params.max_frame == min(DEFAULT_MAX_FRAME, 180) + + +@pytest.mark.asyncio +async def test_negotiate_silence_returns_none() -> None: + dev, conn = make_device([BLETimeoutError], max_queue_size=16) + params = await dev._negotiate_pipe(compressed=True, total_size=10) + assert params is None + + +@pytest.mark.asyncio +async def test_negotiate_nack_bad_params_returns_none() -> None: + dev, _ = make_device([start_nack(0x01)], max_queue_size=16) + assert await dev._negotiate_pipe(compressed=True, total_size=10) is None + + +@pytest.mark.asyncio +async def test_negotiate_nack_compression_retries_uncompressed() -> None: + dev, conn = make_device([start_nack(0x02), start_ack(flags=0x01)], max_queue_size=16) + params = await dev._negotiate_pipe(compressed=True, total_size=10) + assert params is not None + assert params.compressed is False + # Two 0x0080 writes: first compressed, second uncompressed (flags=0). + starts = [w for w in conn.written if w[:2] == b"\x00\x80"] + assert len(starts) == 2 + assert starts[0][3] == PIPE_FLAG_COMPRESSED + assert starts[1][3] == 0 + + +@pytest.mark.asyncio +async def test_negotiate_nack_compression_no_double_retry() -> None: + # Second attempt also NACKs 0x02 → give up (no infinite recursion). + dev, conn = make_device([start_nack(0x02), start_nack(0x02)], max_queue_size=16) + assert await dev._negotiate_pipe(compressed=True, total_size=10) is None + assert len([w for w in conn.written if w[:2] == b"\x00\x80"]) == 2 + + +# ─── Pipe-partial negotiation (_negotiate_pipe_partial) ────────────────────── + + +@pytest.mark.asyncio +async def test_negotiate_partial_ack_bit1_returns_partial_params() -> None: + dev, conn = make_device([start_ack(flags=0x03)], blocks_per_ack=8, max_queue_size=16) + params = await dev._negotiate_pipe_partial( + compressed=False, total_size=16, old_etag=0x01020304, region=make_region() + ) + assert params == PipeParams(window=16, ack_every=8, max_frame=244, selective=True, compressed=False, partial=True) + # The 0x0080 carried the partial flag + geometry (24-byte packet). + start = conn.written[0] + assert start[:2] == b"\x00\x80" + assert start[3] & PIPE_FLAG_PARTIAL + assert len(start) == 24 + # A valid ACK is a valid pipe probe. + assert dev._pipe_probed is True + assert dev._pipe_supported is True + assert dev._pipe_partial_supported is True + + +@pytest.mark.asyncio +async def test_negotiate_partial_ack_bit1_clear_caches_negative() -> None: + # Device ACKs pipe write but does NOT set the partial bit → unsupported. + dev, conn = make_device([start_ack(flags=0x01)], max_queue_size=16) + params = await dev._negotiate_pipe_partial(compressed=False, total_size=16, old_etag=0x1, region=make_region()) + assert params is None + assert dev._pipe_supported is True # pipe write itself works + assert dev._pipe_partial_supported is False + + +@pytest.mark.asyncio +async def test_negotiate_partial_nack_etag_mismatch_raises() -> None: + dev, _ = make_device([start_nack(0x05)], max_queue_size=16) + with pytest.raises(_PipePartialEtagMismatch): + await dev._negotiate_pipe_partial(compressed=False, total_size=16, old_etag=0x1, region=make_region()) + + +@pytest.mark.asyncio +async def test_negotiate_partial_nack_unsupported_raises_and_caches() -> None: + dev, _ = make_device([start_nack(0x06)], max_queue_size=16) + with pytest.raises(_PipePartialRejected): + await dev._negotiate_pipe_partial(compressed=False, total_size=16, old_etag=0x1, region=make_region()) + assert dev._pipe_partial_supported is False # 0x06 caches negative + + +@pytest.mark.asyncio +async def test_negotiate_partial_nack_rect_invalid_raises_no_cache() -> None: + dev, _ = make_device([start_nack(0x07)], max_queue_size=16) + with pytest.raises(_PipePartialRejected): + await dev._negotiate_pipe_partial(compressed=False, total_size=16, old_etag=0x1, region=make_region()) + # Rect-invalid is per-request, not a capability signal → no negative cache. + assert dev._pipe_partial_supported is None + + +@pytest.mark.asyncio +async def test_negotiate_partial_compressed_02_retries_uncompressed_then_caches() -> None: + # Compressed partial → 0x02 → retry uncompressed-still-partial → 0x02 → give up. + dev, conn = make_device([start_nack(0x02), start_nack(0x02)], max_queue_size=16) + params = await dev._negotiate_pipe_partial(compressed=True, total_size=16, old_etag=0x1, region=make_region()) + assert params is None + starts = [w for w in conn.written if w[:2] == b"\x00\x80"] + assert len(starts) == 2 + # Both retained the partial flag; the retry dropped only the compressed bit. + assert starts[0][3] == PIPE_FLAG_COMPRESSED | PIPE_FLAG_PARTIAL # 0x03 + assert starts[1][3] == PIPE_FLAG_PARTIAL # 0x02 + assert dev._pipe_partial_supported is False + + +@pytest.mark.asyncio +async def test_negotiate_partial_silence_returns_none() -> None: + dev, _ = make_device([BLETimeoutError], max_queue_size=16) + params = await dev._negotiate_pipe_partial(compressed=False, total_size=16, old_etag=0x1, region=make_region()) + assert params is None + assert dev._pipe_probed is True + assert dev._pipe_supported is False + + +# ─── Routing / probe-cache via _execute_upload ─────────────────────────────── + + +@pytest.mark.asyncio +async def test_max_queue_size_one_skips_probe() -> None: + """max_queue_size <= 1 → no 0x0080 at all; straight to legacy uncompressed.""" + dev, conn = make_device( + [b"\x00\x70", b"\x00\x71", b"\x00\x72", REFRESH_COMPLETE], + max_queue_size=1, + ) + await dev._execute_upload(b"ABCD", RefreshMode.FULL, use_compression=False) + assert conn.written[0][:2] == b"\x00\x70" # no probe first + assert all(w[:2] != b"\x00\x80" for w in conn.written) + + +@pytest.mark.asyncio +async def test_silence_falls_back_to_legacy_and_caches() -> None: + dev, conn = make_device( + [ + BLETimeoutError, # 0x0080 probe → silence + b"\x00\x70", # legacy START ACK + b"\x00\x71", # data ACK + b"\x00\x72", # END ACK + REFRESH_COMPLETE, + ], + max_queue_size=16, + ) + await dev._execute_upload(b"ABCD", RefreshMode.FULL, use_compression=False) + assert conn.written[0][:2] == b"\x00\x80" # probed once + assert conn.written[1][:2] == b"\x00\x70" # then legacy + assert dev._pipe_probed is True + assert dev._pipe_supported is False + + +@pytest.mark.asyncio +async def test_probe_cache_skips_second_probe() -> None: + dev, conn = make_device( + [ + BLETimeoutError, + b"\x00\x70", + b"\x00\x71", + b"\x00\x72", + REFRESH_COMPLETE, + # second upload: legacy only, no probe + b"\x00\x70", + b"\x00\x71", + b"\x00\x72", + REFRESH_COMPLETE, + ], + max_queue_size=16, + ) + await dev._execute_upload(b"ABCD", RefreshMode.FULL, use_compression=False) + n_probes_1 = len([w for w in conn.written if w[:2] == b"\x00\x80"]) + await dev._execute_upload(b"EFGH", RefreshMode.FULL, use_compression=False) + n_probes_2 = len([w for w in conn.written if w[:2] == b"\x00\x80"]) + assert n_probes_1 == 1 + assert n_probes_2 == 1 # no additional probe on the second upload + + +@pytest.mark.asyncio +async def test_disconnect_resets_pipe_cache() -> None: + dev, _ = make_device([], max_queue_size=16) + dev._pipe_probed = True + dev._pipe_supported = False + dev._pipe_params = PipeParams(8, 4, 244, True, False) + dev._on_ble_disconnect() + assert dev._pipe_probed is False + assert dev._pipe_supported is False + assert dev._pipe_params is None + + +def test_supports_pipe_write_bit() -> None: + assert make_config(transmission_modes=0x10).displays[0].supports_pipe_write is True + assert make_config(transmission_modes=0x02).displays[0].supports_pipe_write is False + + +def test_pipe_frame_overhead_constant() -> None: + # Plaintext data capacity at 244: 244 - 3 = 241. + assert DEFAULT_MAX_FRAME - PIPE_FRAME_OVERHEAD == 241 + + +@pytest.mark.asyncio +async def test_execute_upload_uncompressed_pipe_auto_completes_end_to_end() -> None: + """Uncompressed pipe upload through _execute_upload: firmware flush-ACKs then + auto-completes with an unsolicited END_ACK — no explicit END is sent and the + etag is reported as NOT committed (auto-complete stores no etag).""" + dev, conn = make_device( + [start_ack(), data_ack({0}), END_ACK, REFRESH_COMPLETE], + max_queue_size=16, + ) + committed = await dev._execute_upload(b"ABCD", RefreshMode.FULL, use_compression=False, new_etag=0x1234) + assert committed is False # etag never committed on auto-complete + assert any(w[:2] == b"\x00\x81" for w in conn.written) # data went via pipe frames + assert all(w[:2] != b"\x00\x82" for w in conn.written) # no spurious END + assert all(w[:2] != b"\x00\x70" for w in conn.written) # never fell back to legacy + + +@pytest.mark.asyncio +async def test_execute_upload_skips_pipe_without_config_bit() -> None: + """Config without transmission_modes bit 0x10: no 0x0080 probe is ever sent + and the upload goes straight down the legacy path, even with a pipe-sized + queue. The config bit is a hard pre-flight gate; negotiation only runs + when it passes.""" + dev, conn = make_device( + [b"\x00\x70", b"\x00\x71", b"\x00\x72", REFRESH_COMPLETE], + max_queue_size=16, + transmission_modes=0x02, # zip only — pipe bit clear + ) + await dev._execute_upload(b"ABCD", RefreshMode.FULL, use_compression=False) + assert all(w[:2] != b"\x00\x80" for w in conn.written) # never probed + assert conn.written[0] == b"\x00\x70" # legacy START + assert not dev._pipe_probed # gate skipped before any probe caching + + +@pytest.mark.asyncio +async def test_legacy_uncompressed_flow_byte_identical() -> None: + """With pipe disabled, the legacy uncompressed wire bytes are unchanged.""" + dev, conn = make_device( + [b"\x00\x70", b"\x00\x71", b"\x00\x72", REFRESH_COMPLETE], + max_queue_size=1, + ) + await dev._execute_upload(b"ABCD", RefreshMode.FULL, use_compression=False) + assert conn.written == [ + b"\x00\x70", # START (uncompressed) + b"\x00\x71ABCD", # single DATA chunk + b"\x00\x72\x00", # END, refresh mode 0 + ] + # DATA chunk uses Write Without Response; START/END use Write Request. + assert conn.write_responses == [True, False, True] + # Legacy path drains stale before every write (default True). + assert conn.drain_flags == [True, True, True] + + +def test_pipeline_chunks_removed() -> None: + import opendisplay.protocol.commands as commands + + assert not hasattr(commands, "PIPELINE_CHUNKS") + assert CommandCode.PIPE_WRITE_START == 0x0080 + assert CommandCode.PIPE_WRITE_DATA == 0x0081 + assert CommandCode.PIPE_WRITE_END == 0x0082 diff --git a/tests/unit/test_pipe_write_sender.py b/tests/unit/test_pipe_write_sender.py new file mode 100644 index 0000000..9eaed7d --- /dev/null +++ b/tests/unit/test_pipe_write_sender.py @@ -0,0 +1,634 @@ +"""Unit tests for the PIPE_WRITE sliding-window sender loop. + +Exercises _send_pipe_chunks / _run_pipe_upload / _await_pipe_end_ack directly: +happy path, span-window invariant, seq wrap past 255, selective-repeat loss +recovery, rewind fallback, retransmit pacing, lost-ACK supersession, PTO/RETX +aborts, encryption specifics, and drain_stale enforcement. +""" + +from __future__ import annotations + +import pytest +from test_pipe_write import ( + END_ACK, + END_NACK, + REFRESH_COMPLETE, + ScriptedConn, + data_ack, + data_nack, + make_device, + start_ack, +) + +from opendisplay.crypto import decrypt_response, derive_session_id, derive_session_key +from opendisplay.exceptions import BLETimeoutError, ProtocolError +from opendisplay.models.enums import RefreshMode +from opendisplay.protocol import PipeParams +from opendisplay.protocol.commands import ENCRYPTED_CHUNK_SIZE + + +def _params( + window: int = 4, + ack_every: int = 2, + max_frame: int = 244, + selective: bool = True, + compressed: bool = True, + partial: bool = False, +) -> PipeParams: + """Default compressed=True: the explicit-END contract, where the sender returns + False after all chunks are acked. Uncompressed full-frame transfers ALWAYS + auto-complete (firmware sends an unsolicited END_ACK) — tested separately + below. partial=True also uses the explicit-END contract (partial never + auto-completes, Part 1 §1.5).""" + return PipeParams(window, ack_every, max_frame, selective, compressed, partial=partial) + + +def _data_frames(conn: ScriptedConn) -> list[bytes]: + return [w for w in conn.written if w[:2] == b"\x00\x81"] + + +def _seqs(conn: ScriptedConn) -> list[int]: + return [w[2] for w in _data_frames(conn)] + + +# ─── Happy path ────────────────────────────────────────────────────────────── + + +@pytest.mark.asyncio +async def test_send_all_within_one_window() -> None: + dev, conn = make_device([data_ack({0, 1, 2, 3})], max_queue_size=4) + chunks = [b"a", b"b", b"c", b"d"] + auto = await dev._send_pipe_chunks(chunks, _params(window=4), chunk_timeout=5.0) + assert auto is False + assert _seqs(conn) == [0, 1, 2, 3] + assert all(r is False for r in conn.write_responses) # data frames are WNR + + +@pytest.mark.asyncio +async def test_span_window_invariant_multi_window() -> None: + """With W=3 and 6 chunks, exactly W frames go out before each ACK refunds.""" + dev, conn = make_device([data_ack({0, 1, 2}), data_ack({0, 1, 2, 3, 4, 5})], max_queue_size=3) + chunks = [bytes([i]) for i in range(6)] + await dev._send_pipe_chunks(chunks, _params(window=3), chunk_timeout=5.0) + # writes_at_read snapshots the cumulative write count at each blocking read. + assert conn.writes_at_read[0] == 3 # only W sent before the first ACK + assert conn.writes_at_read[1] == 6 # next window refunded → remaining 3 + assert _seqs(conn) == [0, 1, 2, 3, 4, 5] + + +def _cumulative_acks(n: int, step: int) -> list[bytes]: + """Cumulative ACKs advancing by ``step`` (<= 32) up to n chunks.""" + acks = [] + k = step + while k < n: + acks.append(data_ack(set(range(k)))) + k += step + acks.append(data_ack(set(range(n)))) + return acks + + +@pytest.mark.asyncio +async def test_seq_wraps_past_255() -> None: + n = 260 + # W=32 (the structural cap); step ACKs by 32 so each 32-bit mask fully + # describes the in-flight range. + dev, conn = make_device(_cumulative_acks(n, 32), max_queue_size=32) + chunks = [bytes([i % 251]) for i in range(n)] + await dev._send_pipe_chunks(chunks, _params(window=32), chunk_timeout=5.0) + seqs = _seqs(conn) + assert len(seqs) == n # each chunk sent exactly once (no loss) + assert seqs[255] == 255 + assert seqs[256] == 0 # wrapped + assert seqs[259] == 3 + + +@pytest.mark.asyncio +async def test_ack_cadence_multiple_acks() -> None: + # W=4, three ACKs each advancing by two chunks over 6 chunks. + dev, conn = make_device( + [data_ack({0, 1}), data_ack({0, 1, 2, 3}), data_ack({0, 1, 2, 3, 4, 5})], + max_queue_size=4, + ) + chunks = [bytes([i]) for i in range(6)] + auto = await dev._send_pipe_chunks(chunks, _params(window=4, ack_every=2), chunk_timeout=5.0) + assert auto is False + assert sorted(_seqs(conn)) == [0, 1, 2, 3, 4, 5] + + +# ─── Auto-complete / NACK ──────────────────────────────────────────────────── + + +@pytest.mark.asyncio +async def test_auto_complete_end_ack_is_terminal() -> None: + """An unsolicited END_ACK during send → terminal success, no explicit END.""" + dev, conn = make_device([END_ACK], max_queue_size=8) + auto = await dev._send_pipe_chunks([b"x", b"y"], _params(window=8), chunk_timeout=5.0) + assert auto is True + # Data frames were sent, but the sender returns before emitting any 0x82 END. + assert _seqs(conn) == [0, 1] + + +@pytest.mark.asyncio +async def test_fatal_data_nack_aborts() -> None: + dev, conn = make_device([data_nack(0x03, 1, 0x3)], max_queue_size=4) + with pytest.raises(ProtocolError, match="NACK"): + await dev._send_pipe_chunks([b"a", b"b"], _params(window=4), chunk_timeout=5.0) + + +# ─── Loss recovery (selective repeat, bit0 set) ────────────────────────────── + + +@pytest.mark.asyncio +async def test_selective_repeat_only_missing_chunk() -> None: + """A hole at chunk 1 → ONLY chunk 1 retransmitted; window frozen until repair.""" + # 4 chunks, W=4. First ACK: received {0,2,3} (chunk 1 missing). Window_base + # freezes at 1. Second ACK: {0,1,2,3} all received → done. + ack1 = data_ack({0, 2, 3}) + ack2 = data_ack({0, 1, 2, 3}) + dev, conn = make_device([ack1, ack2], max_queue_size=4) + chunks = [b"a", b"b", b"c", b"d"] + await dev._send_pipe_chunks(chunks, _params(window=4), chunk_timeout=5.0) + seqs = _seqs(conn) + # Initial send 0,1,2,3 then exactly ONE retransmit of seq 1. + assert seqs == [0, 1, 2, 3, 1] + assert seqs.count(1) == 2 # only chunk 1 resent + for s in (0, 2, 3): + assert seqs.count(s) == 1 # nothing else resent + + +@pytest.mark.asyncio +async def test_window_freezes_at_hole() -> None: + """A hole must freeze window_base so no chunk beyond span is sent.""" + # 6 chunks, W=3. First window sends 0,1,2. ACK reports {0,2} (hole at 1), + # window_base stays 0 → span rule blocks sending 3+. Only retransmit of 1. + ack1 = data_ack({0, 2}) + ack2 = data_ack({0, 1, 2, 3, 4, 5}) + dev, conn = make_device([ack1, ack2], max_queue_size=3) + chunks = [bytes([i]) for i in range(6)] + await dev._send_pipe_chunks(chunks, _params(window=3), chunk_timeout=5.0) + # After first window (0,1,2) and ack {0,2}: window_base=0 (1 missing), so no + # new chunks; retransmit 1. writes at first read = 3. + assert conn.writes_at_read[0] == 3 + # The retransmit of chunk 1 happens before the second read. + assert 1 in _seqs(conn)[3:4] + + +@pytest.mark.asyncio +async def test_rewind_fallback_when_bit0_clear() -> None: + """selective=False → rewind: resend from window_base, not just the hole.""" + ack1 = data_ack({0, 2, 3}) # hole at 1 + ack2 = data_ack({0, 1, 2, 3}) + dev, conn = make_device([ack1, ack2], max_queue_size=4) + chunks = [b"a", b"b", b"c", b"d"] + await dev._send_pipe_chunks(chunks, _params(window=4, selective=False), chunk_timeout=5.0) + seqs = _seqs(conn) + # Rewind sets next_to_send=window_base(=1) → chunks 1,2,3 resent. + assert seqs[:4] == [0, 1, 2, 3] + assert seqs[4:] == [1, 2, 3] + + +@pytest.mark.asyncio +async def test_retransmit_repaced_per_new_ack() -> None: + """Two successive ACKs still showing the hole → chunk retransmitted each time.""" + ack1 = data_ack({0, 2, 3}) + ack2 = data_ack({0, 2, 3}) # still missing 1 + ack3 = data_ack({0, 1, 2, 3}) + dev, conn = make_device([ack1, ack2, ack3], max_queue_size=4) + chunks = [b"a", b"b", b"c", b"d"] + await dev._send_pipe_chunks(chunks, _params(window=4), chunk_timeout=5.0) + seqs = _seqs(conn) + # Initial 0,1,2,3 + retransmit on ack1 + retransmit on ack2 = chunk 1 sent 3x. + assert seqs.count(1) == 3 + + +@pytest.mark.asyncio +async def test_lost_ack_superseded_no_spurious_retransmit() -> None: + """A dropped ACK followed by a later cumulative ACK → no retransmit, advance.""" + # W=4, 4 chunks. The receiver's first ACK (for {0,1}) is 'lost'; we only + # deliver the cumulative ACK for all four. No hole exists → no retransmit. + dev, conn = make_device([data_ack({0, 1, 2, 3})], max_queue_size=4) + chunks = [b"a", b"b", b"c", b"d"] + await dev._send_pipe_chunks(chunks, _params(window=4), chunk_timeout=5.0) + seqs = _seqs(conn) + assert seqs == [0, 1, 2, 3] # each sent exactly once + + +# ─── PTO / RETX aborts ─────────────────────────────────────────────────────── + + +@pytest.mark.asyncio +async def test_pto_probe_then_recover() -> None: + """A read timeout triggers a PTO probe (resend oldest unacked), then recovers.""" + dev, conn = make_device([BLETimeoutError, data_ack({0, 1})], max_queue_size=4) + chunks = [b"a", b"b"] + await dev._send_pipe_chunks(chunks, _params(window=4), chunk_timeout=1.0) + seqs = _seqs(conn) + # 0,1 sent; timeout → resend oldest unacked (0); then ACK completes. + assert seqs == [0, 1, 0] + + +@pytest.mark.asyncio +async def test_max_pto_aborts() -> None: + dev, conn = make_device([BLETimeoutError, BLETimeoutError, BLETimeoutError], max_queue_size=4) + with pytest.raises(ProtocolError, match="PTO"): + await dev._send_pipe_chunks([b"a"], _params(window=4), chunk_timeout=0.5) + + +@pytest.mark.asyncio +async def test_max_retx_aborts() -> None: + """ACKs perpetually reporting the same hole → abort after 3*W retransmits.""" + # W=2 → MAX_RETX=6. Deliver many ACKs still missing chunk 1. + acks = [data_ack({0, 2})] * 20 # chunk 1 always missing (2 received, 3 not sent yet) + dev, conn = make_device(acks, max_queue_size=2) + chunks = [b"a", b"b", b"c"] + with pytest.raises(ProtocolError, match="MAX_RETX"): + await dev._send_pipe_chunks(chunks, _params(window=2), chunk_timeout=5.0) + + +# ─── Compressed tail-flush (regression: MAJOR 1) ───────────────────────────── + + +@pytest.mark.asyncio +async def test_compressed_tail_stall_short_timeout_and_dup_probe() -> None: + """n_chunks % N != 0: the tail never earns a cadence ACK — the sender must + wait only the short flush timeout, dup-probe, and finish on the elicited + duplicate-ACK. The scripted firmware ACKs ONLY on cadence and in response to + the duplicate (no convenient full-coverage ACK is volunteered).""" + # 3 chunks, N=2: cadence ACK covers {0,1}; chunk 2 is accepted silently. + dev, conn = make_device( + [data_ack({0, 1}), BLETimeoutError, data_ack({0, 1, 2})], + max_queue_size=4, + ) + chunks = [b"a", b"b", b"c"] + auto = await dev._send_pipe_chunks(chunks, _params(window=4, ack_every=2, compressed=True), chunk_timeout=5.0) + assert auto is False # compressed → caller still sends the explicit END + # First read (cadence ACK owed: 3 unacked >= N) used the full chunk timeout; + # the tail waits (< N unacked, no holes) used the short flush timeout. + assert conn.timeouts[0] == pytest.approx(5.0) + assert conn.timeouts[1] == pytest.approx(0.5) + assert conn.timeouts[2] == pytest.approx(0.5) + # The dup-probe resent exactly the oldest unacked (tail) chunk — this also + # repairs a genuinely lost tail chunk instead of surfacing a fatal END NACK. + assert _seqs(conn) == [0, 1, 2, 2] + + +@pytest.mark.asyncio +async def test_compressed_tail_flush_not_used_while_cadence_ack_owed() -> None: + """With unacked >= N the firmware still owes a cadence ACK → full timeout.""" + # 4 chunks, N=2: after sending all, 4 unacked >= N → no tail-flush shortcut. + dev, conn = make_device([data_ack({0, 1, 2, 3})], max_queue_size=4) + await dev._send_pipe_chunks( + [b"a", b"b", b"c", b"d"], _params(window=4, ack_every=2, compressed=True), chunk_timeout=5.0 + ) + assert conn.timeouts == [pytest.approx(5.0)] + + +@pytest.mark.asyncio +async def test_compressed_tail_flush_not_used_with_known_hole() -> None: + """A known hole keeps the normal loss-recovery timeout, not the tail flush.""" + # 3 chunks, N=2. ACK {0,2}: hole at 1 → next read must use chunk_timeout + # (retransmit pacing path), not the short tail-flush timeout. + dev, conn = make_device([data_ack({0, 2}), data_ack({0, 1, 2})], max_queue_size=4) + await dev._send_pipe_chunks([b"a", b"b", b"c"], _params(window=4, ack_every=2, compressed=True), chunk_timeout=5.0) + assert conn.timeouts[1] == pytest.approx(5.0) # hole known → full timeout + + +# ─── Uncompressed auto-complete contract (regression: MAJOR 2) ─────────────── + + +@pytest.mark.asyncio +async def test_uncompressed_waits_for_unsolicited_end_ack() -> None: + """Actual firmware ordering: flush-ACK (acks everything) THEN unsolicited + {0x00,0x82}. The sender must NOT stop at the flush-ACK — it keeps reading + until the END_ACK and reports auto-complete.""" + dev, conn = make_device([data_ack({0, 1}), END_ACK], max_queue_size=4) + auto = await dev._send_pipe_chunks([b"a", b"b"], _params(window=4, compressed=False), chunk_timeout=90.0) + assert auto is True + # No explicit END is ever written on the uncompressed path. + assert all(w[:2] != b"\x00\x82" for w in conn.written) + + +@pytest.mark.asyncio +async def test_uncompressed_missing_end_ack_aborts() -> None: + """All chunks acked but the auto-complete END_ACK never arrives → abort + loudly (nothing left to probe), not a spurious END.""" + dev, conn = make_device([data_ack({0}), BLETimeoutError], max_queue_size=8) + with pytest.raises(ProtocolError, match="auto-complete"): + await dev._send_pipe_chunks([b"z"], _params(window=8, compressed=False), chunk_timeout=0.5) + assert all(w[:2] != b"\x00\x82" for w in conn.written) + + +# ─── Uncompressed partial explicit-END contract (Part 1 §1.5) ──────────────── + + +@pytest.mark.asyncio +async def test_uncompressed_partial_returns_false_after_all_acked() -> None: + """Uncompressed *partial* never auto-completes: once every chunk is acked the + sender breaks and returns False (caller sends the explicit END) — it must NOT + keep reading for an unsolicited END_ACK the way full-frame uncompressed does.""" + dev, conn = make_device([data_ack({0, 1})], max_queue_size=4) + auto = await dev._send_pipe_chunks( + [b"a", b"b"], _params(window=4, compressed=False, partial=True), chunk_timeout=5.0 + ) + assert auto is False # explicit-END contract → caller sends END + assert all(w[:2] != b"\x00\x82" for w in conn.written) # sender never emits END itself + + +@pytest.mark.asyncio +async def test_uncompressed_partial_tail_flush_dup_probe() -> None: + """A < N_eff partial tail earns no cadence ACK → short flush timeout + dup-probe, + exactly like the compressed tail (the explicit_end substitution).""" + dev, conn = make_device( + [data_ack({0, 1}), BLETimeoutError, data_ack({0, 1, 2})], + max_queue_size=4, + ) + auto = await dev._send_pipe_chunks( + [b"a", b"b", b"c"], _params(window=4, ack_every=2, compressed=False, partial=True), chunk_timeout=5.0 + ) + assert auto is False + assert conn.timeouts[0] == pytest.approx(5.0) # cadence ACK owed (3 unacked >= N) + assert conn.timeouts[1] == pytest.approx(0.5) # tail flush (1 unacked < N, no holes) + assert conn.timeouts[2] == pytest.approx(0.5) + assert _seqs(conn) == [0, 1, 2, 2] # dup-probe resent the oldest unacked tail chunk + + +@pytest.mark.asyncio +async def test_single_chunk_partial_completes() -> None: + dev, conn = make_device([data_ack({0})], max_queue_size=8) + auto = await dev._send_pipe_chunks([b"z"], _params(window=8, compressed=False, partial=True), chunk_timeout=5.0) + assert auto is False + assert _seqs(conn) == [0] + + +@pytest.mark.asyncio +async def test_run_pipe_upload_partial_auto_complete_is_contract_violation() -> None: + """If firmware wrongly auto-completes a partial transfer, _run_pipe_upload must + raise (an unsolicited END_ACK would have refreshed FULL with no committed etag).""" + dev, conn = make_device([END_ACK], max_queue_size=8) + with pytest.raises(ProtocolError, match="auto-completed unexpectedly"): + await dev._run_pipe_upload( + b"z", + _params(window=8, compressed=False, partial=True), + RefreshMode.PARTIAL, + progress_callback=None, + new_etag=0xABCD, + ) + + +@pytest.mark.asyncio +async def test_run_pipe_upload_partial_sends_explicit_end_selector_2() -> None: + """Partial upload always sends END = [0x02][new_etag BE].""" + dev, conn = make_device([data_ack({0}), END_ACK, REFRESH_COMPLETE], max_queue_size=8) + committed = await dev._run_pipe_upload( + b"z", + _params(window=8, compressed=False, partial=True), + RefreshMode.PARTIAL, + progress_callback=None, + new_etag=0xABCD, + ) + assert committed is True + end_frames = [w for w in conn.written if w[:2] == b"\x00\x82"] + assert len(end_frames) == 1 + assert end_frames[0] == b"\x00\x82" + bytes([2]) + (0xABCD).to_bytes(4, "big") + + +# ─── _run_pipe_upload / _await_pipe_end_ack ────────────────────────────────── + + +@pytest.mark.asyncio +async def test_run_pipe_upload_compressed_full_flow_with_etag() -> None: + """Compressed contract unchanged: explicit END (with etag) is still sent.""" + dev, conn = make_device( + [data_ack({0}), END_ACK, REFRESH_COMPLETE], + max_queue_size=8, + ) + committed = await dev._run_pipe_upload( + b"hello", + _params(window=8, compressed=True), + RefreshMode.FULL, + progress_callback=None, + new_etag=0xABCD, + ) + assert committed is True # END-with-etag was sent + end_frames = [w for w in conn.written if w[:2] == b"\x00\x82"] + assert len(end_frames) == 1 + assert end_frames[0] == b"\x00\x82" + bytes([0]) + (0xABCD).to_bytes(4, "big") + + +@pytest.mark.asyncio +async def test_run_pipe_upload_uncompressed_auto_completes_no_end_no_etag() -> None: + """Uncompressed: firmware flush-ACKs then auto-completes with an unsolicited + END_ACK, resetting pipe state — the client must not send END, and must NOT + report the etag as committed (firmware auto-complete stores no etag; a + recorded phantom etag would poison every subsequent partial update).""" + dev, conn = make_device([data_ack({0}), END_ACK, REFRESH_COMPLETE], max_queue_size=8) + committed = await dev._run_pipe_upload( + b"z", + _params(window=8, compressed=False), + RefreshMode.FULL, + progress_callback=None, + new_etag=0xABCD, + ) + assert committed is False # etag never committed on auto-complete + assert all(w[:2] != b"\x00\x82" for w in conn.written) # no spurious END + + +@pytest.mark.asyncio +async def test_run_pipe_upload_no_etag_returns_false() -> None: + dev, conn = make_device([data_ack({0}), END_ACK, REFRESH_COMPLETE], max_queue_size=8) + committed = await dev._run_pipe_upload( + b"z", _params(window=8), RefreshMode.FULL, progress_callback=None, new_etag=None + ) + assert committed is False # no etag committed + + +@pytest.mark.asyncio +async def test_run_pipe_upload_auto_complete_skips_end() -> None: + """Firmware auto-completes (END_ACK mid-send) → no explicit 0x82 END written.""" + dev, conn = make_device([END_ACK, REFRESH_COMPLETE], max_queue_size=8) + committed = await dev._run_pipe_upload( + b"z", _params(window=8), RefreshMode.FULL, progress_callback=None, new_etag=0xAB + ) + assert committed is False # auto-completed → etag never committed + assert all(w[:2] != b"\x00\x82" for w in conn.written) + + +@pytest.mark.asyncio +async def test_await_end_ack_skips_tail_flush_ack() -> None: + # A trailing PIPE_ACK precedes the END_ACK — must be ignored. + dev, conn = make_device([data_ack({0, 1}), data_ack({0, 1}), END_ACK, REFRESH_COMPLETE], max_queue_size=8) + await dev._run_pipe_upload(b"ab", _params(window=8), RefreshMode.FULL, progress_callback=None, new_etag=None) + # Completed without error. + assert any(w[:2] == b"\x00\x82" for w in conn.written) + + +@pytest.mark.asyncio +async def test_await_end_nack_aborts() -> None: + dev, conn = make_device([data_ack({0}), END_NACK], max_queue_size=8) + with pytest.raises(ProtocolError, match="END NACK"): + await dev._run_pipe_upload(b"z", _params(window=8), RefreshMode.FULL, progress_callback=None, new_etag=None) + + +@pytest.mark.asyncio +async def test_refresh_timeout_raises() -> None: + dev, conn = make_device([data_ack({0}), END_ACK, b"\x00\x74"], max_queue_size=8) + with pytest.raises(ProtocolError, match="refresh timed out"): + await dev._run_pipe_upload(b"z", _params(window=8), RefreshMode.FULL, progress_callback=None, new_etag=None) + + +# ─── drain_stale enforcement ───────────────────────────────────────────────── + + +@pytest.mark.asyncio +async def test_all_in_stream_writes_pass_drain_stale_false() -> None: + dev, conn = make_device([data_ack({0, 1}), END_ACK, REFRESH_COMPLETE], max_queue_size=8) + await dev._run_pipe_upload(b"ab", _params(window=8), RefreshMode.FULL, progress_callback=None, new_etag=None) + # Data (0x81) and END (0x82) writes → drain_stale False. + for frame, drain in zip(conn.written, conn.drain_flags): + if frame[:2] in (b"\x00\x81", b"\x00\x82"): + assert drain is False, f"{frame[:2].hex()} must not drain" + + +@pytest.mark.asyncio +async def test_probe_write_uses_default_drain_stale_true() -> None: + dev, conn = make_device( + [BLETimeoutError, b"\x00\x70", b"\x00\x71", b"\x00\x72", REFRESH_COMPLETE], max_queue_size=8 + ) + await dev._execute_upload(b"AB", RefreshMode.FULL, use_compression=False) + # The 0x0080 probe went through _write → default drain_stale True. + probe_idx = next(i for i, w in enumerate(conn.written) if w[:2] == b"\x00\x80") + assert conn.drain_flags[probe_idx] is True + + +# ─── Encryption specifics ──────────────────────────────────────────────────── + + +def _make_session(dev) -> None: + key = b"\x11" * 16 + cn, sn, did = b"\x01" * 16, b"\x02" * 16, b"\x00\x00\x00\x01" + dev._session_key = derive_session_key(key, cn, sn, did) + dev._session_id = derive_session_id(dev._session_key, cn, sn) + dev._nonce_counter = 0 + + +@pytest.mark.asyncio +async def test_encrypted_frames_and_fresh_nonce_on_retransmit() -> None: + dev, conn = make_device([data_ack({0, 2}), data_ack({0, 1, 2})], max_queue_size=4) + _make_session(dev) + chunks = [b"a", b"b", b"c"] + start_nonce = dev._nonce_counter + await dev._send_pipe_chunks(chunks, _params(window=4), chunk_timeout=5.0) + frames = _data_frames(conn) + # Every data frame is a full CCM envelope (>= 31 bytes). + for f in frames: + assert len(f) >= 31 + # 3 initial sends + 1 retransmit of chunk 1 = 4 encryptions → nonce advanced 4x. + assert dev._nonce_counter == start_nonce + 4 + # The retransmit carries a higher nonce than the original send of chunk 1. + seq1_frames = [f for f in frames if decrypt_response(dev._session_key, f)[1][0] == 1] + assert len(seq1_frames) == 2 + n0 = int.from_bytes(seq1_frames[0][2:18], "big") + n1 = int.from_bytes(seq1_frames[1][2:18], "big") + assert n1 > n0 # fresh, higher nonce on retransmit + + +@pytest.mark.asyncio +async def test_encrypted_seq_is_first_plaintext_byte() -> None: + dev, conn = make_device([data_ack({0, 1, 2, 3})], max_queue_size=4) + _make_session(dev) + chunks = [b"AA", b"BB", b"CC", b"DD"] + await dev._send_pipe_chunks(chunks, _params(window=4), chunk_timeout=5.0) + for expected_seq, f in enumerate(_data_frames(conn)): + _cmd, payload = decrypt_response(dev._session_key, f) + assert payload[0] == expected_seq # seq is the first plaintext byte + assert payload[1:] == chunks[expected_seq] # then the chunk + + +@pytest.mark.asyncio +async def test_encrypted_data_size_212_at_244() -> None: + dev, _ = make_device([], max_queue_size=4) + _make_session(dev) + assert dev._pipe_data_size(244) == 212 # 244 - 31 - 1 + + +@pytest.mark.asyncio +async def test_plaintext_data_size_241_at_244() -> None: + dev, _ = make_device([], max_queue_size=4) + assert dev._pipe_data_size(244) == 241 # 244 - 3 + + +@pytest.mark.asyncio +async def test_encrypted_ack_decrypts_and_classifies() -> None: + """An encrypted 7-byte ACK survives _read decryption and classifies as PIPE_ACK.""" + from opendisplay.crypto import encrypt_command + from opendisplay.protocol.responses import PIPE_FRAME_ACK, classify_pipe_frame + + dev, conn = make_device([], max_queue_size=4) + _make_session(dev) + # Firmware would encrypt {0x00,0x81,hs,mask} via sendResponse: cmd=0x0081, + # payload=[hs,mask:4]. Reproduce that envelope and confirm _read yields the ACK. + ack_plain = data_ack({0, 1, 2}) + env = encrypt_command(dev._session_key, dev._session_id, 0, ack_plain[:2], ack_plain[2:]) + conn._responses = [env] + decoded = await dev._read(timeout=5.0) + assert classify_pipe_frame(decoded) == PIPE_FRAME_ACK + assert decoded == ack_plain + + +@pytest.mark.asyncio +async def test_encrypted_nack_decrypts_and_aborts() -> None: + """Encrypted pipe NACKs (byte[2]=err != 0xFF) are decrypted by _read, then + classified on plaintext and abort the send (firmware clarification #1).""" + from opendisplay.crypto import encrypt_command + + dev, conn = make_device([], max_queue_size=4) + _make_session(dev) + nack_plain = data_nack(0x03, 1, 0x1) # {0xFF,0x81,err,hs,mask} + env = encrypt_command(dev._session_key, dev._session_id, 0, nack_plain[:2], nack_plain[2:]) + assert len(env) >= 31 # genuinely encrypted envelope + conn._responses = [env] + with pytest.raises(ProtocolError, match="NACK"): + await dev._send_pipe_chunks([b"a", b"b"], _params(window=4), chunk_timeout=5.0) + + +def test_encrypted_chunk_capacity_beats_legacy() -> None: + # Sanity: 212 > legacy encrypted 154. + assert 212 > ENCRYPTED_CHUNK_SIZE + + +@pytest.mark.asyncio +async def test_encrypted_full_pipe_end_to_end_through_execute_upload() -> None: + """Encrypted session, FULL-frame pipe, all the way through _execute_upload: + encrypted 0x0080 negotiation, encrypted 0x0081 data, firmware auto-complete + (encrypted unsolicited END_ACK at the 31-byte decrypt floor), encrypted + 0x73. Every host frame must be a valid CCM envelope; no plaintext leak and + no spurious explicit END.""" + from opendisplay.crypto import encrypt_command + + dev, conn = make_device([], max_queue_size=16) + _make_session(dev) + + def enc(plain: bytes, ctr: int) -> bytes: + # Device→host responses use the same CCM envelope as commands. + return encrypt_command(dev._session_key, dev._session_id, ctr, plain[:2], plain[2:]) + + # start_ack (8 B), data ack for chunk 0, unsolicited auto-complete END_ACK + # (2-byte plaintext → 31-byte envelope, the exact decrypt-floor boundary), + # then refresh-complete. + conn._responses = [enc(start_ack(), 0), enc(data_ack({0}), 1), enc(END_ACK, 2), enc(REFRESH_COMPLETE, 3)] + assert len(conn._responses[2]) == 31 # END_ACK rides the >=31 decrypt boundary + + committed = await dev._execute_upload(b"ABCD", RefreshMode.FULL, use_compression=False, new_etag=0x1234) + + assert committed is False # auto-complete never commits an etag + # Every write is encrypted: valid envelope, no plaintext opcode on the wire. + plaintexts = [] + for w in conn.written: + assert len(w) >= 31 + cmd_code, payload = decrypt_response(dev._session_key, w) + plaintexts.append(cmd_code.to_bytes(2, "big") + payload) + assert plaintexts[0][:2] == b"\x00\x80" + assert len(plaintexts[0]) == 12 # full-frame START: no partial extension + assert any(p[:2] == b"\x00\x81" and p[3:] == b"ABCD" for p in plaintexts) + assert all(p[:2] != b"\x00\x82" for p in plaintexts) # auto-complete: no explicit END + assert all(p[:2] != b"\x00\x70" for p in plaintexts) # never fell back to legacy diff --git a/uv.lock b/uv.lock index acf8125..f416ed9 100644 --- a/uv.lock +++ b/uv.lock @@ -891,7 +891,7 @@ wheels = [ [[package]] name = "py-opendisplay" -version = "7.11.2" +version = "7.12.0.dev2" source = { editable = "." } dependencies = [ { name = "bleak" },