feat: host-side helpers so callers stop reimplementing the protocol

src/opendisplay/__init__.py

@@ -6,6 +6,7 @@
 from epaper_dithering import ColorScheme, DitherMode
 
 from .battery import voltage_to_percent
+from .crypto import KEY_LENGTH_BYTES, KEY_LENGTH_HEX, parse_encryption_key
 from .device import OpenDisplayDevice, prepare_image
 from .discovery import discover_devices, discover_devices_with_adv
 from .discovery_ip import IpDeviceInfo, discover_ip_devices
@@ -18,6 +19,7 @@
     ConfigParseError,
     ImageEncodingError,
     IntegrityCheckError,
+    InvalidEncryptionKeyError,
     InvalidResponseError,
     NfcNotSupportedError,
     NfcWriteError,
@@ -88,12 +90,32 @@
     get_board_type_name,
     get_manufacturer_name,
 )
-from .models.firmware import firmware_ota_asset, firmware_release_repo
-from .models.led_flash import LedFlashConfig, LedFlashStep
+from .models.firmware import (
+    firmware_ota_asset,
+    firmware_release_repo,
+    format_firmware_version,
+    supports_ble_ota_install,
+)
+from .models.led_flash import (
+    DELAY_UNIT_MS,
+    LedFlashConfig,
+    LedFlashStep,
+    ms_to_inter_delay_units,
+    ms_to_loop_delay_units,
+    pack_led_color,
+    unpack_led_color,
+)
 from .ota import find_nrf_dfu_device, perform_nrf_dfu, perform_silabs_ota
 from .partial import PartialState
-from .protocol import MANUFACTURER_ID, SERVICE_UUID
+from .protocol import (
+    MANUFACTURER_ID,
+    NFC_MIME_TYPE_MAX,
+    NFC_WRITE_MAX_TOTAL,
+    SERVICE_UUID,
+    build_nfc_payload,
+)
 from .sensors import SensorReading, read_sensor_values
+from .sleep import DEFAULT_WAKE_WINDOW_MS, SleepModel
 from .transport import BleTransport, TcpTransport, Transport
 
 __version__ = "0.1.0"
@@ -127,6 +149,7 @@
     "InvalidResponseError",
     "ImageEncodingError",
     "IntegrityCheckError",
+    "InvalidEncryptionKeyError",
     "NfcNotSupportedError",
     "NfcWriteError",
     "OTAError",
@@ -147,7 +170,14 @@
     "note_to_index",
     "LedFlashConfig",
     "LedFlashStep",
+    "DELAY_UNIT_MS",
+    "pack_led_color",
+    "unpack_led_color",
+    "ms_to_loop_delay_units",
+    "ms_to_inter_delay_units",
     "firmware_ota_asset",
+    "format_firmware_version",
+    "supports_ble_ota_install",
     "firmware_release_repo",
     "SensorData",
     "SensorReading",
@@ -207,4 +237,12 @@
     # Constants
     "SERVICE_UUID",
     "MANUFACTURER_ID",
+    "NFC_MIME_TYPE_MAX",
+    "NFC_WRITE_MAX_TOTAL",
+    "build_nfc_payload",
+    "SleepModel",
+    "DEFAULT_WAKE_WINDOW_MS",
+    "KEY_LENGTH_BYTES",
+    "KEY_LENGTH_HEX",
+    "parse_encryption_key",
 ]

src/opendisplay/cli.py

@@ -23,13 +23,15 @@
 from rich.tree import Tree
 
 from .battery import voltage_to_percent
+from .crypto import parse_encryption_key
 from .device import OpenDisplayDevice
 from .discovery import discover_devices_with_adv
 from .exceptions import (
     AuthenticationFailedError,
     AuthenticationRequiredError,
     BLEConnectionError,
     BLETimeoutError,
+    InvalidEncryptionKeyError,
     OpenDisplayError,
 )
 from .models.config import GlobalConfig, SensorData
@@ -52,7 +54,7 @@
     TouchIcType,
     WifiEncryption,
 )
-from .models.firmware import FirmwareVersion
+from .models.firmware import FirmwareVersion, format_firmware_version
 from .partial import PartialState
 from .sensors import SensorReading
 
@@ -99,16 +101,15 @@ def _handle_ble_error(exc: OpenDisplayError) -> NoReturn:
 
 
 def _parse_hex_key(hex_str: str | None) -> bytes | None:
-    """Convert hex string to 16-byte AES key, or None if not provided."""
-    if hex_str is None:
-        return None
-    cleaned = hex_str.strip().replace(" ", "").replace(":", "")
-    if len(cleaned) != 32:
-        _error(f"--key must be exactly 32 hex characters (16 bytes), got {len(cleaned)}")
+    """Convert hex string to 16-byte AES key, or None if not provided.
+
+    Delegates the format itself to the library so the CLI cannot drift from what
+    every other host accepts; only the argparse-flavoured error text is local.
+    """
     try:
-        return bytes.fromhex(cleaned)
-    except ValueError as exc:
-        _error(f"--key contains invalid hex characters: {exc}")
+        return parse_encryption_key(hex_str)
+    except InvalidEncryptionKeyError as exc:
+        _error(f"--key is not a valid encryption key: {exc}")
 
 
 def _parse_compression_value(flag: str, value: str) -> float | str:
@@ -943,7 +944,10 @@ def _build_info_tree(ctx: _InfoContext) -> Tree:
             group.add(list_section.line(item))
 
     fw = ctx.fw
-    version = f"{fw['major']}.{fw['minor']}.{fw.get('patch', 0)}"
+    # Formatted through the library so the CLI, a host's device registry and a
+    # release tag all render the same version. Before this, the CLI always
+    # printed three parts and showed 1.6.0 where the tag is 1.6.
+    version = format_firmware_version(fw["major"], fw["minor"], fw.get("patch"))
     tree.add(f"[bold]Firmware[/bold]          {version}  [dim](sha: {fw['sha']})[/dim]")
     return tree
 

src/opendisplay/crypto.py

@@ -7,17 +7,67 @@
 from __future__ import annotations
 
 import os
+from typing import Final
 
 from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes
 from cryptography.hazmat.primitives.ciphers.aead import AESCCM
 from cryptography.hazmat.primitives.cmac import CMAC
 
+from .exceptions import InvalidEncryptionKeyError
+
 # Firmware placeholder device ID (hardcoded in firmware, never changes)
 _DEVICE_ID = bytes([0x00, 0x00, 0x00, 0x01])
 
 # CCM auth tag length used by firmware
 _TAG_LEN = 12
 
+#: Length of the AES-128 master key, in bytes.
+KEY_LENGTH_BYTES: Final = 16
+#: Length of the same key written as hex, which is how hosts store it.
+KEY_LENGTH_HEX: Final = KEY_LENGTH_BYTES * 2
+
+
+def parse_encryption_key(raw: bytes | str | None) -> bytes | None:
+    """Normalize an encryption key into the bytes the device API expects.
+
+    ``raw`` is the form a host persists or a user pastes: 32 hex characters for
+    the 16-byte AES-128 master key, the 16 raw bytes themselves, or None when the
+    device is unencrypted. Bytes are validated and returned unchanged, so a
+    caller holding either form can pass it straight through without branching.
+
+    Normalization is deliberately forgiving, because a key is something a human
+    copies between a config tool, a shell and a settings field. Case is not
+    significant, and spaces and colons are separators rather than content, so
+    ``AA:BB:CC...``, ``aa bb cc...`` and ``aabbcc...`` are the same key.
+
+    This is the single definition of the stored-key format. Hosts that
+    reimplement it drift from the library and from each other, and each such copy
+    is a place where a malformed key produces a different, less useful error.
+
+    Returns:
+        The 16 key bytes, or None if ``raw`` is None.
+
+    Raises:
+        InvalidEncryptionKeyError: If the key is the wrong length or not hex.
+            An empty string raises rather than being read as "no key" - absence
+            is expressed by None, so an empty value is a malformed key.
+    """
+    if raw is None:
+        return None
+    if isinstance(raw, bytes):
+        if len(raw) != KEY_LENGTH_BYTES:
+            raise InvalidEncryptionKeyError(f"encryption key must be {KEY_LENGTH_BYTES} bytes, got {len(raw)}")
+        return raw
+    candidate = raw.strip().replace(" ", "").replace(":", "")
+    if len(candidate) != KEY_LENGTH_HEX:
+        raise InvalidEncryptionKeyError(
+            f"encryption key must be {KEY_LENGTH_HEX} hex characters ({KEY_LENGTH_BYTES} bytes), got {len(candidate)}"
+        )
+    try:
+        return bytes.fromhex(candidate)
+    except ValueError as err:
+        raise InvalidEncryptionKeyError("encryption key is not valid hexadecimal") from err
+
 
 def aes_cmac(key: bytes, data: bytes) -> bytes:
     """Compute AES-128-CMAC."""

src/opendisplay/device.py

@@ -25,6 +25,7 @@
     derive_session_key,
     encrypt_command,
     generate_client_nonce,
+    parse_encryption_key,
 )
 from .display_palettes import PANELS_4GRAY, get_bwry_codes, get_gray4_codes, get_palette_for_display
 from .encoding import (
@@ -45,6 +46,7 @@
     BLETimeoutError,
     ImageEncodingError,
     IntegrityCheckError,
+    InvalidEncryptionKeyError,
     InvalidResponseError,
     NfcNotSupportedError,
     ProtocolError,
@@ -100,6 +102,7 @@
     build_direct_write_start_uncompressed,
     build_enter_dfu_command,
     build_led_activate_command,
+    build_nfc_payload,
     build_nfc_write_data_command,
     build_nfc_write_end_command,
     build_nfc_write_inline_command,
@@ -280,7 +283,7 @@ def prepare_image(
     use_measured_palettes: bool = True,
     panel_ic_type: int | None = None,
     dither_mode: DitherMode = DitherMode.BURKES,
-    compress: bool = True,
+    compress: bool | None = True,
     serpentine: bool = True,
     exposure: float = 1.0,
     saturation: float = 1.0,
@@ -306,7 +309,10 @@ def prepare_image(
         panel_ic_type: Panel IC type for palette lookup. If None, extracted
             from config.
         dither_mode: Dithering algorithm to use (default: BURKES)
-        compress: Whether to compress the image data (default: True)
+        compress: Whether to compress the image data (default: True). Pass None
+            to derive it from ``config``, matching what upload_image() does:
+            compress only when the panel advertises ZIP or streaming
+            decompression.
         serpentine: Alternate scan direction each row to reduce artifacts (default: True)
         exposure: Exposure multiplier, >1.0 brightens (default: 1.0)
         saturation: Saturation multiplier, >1.0 boosts (default: 1.0)
@@ -385,7 +391,14 @@ def prepare_image(
     else:
         image_data = encode_image(dithered, color_scheme)
 
-    # Optionally compress
+    # Optionally compress. compress=None means "ask the config", which is what
+    # upload_image() does internally; a caller preparing an image up front (to
+    # move the CPU work off an event loop, say) would otherwise have to
+    # reimplement this check and keep it in sync by hand.
+    if compress is None:
+        display_cfg = config.displays[0] if config is not None and config.displays else None
+        compress = display_cfg.supports_compression if display_cfg else True
+
     compressed_data = None
     if compress:
         # Current firmware compiles uzlib with a 9-bit window and hard-rejects any
@@ -504,7 +517,7 @@ def __init__(
         max_attempts: int = 4,
         use_services_cache: bool = True,
         use_measured_palettes: bool = True,
-        encryption_key: bytes | None = None,
+        encryption_key: bytes | str | None = None,
         blocks_per_ack: int = 8,
         max_queue_size: int = 16,
     ):
@@ -532,7 +545,11 @@ def __init__(
             max_attempts: Maximum connection attempts for bleak-retry-connector (default: 4)
             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).
+            encryption_key: AES-128 master key for encrypted devices (optional),
+                as the 16 raw bytes or as 32 hex characters. A malformed key is
+                rejected here rather than at the first command, so a bad key
+                cannot cost a connection (or a sleeping device's wake window)
+                before it is reported.
             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
@@ -580,7 +597,10 @@ def __init__(
         self._fw_version: FirmwareVersion | None = None
 
         # Encryption session state (populated by authenticate())
-        self._encryption_key = encryption_key
+        # Normalize (and validate) up front: a str key used to be accepted
+        # silently and fail with an opaque TypeError inside authenticate(),
+        # after the connect had already happened.
+        self._encryption_key = parse_encryption_key(encryption_key)
         self._session_key: bytes | None = None
         self._session_id: bytes | None = None
         self._nonce_counter: int = 0
@@ -850,19 +870,27 @@ async def _read(self, timeout: float) -> bytes:
         return raw
 
     @_serialized
-    async def authenticate(self, key: bytes) -> None:
+    async def authenticate(self, key: bytes | str) -> None:
         """Perform two-step challenge-response authentication with the device.
 
         After successful authentication, all subsequent commands and responses
         are transparently encrypted/decrypted via _write() and _read().
 
         Args:
-            key: 16-byte AES-128 master key
+            key: AES-128 master key, as the 16 raw bytes or 32 hex characters
 
         Raises:
             AuthenticationFailedError: If the device rejects the key or is rate-limited
+            InvalidEncryptionKeyError: If the key is malformed. Raised before any
+                frame is sent, so a bad key is never mistaken for a device that
+                refused a good one.
             InvalidResponseError: If device sends malformed response
         """
+        parsed_key = parse_encryption_key(key)
+        if parsed_key is None:  # pragma: no cover - excluded by the signature
+            raise InvalidEncryptionKeyError("authenticate() requires an encryption key")
+        key = parsed_key
+
         _LOGGER.debug("Authenticating with device %s", self.mac_address)
 
         # Step 1: Request server nonce (retry once if device reports existing session)
@@ -1548,7 +1576,7 @@ async def write_nfc_url(self, url: str, timeout: float | None = None) -> None:
             url: URL to write.
             timeout: Optional override; see write_nfc.
         """
-        await self.write_nfc(NfcRecordType.URI, url.encode("utf-8"), timeout)
+        await self.write_nfc(NfcRecordType.URI, build_nfc_payload(NfcRecordType.URI, url), timeout)
 
     async def write_nfc_text(self, text: str, timeout: float | None = None) -> None:
         """Write a TEXT NDEF record containing the given text.
@@ -1559,7 +1587,7 @@ async def write_nfc_text(self, text: str, timeout: float | None = None) -> None:
             text: Text to write.
             timeout: Optional override; see write_nfc.
         """
-        await self.write_nfc(NfcRecordType.TEXT, text.encode("utf-8"), timeout)
+        await self.write_nfc(NfcRecordType.TEXT, build_nfc_payload(NfcRecordType.TEXT, text), timeout)
 
     async def write_nfc_mime(
         self,
@@ -1580,13 +1608,10 @@ async def write_nfc_mime(
             timeout: Optional override; see write_nfc.
 
         Raises:
-            ValueError: If the encoded MIME type is outside 1..255 bytes.
+            ValueError: If the encoded MIME type is outside 1..255 bytes, or the
+                assembled payload exceeds NFC_WRITE_MAX_TOTAL.
         """
-        mt = mime_type.encode("utf-8")
-        if not 1 <= len(mt) <= 255:
-            raise ValueError(f"mime_type must encode to 1..255 bytes, got {len(mt)}")
-        body_bytes = body.encode("utf-8") if isinstance(body, str) else body
-        payload = bytes([len(mt)]) + mt + body_bytes
+        payload = build_nfc_payload(NfcRecordType.MIME, body, mime_type)
         await self.write_nfc(NfcRecordType.MIME, payload, timeout)
 
     @_serialized
@@ -1799,9 +1824,7 @@ async def upload_image(
         # compressed uploads either way (<=1.81 NACKs without the ZIP bit and
         # the upload falls back to uncompressed).
         display_cfg = self._config.displays[0] if (self._config and self._config.displays) else None
-        supports_compression = (
-            (display_cfg.supports_zip or display_cfg.supports_streaming_decompression) if display_cfg else True
-        )
+        supports_compression = display_cfg.supports_compression if display_cfg else True
 
         # When a partial upload may succeed, defer full-frame compression: it is
         # pure waste if the partial path handles the update. _dispatch_upload
@@ -1919,9 +1942,7 @@ async def _dispatch_upload(
         sent), False if the firmware auto-completed the upload.
         """
         display_cfg = self._config.displays[0] if (self._config and self._config.displays) else None
-        supports_compression = (
-            (display_cfg.supports_zip or display_cfg.supports_streaming_decompression) if display_cfg else True
-        )
+        supports_compression = display_cfg.supports_compression if display_cfg else True
         streaming_decompression = bool(display_cfg and display_cfg.supports_streaming_decompression)
         if (
             compress