diff --git a/FINDINGS.md b/FINDINGS.md index ba7aef7..c5f20c9 100644 --- a/FINDINGS.md +++ b/FINDINGS.md @@ -92,14 +92,17 @@ machinery exists. ### P5. 🟠 BLE throughput: stop-and-wait, write-with-response, fixed chunk sizes -- Every chunk incurs a GATT write-with-response (`transport/connection.py:262-266`, - `response=True`) **and** a firmware ACK notification before the next chunk - (`device.py:1563-1598`). Over an ESPHome/HA proxy that is ~2 RTTs per 154/230-byte - chunk β€” hundreds of serialized round-trips for a 50 KB image. This is the dominant - transfer-time limiter. `PIPELINE_CHUNKS` exists (`protocol/commands.py:51`) but is - hardcoded to 1 and unused. If firmware tolerates it, a small in-flight window or - write-without-response for 0x71 data with periodic ACK checkpoints would multiply - throughput. (Firmware ACKs every 0x71, so this needs a firmware-coordinated change.) +- **Partially addressed:** 0x71 data chunks now use BLE Write Without Response + (`transport/connection.py` `write_command(..., response=False)`, opted in from + `_send_data_chunks`/`_send_partial_chunks`), removing the GATT write-with-response + round-trip. The firmware ACK notification per chunk is still awaited (stop-and-wait), + which preserves flow control and needs no firmware change β€” so exactly one write is in + flight at a time. This roughly halves the RTTs per chunk. The characteristic is probed + for the `write-without-response` property and falls back to write-with-response if absent. +- Remaining: the per-chunk firmware ACK is still serialized. A true in-flight window / + periodic ACK checkpoints would multiply throughput further, but that DOES need a + firmware-coordinated change (firmware ACKs every 0x71, and the ESP32 command queue is + only 5 deep). `PIPELINE_CHUNKS` (`protocol/commands.py:52`) stays 1 for now. - No MTU negotiation: `CHUNK_SIZE = 230` / `ENCRYPTED_CHUNK_SIZE = 154` (`protocol/commands.py:47-48`) are constants; `client.mtu_size` is never consulted, so links that negotiate 247+ leave throughput on the table. diff --git a/src/opendisplay/cli.py b/src/opendisplay/cli.py index 68165d3..a5d135b 100644 --- a/src/opendisplay/cli.py +++ b/src/opendisplay/cli.py @@ -703,6 +703,37 @@ async def _reboot(device_kwargs: dict[str, Any]) -> None: _console.print("Reboot command sent. Device will restart.") +# ── sleep ───────────────────────────────────────────────────────────────────── + + +def _add_sleep_parser(subparsers: argparse._SubParsersAction[argparse.ArgumentParser]) -> None: + p = subparsers.add_parser("sleep", help="Put the device into deep sleep (command 0x0052)") + _add_device_options(p) + p.set_defaults(func=_cmd_sleep) + + +def _cmd_sleep(args: argparse.Namespace) -> None: + key = _parse_hex_key(args.key) + _run(_sleep(_device_kwargs(args.device, key, args.timeout))) + + +async def _sleep(device_kwargs: dict[str, Any]) -> None: + slept = False + with _spinner() as progress: + progress.add_task("Connecting...", total=None) + try: + async with OpenDisplayDevice(**device_kwargs) as device: + await device.deep_sleep() + slept = True + except (BLEConnectionError, BLETimeoutError): + if not slept: + _error("BLE connection failed before deep sleep command could be sent.") + # else: expected drop after the device sleeps + except OpenDisplayError as exc: + _handle_ble_error(exc) + _console.print("Deep sleep command sent. Device will sleep until its next wake.") + + # ── export-config ───────────────────────────────────────────────────────────── @@ -792,6 +823,7 @@ def main() -> None: _add_info_parser(subparsers) _add_upload_parser(subparsers) _add_reboot_parser(subparsers) + _add_sleep_parser(subparsers) _add_export_config_parser(subparsers) _add_write_config_parser(subparsers) diff --git a/src/opendisplay/device.py b/src/opendisplay/device.py index 17b323c..1d400e2 100644 --- a/src/opendisplay/device.py +++ b/src/opendisplay/device.py @@ -71,6 +71,7 @@ build_authenticate_step1, build_authenticate_step2, build_buzzer_activate_command, + build_deep_sleep_command, build_direct_write_data_command, build_direct_write_end_command, build_direct_write_end_with_etag, @@ -558,17 +559,24 @@ def _on_ble_disconnect(self) -> None: _LOGGER.debug("Link to %s dropped; clearing session state", self.mac_address) self._clear_session() - async def _write(self, data: bytes) -> None: - """Write a command, encrypting it if an active session exists.""" + async def _write(self, data: bytes, response: bool = True) -> None: + """Write a command, encrypting it if an active session exists. + + Args: + data: Plaintext command frame (opcode + payload). + response: Passed through to the transport. False requests a BLE Write + Without Response (used for 0x71 data chunks); applies whether or not + the frame is encrypted. + """ 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) + await self._conn.write_command(encrypted, response=response) else: - await self._conn.write_command(data) + await self._conn.write_command(data, response=response) async def _reauthenticate_if_needed(self) -> None: """Re-authenticate proactively at 90% of session_timeout_seconds.""" @@ -950,6 +958,67 @@ async def reboot(self) -> None: # Device will reset immediately - no ACK expected _LOGGER.info("Reboot command sent to %s - device will reset (connection will drop)", self.mac_address) + @_serialized + async def deep_sleep(self) -> None: + """Put the device into deep sleep (command 0x0052). + + Supported on ESP32 and Silabs Flex; nRF targets do not implement deep + sleep. The command is sent encrypted when an active session exists. + + The firmware's exact behavior depends on the target, and this method + tolerates all of them β€” in every supported case the BLE link drops during + or right after the command, so a disconnect (or a missing ACK) is treated + as success, mirroring reboot() and trigger_dfu_bootloader(): + + - ESP32 with a D-FF power latch: firmware ACKs 0x0052, then powers off + after ~100 ms (the link drops). + - ESP32 without a power latch: firmware enters deep sleep immediately, + tearing down BLE with no ACK (the write or read fails as the link drops). + - Silabs Flex: firmware ACKs 0x0052, then closes the connection and enters + EM4 (wake on button/NFC). + + Raises: + ProtocolError: If the device explicitly reports that deep sleep is not + supported (protocol error frame 0xFF52). + """ + from .exceptions import BLEConnectionError + + _LOGGER.debug("Sending deep sleep command (0x0052) to device %s", self.mac_address) + + try: + await self._write(build_deep_sleep_command()) + except BLEConnectionError as exc: + # An ESP32 without a power latch tears down BLE synchronously as it + # enters deep sleep, so the write-with-response confirmation can fail + # (e.g. a GATT/disconnect error over a Bluetooth proxy) even though the + # command was delivered. Treat that as the device having gone to sleep. + _LOGGER.debug( + "Deep sleep write did not confirm (expected β€” device sleeps before responding): %s", + exc, + ) + _LOGGER.info("Deep sleep command sent to %s β€” device is sleeping (connection dropped)", self.mac_address) + return + + # Targets that ACK before sleeping (ESP32 power-latch, Silabs Flex) reply + # with 0x0052 and then drop the link; a device that does not support the + # command replies with the 0xFF52 error frame (protocol: 0xFF [command_low]). + # A disconnect or timeout here means the device slept without acking. + try: + response = await self._read(self.TIMEOUT_ACK) + except (BLEConnectionError, BLETimeoutError) as exc: + _LOGGER.debug( + "No deep sleep ACK (expected β€” device dropped the link or sleeps silently): %s", + exc, + ) + _LOGGER.info("Deep sleep command sent to %s β€” device is sleeping", self.mac_address) + return + + if len(response) >= 2 and unpack_command_code(response) == 0xFF52: + raise ProtocolError("Device reported deep sleep is not supported (command 0x0052)") + + validate_ack_response(response, CommandCode.DEEP_SLEEP) + _LOGGER.info("Deep sleep command acknowledged by %s β€” device is sleeping", self.mac_address) + @_serialized async def trigger_dfu_bootloader(self) -> None: """Trigger the DFU bootloader on nRF devices (command 0x0051). @@ -1532,7 +1601,8 @@ async def _send_partial_chunks( offset = 0 while offset < len(remaining): chunk = remaining[offset : offset + chunk_size] - await self._write(build_direct_write_data_command(chunk)) + # Write Without Response; the per-chunk ACK read below keeps flow control. + await self._write(build_direct_write_data_command(chunk), response=False) ack = await self._read(self.TIMEOUT_ACK) nack = parse_nack(ack) if nack is not None: @@ -1817,7 +1887,10 @@ async def _send_data_chunks( chunk_size = ENCRYPTED_CHUNK_SIZE if self._session_key is not None else CHUNK_SIZE chunk_data = image_data[bytes_sent : bytes_sent + chunk_size] - await self._write(build_direct_write_data_command(chunk_data)) + # Send the data chunk without waiting for the ATT write confirmation + # (Write Without Response). Flow control is preserved by the per-chunk + # application ACK read below, so only one write is ever in flight. + await self._write(build_direct_write_data_command(chunk_data), response=False) bytes_sent += len(chunk_data) chunks_sent += 1 diff --git a/src/opendisplay/models/config.py b/src/opendisplay/models/config.py index 4389ad3..3ddcc82 100644 --- a/src/opendisplay/models/config.py +++ b/src/opendisplay/models/config.py @@ -220,6 +220,15 @@ def power_mode_enum(self) -> PowerMode | int: except ValueError: return self.power_mode + @property + def deep_sleep_enabled(self) -> bool: + """Return True if the device is configured for deep sleep. + + Mirrors the firmware's own sleep-entry condition: battery power mode + (PowerMode.BATTERY) with a non-zero timer-wake interval. + """ + return self.power_mode == PowerMode.BATTERY and self.deep_sleep_time_seconds > 0 + @property def capacity_estimator_enum(self) -> CapacityEstimator | int: """Get battery chemistry estimator as enum, or raw int if unknown.""" diff --git a/src/opendisplay/protocol/__init__.py b/src/opendisplay/protocol/__init__.py index d78d935..eeb2299 100644 --- a/src/opendisplay/protocol/__init__.py +++ b/src/opendisplay/protocol/__init__.py @@ -12,6 +12,7 @@ build_authenticate_step1, build_authenticate_step2, build_buzzer_activate_command, + build_deep_sleep_command, build_direct_write_data_command, build_direct_write_end_command, build_direct_write_end_with_etag, @@ -49,6 +50,7 @@ "build_read_config_command", "build_read_fw_version_command", "build_enter_dfu_command", + "build_deep_sleep_command", "build_reboot_command", "build_write_config_command", "build_direct_write_start_compressed", diff --git a/src/opendisplay/protocol/commands.py b/src/opendisplay/protocol/commands.py index e6b88f2..81afefb 100644 --- a/src/opendisplay/protocol/commands.py +++ b/src/opendisplay/protocol/commands.py @@ -36,6 +36,7 @@ class CommandCode(IntEnum): DIRECT_WRITE_PARTIAL_START = 0x0076 # Start a partial update transfer (stream via 0x71) BUZZER_ACTIVATE = 0x0077 # Hostβ†’device: trigger buzzer pattern (firmware 1.61+) ENTER_DFU = 0x0051 # Trigger DFU bootloader mode (nRF only) + DEEP_SLEEP = 0x0052 # Enter deep sleep now (ESP32 timer-wake / Silabs EM4; nRF unsupported) # Protocol constants @@ -48,7 +49,9 @@ 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 # Wait for ACK after each chunk +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) @@ -97,6 +100,28 @@ def build_enter_dfu_command() -> bytes: return CommandCode.ENTER_DFU.to_bytes(2, byteorder="big") +def build_deep_sleep_command() -> bytes: + """Build command to put the device into deep sleep (command 0x0052). + + Supported on ESP32 (enters timer-wake deep sleep, or releases the D-FF power + latch when one is configured) and Silabs Flex (arms EM4 button/NFC wake and + sleeps once the BLE connection closes). nRF targets do not implement deep + sleep and only log the command. + + The response behavior varies by target and is best-effort β€” callers should + tolerate the connection dropping during or right after the command: + - ESP32 with a power latch: replies 0x0052, then powers off after ~100 ms. + - ESP32 without a power latch: enters deep sleep immediately with no ACK; + the BLE connection drops. + - Silabs Flex: replies 0x0052, then closes the link and enters EM4. + - nRF: no response (deep sleep not supported). + + Returns: + Command bytes: 0x0052 (2 bytes, big-endian) + """ + return CommandCode.DEEP_SLEEP.to_bytes(2, byteorder="big") + + def build_direct_write_start_compressed( uncompressed_size: int, compressed_data: bytes, diff --git a/src/opendisplay/transport/connection.py b/src/opendisplay/transport/connection.py index be29408..7f4a3b0 100644 --- a/src/opendisplay/transport/connection.py +++ b/src/opendisplay/transport/connection.py @@ -60,6 +60,9 @@ def __init__( self._client: BleakClient | None = None self._notification_queue: asyncio.Queue[bytes] = asyncio.Queue() self._notification_characteristic: BleakGATTCharacteristic | None = None + # Whether the command characteristic advertises Write Without Response. + # Set during notification setup; used to safely enable/disable WNR writes. + self._write_no_response_supported: bool = False self.device_name: str | None = None async def __aenter__(self) -> BLEConnection: @@ -231,6 +234,16 @@ async def _setup_notifications(self) -> None: self._notification_characteristic = characteristics[0] + # Record whether the characteristic advertises Write Without Response so + # writes can opt into it (0x71 data chunks) and gracefully fall back to + # write-with-response on devices/stacks that don't support it. + props = getattr(self._notification_characteristic, "properties", []) or [] + self._write_no_response_supported = "write-without-response" in props + _LOGGER.debug( + "Command characteristic write-without-response supported: %s", + self._write_no_response_supported, + ) + # Start notifications await self._client.start_notify( self._notification_characteristic, @@ -283,11 +296,17 @@ def drain_notifications(self) -> int: _LOGGER.warning("Discarded %d stale notification(s) before command", dropped) return dropped - async def write_command(self, data: bytes) -> None: + async def write_command(self, data: bytes, response: bool = True) -> None: """Write command to device. Args: data: Command bytes to write + response: If True, use a BLE Write Request and wait for the ATT write + confirmation. If False, use a Write Without Response (Write Command) + to skip the ATT round-trip β€” used for bulk 0x71 image-data chunks, + 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. Raises: BLEConnectionError: If not connected or write fails @@ -302,11 +321,15 @@ async def write_command(self, data: bytes) -> None: # from a clean queue (see drain_notifications). self.drain_notifications() + # Only skip the write confirmation when the caller opts out AND the + # characteristic actually supports it; otherwise keep write-with-response. + effective_response = response or not self._write_no_response_supported + try: await self._client.write_gatt_char( self._notification_characteristic, data, - response=True, # Wait for write confirmation + response=effective_response, ) except Exception as e: raise BLEConnectionError(f"Write failed: {e}") from e diff --git a/tests/unit/test_auth_server_proof.py b/tests/unit/test_auth_server_proof.py index 10eefdc..4321791 100644 --- a/tests/unit/test_auth_server_proof.py +++ b/tests/unit/test_auth_server_proof.py @@ -21,7 +21,7 @@ class _FakeConn: def __init__(self, responses: list[bytes]) -> None: self._responses = responses - async def write_command(self, data: bytes) -> None: + async def write_command(self, data: bytes, response: bool = True) -> None: pass async def read_response(self, timeout: float) -> bytes: diff --git a/tests/unit/test_connection_write_no_response.py b/tests/unit/test_connection_write_no_response.py new file mode 100644 index 0000000..5451a9e --- /dev/null +++ b/tests/unit/test_connection_write_no_response.py @@ -0,0 +1,77 @@ +"""Tests for BLE Write Without Response on 0x71 data writes (send-without-reply).""" + +from __future__ import annotations + +from unittest.mock import AsyncMock, MagicMock + +import pytest + +from opendisplay.transport.connection import BLEConnection + + +def _make_conn_with_client(supports_wnr: bool) -> tuple[BLEConnection, MagicMock]: + conn = BLEConnection("AA:BB:CC:DD:EE:FF") + client = MagicMock() + client.is_connected = True + client.write_gatt_char = AsyncMock() + conn._client = client + conn._notification_characteristic = MagicMock() + conn._write_no_response_supported = supports_wnr + return conn, client + + +@pytest.mark.asyncio +async def test_write_command_uses_write_without_response_when_supported() -> None: + conn, client = _make_conn_with_client(supports_wnr=True) + await conn.write_command(b"\x00\x71data", response=False) + _, kwargs = client.write_gatt_char.call_args + assert kwargs["response"] is False + + +@pytest.mark.asyncio +async def test_write_command_falls_back_when_wnr_unsupported() -> None: + """If the characteristic doesn't advertise WNR, response=False degrades to a Write Request.""" + conn, client = _make_conn_with_client(supports_wnr=False) + await conn.write_command(b"\x00\x71data", response=False) + _, kwargs = client.write_gatt_char.call_args + assert kwargs["response"] is True + + +@pytest.mark.asyncio +async def test_write_command_defaults_to_write_with_response() -> None: + conn, client = _make_conn_with_client(supports_wnr=True) + await conn.write_command(b"\x00\x70start") + _, kwargs = client.write_gatt_char.call_args + assert kwargs["response"] is True + + +def _make_client_with_char(properties: list[str]) -> MagicMock: + char = MagicMock() + char.properties = properties + service = MagicMock() + service.characteristics = [char] + client = MagicMock() + client.is_connected = True + client.services.get_service.return_value = service + client.start_notify = AsyncMock() + return client + + +@pytest.mark.asyncio +async def test_setup_notifications_detects_wnr_property() -> None: + conn = BLEConnection("AA:BB:CC:DD:EE:FF") + conn._client = _make_client_with_char(["read", "write", "write-without-response", "notify"]) + + await conn._setup_notifications() + + assert conn._write_no_response_supported is True + + +@pytest.mark.asyncio +async def test_setup_notifications_without_wnr_property() -> None: + conn = BLEConnection("AA:BB:CC:DD:EE:FF") + conn._client = _make_client_with_char(["read", "write", "notify"]) + + await conn._setup_notifications() + + assert conn._write_no_response_supported is False diff --git a/tests/unit/test_device_buzzer_activate.py b/tests/unit/test_device_buzzer_activate.py index fe5e795..a3594ff 100644 --- a/tests/unit/test_device_buzzer_activate.py +++ b/tests/unit/test_device_buzzer_activate.py @@ -17,7 +17,7 @@ def __init__(self, response: bytes | list[bytes]): self.written: list[bytes] = [] self.read_timeout: float | None = None - async def write_command(self, cmd: bytes) -> None: + async def write_command(self, cmd: bytes, response: bool = True) -> None: self.written.append(cmd) async def read_response(self, timeout: float) -> bytes: diff --git a/tests/unit/test_device_deep_sleep.py b/tests/unit/test_device_deep_sleep.py new file mode 100644 index 0000000..fa8cc2a --- /dev/null +++ b/tests/unit/test_device_deep_sleep.py @@ -0,0 +1,123 @@ +"""Test deep_sleep() (command 0x0052) on OpenDisplayDevice. + +Firmware ground truth (verified against Firmware/src/device_control.cpp, +Firmware/src/communication.cpp, and Firmware_Silabs/opendisplay_pipe.c): +- ESP32 with a D-FF power latch: replies 0x0052, then powers off after ~100 ms. +- ESP32 without a power latch: enters deep sleep immediately with no ACK; link drops. +- Silabs Flex: replies 0x0052, then closes the connection and enters EM4. +- nRF: does not support the command. + +In every supported case the connection drops during or right after the command, +so a disconnect/missing ACK is treated as success. +""" + +from __future__ import annotations + +import pytest + +from opendisplay import OpenDisplayDevice +from opendisplay.exceptions import BLEConnectionError, BLETimeoutError, ProtocolError + + +class _FakeConnection: + def __init__( + self, + response: bytes | None = None, + *, + write_error: Exception | None = None, + read_error: Exception | None = None, + ): + self._response = response + self._write_error = write_error + self._read_error = read_error + self.written: list[bytes] = [] + self.read_timeout: float | None = None + self.read_called = False + + async def write_command(self, cmd: bytes, response: bool = True) -> None: + self.written.append(cmd) + if self._write_error is not None: + raise self._write_error + + async def read_response(self, timeout: float) -> bytes: + self.read_called = True + self.read_timeout = timeout + if self._read_error is not None: + raise self._read_error + assert self._response is not None + return self._response + + +@pytest.mark.asyncio +async def test_deep_sleep_sends_0052_and_accepts_silabs_ack() -> None: + """Silabs Flex replies with the 2-byte 0x0052 ACK before sleeping.""" + device = OpenDisplayDevice(mac_address="AA:BB:CC:DD:EE:FF") + fake = _FakeConnection(response=b"\x00\x52") + device._connection = fake + + await device.deep_sleep() # must not raise + + assert fake.written == [b"\x00\x52"] + assert fake.read_timeout == device.TIMEOUT_ACK + + +@pytest.mark.asyncio +async def test_deep_sleep_accepts_esp32_power_latch_ack() -> None: + """ESP32 with a power latch replies with the 4-byte 0x0052 0x0000 ACK.""" + device = OpenDisplayDevice(mac_address="AA:BB:CC:DD:EE:FF") + fake = _FakeConnection(response=b"\x00\x52\x00\x00") + device._connection = fake + + await device.deep_sleep() # must not raise + + assert fake.written == [b"\x00\x52"] + + +@pytest.mark.asyncio +async def test_deep_sleep_raises_on_not_supported_nack() -> None: + """A 0xFF52 error frame surfaces as ProtocolError (deep sleep not supported).""" + device = OpenDisplayDevice(mac_address="AA:BB:CC:DD:EE:FF") + fake = _FakeConnection(response=b"\xff\x52") + device._connection = fake + + with pytest.raises(ProtocolError, match="not supported"): + await device.deep_sleep() + + +@pytest.mark.asyncio +async def test_deep_sleep_tolerates_write_drop() -> None: + """An ESP32 without a power latch tears down BLE before the write confirms; + that disconnect is expected and must not raise.""" + device = OpenDisplayDevice(mac_address="AA:BB:CC:DD:EE:FF") + fake = _FakeConnection(write_error=BLEConnectionError("Write failed: ... error=133")) + device._connection = fake + + await device.deep_sleep() # must not raise + + assert fake.written == [b"\x00\x52"] + assert fake.read_called is False # never got to reading a response + + +@pytest.mark.asyncio +async def test_deep_sleep_tolerates_read_timeout() -> None: + """A device that sleeps silently after the write leaves the ACK read to time out; + that is expected and must not raise.""" + device = OpenDisplayDevice(mac_address="AA:BB:CC:DD:EE:FF") + fake = _FakeConnection(read_error=BLETimeoutError("No response received within 5s")) + device._connection = fake + + await device.deep_sleep() # must not raise + + assert fake.written == [b"\x00\x52"] + assert fake.read_called is True + + +@pytest.mark.asyncio +async def test_deep_sleep_tolerates_read_disconnect() -> None: + """A device that drops the link right after acking surfaces as a read-time + connection error; that is expected and must not raise.""" + device = OpenDisplayDevice(mac_address="AA:BB:CC:DD:EE:FF") + fake = _FakeConnection(read_error=BLEConnectionError("Not connected")) + device._connection = fake + + await device.deep_sleep() # must not raise diff --git a/tests/unit/test_device_led_activate.py b/tests/unit/test_device_led_activate.py index 6c9e59d..2b2a541 100644 --- a/tests/unit/test_device_led_activate.py +++ b/tests/unit/test_device_led_activate.py @@ -18,7 +18,7 @@ def __init__(self, response: bytes | list[bytes]): self.written: list[bytes] = [] self.read_timeout: float | None = None - async def write_command(self, cmd: bytes) -> None: + async def write_command(self, cmd: bytes, response: bool = True) -> None: self.written.append(cmd) async def read_response(self, timeout: float) -> bytes: diff --git a/tests/unit/test_device_upload_flow.py b/tests/unit/test_device_upload_flow.py index d30d485..1c27f3c 100644 --- a/tests/unit/test_device_upload_flow.py +++ b/tests/unit/test_device_upload_flow.py @@ -45,11 +45,13 @@ class _FakeConnection: def __init__(self, responses: list[bytes]) -> None: self.written: list[bytes] = [] + self.write_responses: list[bool] = [] self._responses = list(responses) self.timeouts: list[float] = [] - async def write_command(self, data: bytes) -> None: + async def write_command(self, data: bytes, response: bool = True) -> None: self.written.append(data) + self.write_responses.append(response) async def read_response(self, timeout: float) -> bytes: self.timeouts.append(timeout) @@ -210,6 +212,8 @@ async def test_uncompressed_full_sequence() -> None: assert fake.written[0][:2] == b"\x00\x70" # START assert fake.written[1][:2] == b"\x00\x71" # DATA assert fake.written[2][:2] == b"\x00\x72" # END + # 0x71 DATA is sent Write-Without-Response; START/END use write-with-response. + assert fake.write_responses == [True, False, True] # ─── Compressed upload path ─────────────────────────────────────────────────── diff --git a/tests/unit/test_models_config.py b/tests/unit/test_models_config.py index 2c44b58..e7b2d9c 100644 --- a/tests/unit/test_models_config.py +++ b/tests/unit/test_models_config.py @@ -2,15 +2,34 @@ import pytest -from opendisplay.models.config import DisplayConfig, ManufacturerData +from opendisplay.models.config import DisplayConfig, ManufacturerData, PowerOption from opendisplay.models.enums import ( BoardManufacturer, DIYBoardType, + PowerMode, SeeedBoardType, WaveshareBoardType, ) +def _power_option(power_mode: int, deep_sleep_time_seconds: int) -> PowerOption: + return PowerOption( + power_mode=power_mode, + 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=deep_sleep_time_seconds, + reserved=b"\x00" * 10, + ) + + def _display_config(active_width_mm: int, active_height_mm: int) -> DisplayConfig: return DisplayConfig( instance_number=0, @@ -105,6 +124,26 @@ def test_unknown_board_type_falls_back_to_int(self): assert mfg.board_type_name is None +class TestPowerOptionDeepSleepEnabled: + """Test PowerOption.deep_sleep_enabled (mirrors firmware sleep-entry condition).""" + + def test_enabled_when_battery_and_positive_interval(self): + power = _power_option(power_mode=PowerMode.BATTERY, deep_sleep_time_seconds=300) + assert power.deep_sleep_enabled is True + + def test_disabled_when_interval_zero(self): + power = _power_option(power_mode=PowerMode.BATTERY, deep_sleep_time_seconds=0) + assert power.deep_sleep_enabled is False + + def test_disabled_when_not_battery(self): + power = _power_option(power_mode=PowerMode.USB, deep_sleep_time_seconds=300) + assert power.deep_sleep_enabled is False + + def test_disabled_when_usb_and_zero_interval(self): + power = _power_option(power_mode=PowerMode.USB, deep_sleep_time_seconds=0) + assert power.deep_sleep_enabled is False + + class TestDisplayConfigTransmissionModes: """Test DisplayConfig.supports_zip from transmission_modes bitfield.""" diff --git a/tests/unit/test_protocol_commands.py b/tests/unit/test_protocol_commands.py index ff88815..92463b9 100644 --- a/tests/unit/test_protocol_commands.py +++ b/tests/unit/test_protocol_commands.py @@ -7,6 +7,7 @@ CONFIG_CHUNK_SIZE, CommandCode, build_buzzer_activate_command, + build_deep_sleep_command, build_direct_write_data_command, build_direct_write_end_command, build_direct_write_start_compressed, @@ -46,6 +47,13 @@ def test_build_reboot_command(self): assert len(cmd) == 2 assert cmd == b"\x00\x0f" # 0x000F big-endian + def test_build_deep_sleep_command(self): + """Test DEEP_SLEEP command builder.""" + cmd = build_deep_sleep_command() + assert len(cmd) == 2 + assert cmd == b"\x00\x52" # 0x0052 big-endian + assert cmd == CommandCode.DEEP_SLEEP.to_bytes(2, "big") + def test_build_direct_write_start_uncompressed(self, real_upload_start_command): """Test uncompressed START command matches real data.""" cmd = build_direct_write_start_uncompressed() diff --git a/tests/unit/test_protocol_error_frames.py b/tests/unit/test_protocol_error_frames.py index d394122..16ac857 100644 --- a/tests/unit/test_protocol_error_frames.py +++ b/tests/unit/test_protocol_error_frames.py @@ -37,7 +37,7 @@ class _FakeConn: def __init__(self, responses: list[bytes]) -> None: self._responses = responses - async def write_command(self, data: bytes) -> None: + async def write_command(self, data: bytes, response: bool = True) -> None: pass async def read_response(self, timeout: float) -> bytes: @@ -61,7 +61,7 @@ class _ScriptedConn: def __init__(self, responses: list[bytes | Exception]) -> None: self._responses = responses - async def write_command(self, data: bytes) -> None: + async def write_command(self, data: bytes, response: bool = True) -> None: pass async def read_response(self, timeout: float) -> bytes: