diff --git a/examples/animate.py b/examples/animate.py new file mode 100644 index 0000000..b2b5f5a --- /dev/null +++ b/examples/animate.py @@ -0,0 +1,286 @@ +"""Animate a sequence of images on an OpenDisplay device. + +Connects once, sends the first frame as a full update, then loops through +subsequent frames using partial (delta) updates. Repeats indefinitely. + +Usage: + python examples/animate.py --device AA:BB:CC:DD:EE:FF --interval 500 frame1.png frame2.png ... + python examples/animate.py --device AA:BB:CC:DD:EE:FF --interval 1000 "frames/*.png" +""" + +from __future__ import annotations + +import argparse +import asyncio +import glob +import logging +import sys +import time +from collections.abc import Coroutine +from pathlib import Path +from typing import Any, NoReturn, TypeVar + +from epaper_dithering import DitherMode +from PIL import Image, UnidentifiedImageError +from rich.console import Console +from rich.live import Live +from rich.logging import RichHandler +from rich.progress import BarColumn, Progress, SpinnerColumn, TaskProgressColumn, TextColumn + +from opendisplay.device import OpenDisplayDevice, prepare_image +from opendisplay.exceptions import ( + AuthenticationFailedError, + AuthenticationRequiredError, + BLEConnectionError, + BLETimeoutError, + OpenDisplayError, +) +from opendisplay.models.enums import RefreshMode +from opendisplay.partial import PartialState + +_T = TypeVar("_T") + +_console = Console(stderr=True) + +_DITHER_CHOICES: dict[str, DitherMode] = {m.name.lower().replace("_", "-"): m for m in DitherMode} + + +def _run(coro: Coroutine[Any, Any, _T]) -> _T: + return asyncio.run(coro) + + +def _error(msg: str) -> NoReturn: + _console.print(f"[bold red]Error:[/bold red] {msg}") + sys.exit(1) + + +def _handle_ble_error(exc: OpenDisplayError) -> NoReturn: + if isinstance(exc, AuthenticationRequiredError): + _error("Device requires an encryption key. Pass --key HEX.") + if isinstance(exc, AuthenticationFailedError): + _error("Authentication failed. Check that --key is correct.") + if isinstance(exc, BLETimeoutError): + _error(f"BLE timeout: {exc}") + if isinstance(exc, BLEConnectionError): + _error(f"BLE connection failed: {exc}") + _error(f"Device error: {exc}") + + +def _parse_hex_key(hex_str: str | None) -> bytes | None: + 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)}") + try: + return bytes.fromhex(cleaned) + except ValueError as exc: + _error(f"--key contains invalid hex characters: {exc}") + + +def _device_kwargs(device: str, key: bytes | None, timeout: float) -> dict[str, Any]: + kwargs: dict[str, Any] = {"timeout": timeout, "encryption_key": key} + if ":" in device or (len(device) == 36 and device.count("-") == 4): + kwargs["mac_address"] = device + else: + kwargs["device_name"] = device + return kwargs + + +def _setup_logging(verbose: bool) -> None: + level = logging.DEBUG if verbose else logging.WARNING + logging.basicConfig( + level=level, + format="%(message)s", + handlers=[RichHandler(console=_console, rich_tracebacks=True)], + force=True, + ) + logging.getLogger("bleak").setLevel(logging.INFO) + logging.getLogger("PIL").setLevel(logging.INFO) + + +def _expand_paths(patterns: list[str]) -> list[str]: + expanded: list[str] = [] + for p in patterns: + matches = sorted(glob.glob(p)) + expanded.extend(matches if matches else [p]) + return expanded + + +def _load_images(paths: list[str]) -> list[Image.Image]: + images: list[Image.Image] = [] + for p in paths: + try: + img = Image.open(p) + img.load() + images.append(img) + except FileNotFoundError: + _error(f"Image file not found: {p}") + except UnidentifiedImageError: + _error(f"Cannot open image (unsupported format): {p}") + return images + + +async def _animate( + device_kwargs: dict[str, Any], + images: list[Image.Image], + names: list[str], + interval_ms: int, + dither_mode: DitherMode, +) -> None: + total = len(images) + delay = interval_ms / 1000.0 + + spinner_progress = Progress( + SpinnerColumn(finished_text="[green]✓[/green]"), + TextColumn("{task.description}"), + console=_console, + ) + bar_progress = Progress( + BarColumn(), + TaskProgressColumn(), + console=_console, + ) + + class _Display: + def __rich_console__(self, _con, _opts): # type: ignore[no-untyped-def] + yield spinner_progress + if any(t.visible for t in bar_progress.tasks): + yield bar_progress + + try: + with Live(_Display(), console=_console, refresh_per_second=10, transient=False): + status_task = spinner_progress.add_task("Connecting...", total=None) + bar_task = bar_progress.add_task("", total=None, visible=False) + + async with OpenDisplayDevice(**device_kwargs) as device: + spinner_progress.update(status_task, description=f"Pre-processing {total} frame(s)...") + prep_times: list[float] = [] + prepared = [] + for img in images: + t0 = time.perf_counter() + prepared.append( + prepare_image( + img, config=device.config, capabilities=device.capabilities, dither_mode=dither_mode + ) + ) + prep_times.append(time.perf_counter() - t0) + + state = PartialState() + frame_count = 0 + + try: + while True: + idx = frame_count % total + is_first = frame_count == 0 + refresh_mode = RefreshMode.FULL if is_first else RefreshMode.PARTIAL + update_type = "full" if is_first else "partial" + name = names[idx] + + def _status(phase: str, _idx: int = idx, _name: str = name, _ut: str = update_type) -> str: + return f"{phase} {_idx + 1}/{total} [dim]{_name}[/dim] ({_ut})" + + send_end: list[float] = [] + bytes_transferred: list[int] = [] + + def on_progress(sent: int, total_bytes: int) -> None: + bar_progress.update(bar_task, total=total_bytes, completed=sent, visible=True) + if sent == total_bytes: + send_end.append(time.perf_counter()) + bytes_transferred.append(total_bytes) + bar_progress.update(bar_task, visible=False) + spinner_progress.update(status_task, description=_status("Refreshing...")) + + t_upload_start = time.perf_counter() + spinner_progress.update(status_task, description=_status("Sending")) + await device.upload_prepared_image( + prepared[idx], + refresh_mode=refresh_mode, + state=state, + progress_callback=on_progress, + ) + t_upload_end = time.perf_counter() + + prep_ms = prep_times[idx] * 1000 + send_ms = (send_end[0] - t_upload_start) * 1000 + refresh_ms = (t_upload_end - send_end[0]) * 1000 + kb = bytes_transferred[0] / 1024 + stats = ( + f"prep {prep_ms:.0f}ms · send {send_ms:.0f}ms ({kb:.1f} KB) · refresh {refresh_ms:.0f}ms" + ) + + _console.print(f"[dim]{idx + 1}/{total} {name} ({update_type}) {stats}[/dim]") + spinner_progress.update(status_task, description=_status("Showing")) + frame_count += 1 + await asyncio.sleep(delay) + + except (KeyboardInterrupt, asyncio.CancelledError): + spinner_progress.update(status_task, description="Stopped.", total=1, completed=1) + return + + except OpenDisplayError as exc: + _handle_ble_error(exc) + + +def _cmd_animate(args: argparse.Namespace) -> None: + key = _parse_hex_key(args.key) + paths = _expand_paths(args.images) + + if len(paths) < 2: + _error(f"At least 2 images are required for animation, got {len(paths)}.") + + images = _load_images(paths) # validate and read all files before connecting + + names = [Path(p).name for p in paths] + listing = ", ".join(names) if len(names) <= 4 else f"{names[0]}, {names[1]}, ..., {names[-1]}" + _console.print(f"Found {len(paths)} image(s): {listing}") + + _run( + _animate( + _device_kwargs(args.device, key, args.timeout), + images, + names, + args.interval, + _DITHER_CHOICES[args.dither_mode], + ) + ) + + +def main() -> None: + parser = argparse.ArgumentParser( + prog="animate", + description="Animate a sequence of images on an OpenDisplay e-ink device.", + epilog='Example: python examples/animate.py --device AA:BB:CC:DD:EE:FF --interval 500 "frames/*.png"', + ) + parser.add_argument("-v", "--verbose", action="store_true", help="Enable debug logging") + parser.add_argument("--device", required=True, metavar="ADDR", help="Device MAC address or name") + parser.add_argument( + "--interval", + type=int, + default=1000, + metavar="MS", + help="Delay between frames in milliseconds (default: 1000)", + ) + parser.add_argument( + "--dither-mode", + choices=list(_DITHER_CHOICES), + default="burkes", + help="Dithering algorithm (default: burkes)", + ) + parser.add_argument("--key", default=None, metavar="HEX", help="Encryption key as 32 hex characters") + parser.add_argument( + "--timeout", + type=float, + default=10.0, + metavar="SECS", + help="BLE timeout in seconds (default: 10.0)", + ) + parser.add_argument("images", nargs="+", metavar="IMAGE", help="Image files to animate (in order)") + + args = parser.parse_args() + _setup_logging(args.verbose) + _cmd_animate(args) + + +if __name__ == "__main__": + main() diff --git a/src/opendisplay/cli.py b/src/opendisplay/cli.py index c0f2ef8..f66b837 100644 --- a/src/opendisplay/cli.py +++ b/src/opendisplay/cli.py @@ -6,8 +6,10 @@ import asyncio import json import logging +import os import sys from collections.abc import Coroutine +from pathlib import Path from typing import Any, NoReturn, TypeVar from epaper_dithering import DitherMode @@ -40,6 +42,7 @@ SensorType, WifiEncryption, ) +from .partial import PartialState _T = TypeVar("_T") @@ -472,6 +475,25 @@ async def _info(device_kwargs: dict[str, Any], output_json: bool) -> None: # ── upload ──────────────────────────────────────────────────────────────────── +def _load_partial_state(path: str) -> PartialState: + """Load PartialState from path, or return a fresh one if the file is absent.""" + p = Path(path) + if not p.exists(): + return PartialState() + try: + return PartialState.from_bytes(p.read_bytes()) + except (OSError, ValueError) as exc: + _error(f"Failed to load --state-file {path}: {exc}") + + +def _save_partial_state(path: str, state: PartialState) -> None: + """Atomically write PartialState to path (write to .tmp then rename).""" + p = Path(path) + tmp = p.with_suffix(p.suffix + ".tmp") + tmp.write_bytes(state.to_bytes()) + os.replace(tmp, p) + + def _add_upload_parser(subparsers: argparse._SubParsersAction[argparse.ArgumentParser]) -> None: p = subparsers.add_parser("upload", help="Upload an image to the device") _add_device_options(p) @@ -522,6 +544,14 @@ def _add_upload_parser(subparsers: argparse._SubParsersAction[argparse.ArgumentP metavar="VALUE", help='Gamut compression: "auto", "off", or 0.0–1.0 (default: 0)', ) + p.add_argument( + "--state-file", + metavar="PATH", + default=None, + help="Persistent partial-rendering state file. If the file exists, it's loaded and " + "the upload attempts a partial transfer; on success the file is rewritten. If it " + "does not exist, a fresh state is created (forcing a full upload first time).", + ) p.set_defaults(func=_cmd_upload) @@ -545,6 +575,7 @@ def _cmd_upload(args: argparse.Namespace) -> None: args.highlights, tone, gamut, + args.state_file, ) ) @@ -564,6 +595,7 @@ async def _upload( highlights: float, tone: float | str, gamut: float | str, + state_file: str | None, ) -> None: try: image = Image.open(image_path) @@ -611,6 +643,8 @@ def on_progress(sent: int, total: int) -> None: spinner_progress.update(upload_task, visible=False) spinner_progress.update(refresh_task, visible=True) + state = _load_partial_state(state_file) if state_file else None + await device.upload_image( image, refresh_mode=refresh_mode, @@ -626,8 +660,12 @@ def on_progress(sent: int, total: int) -> None: fit=fit, rotate=rotate, progress_callback=on_progress, + state=state, ) + if state_file and state is not None: + _save_partial_state(state_file, state) + spinner_progress.update(refresh_task, visible=False) spinner_progress.update(upload_task, visible=True, description="[green]Done.[/green]", total=1, completed=1) except OpenDisplayError as exc: diff --git a/src/opendisplay/device.py b/src/opendisplay/device.py index f4db27a..5dca12c 100644 --- a/src/opendisplay/device.py +++ b/src/opendisplay/device.py @@ -4,6 +4,7 @@ import logging import time +import zlib from collections.abc import Callable from typing import TYPE_CHECKING @@ -28,6 +29,7 @@ from .exceptions import ( AuthenticationRequiredError, AuthenticationSessionExistsError, + BLETimeoutError, ImageEncodingError, InvalidResponseError, ProtocolError, @@ -37,6 +39,16 @@ from .models.enums import BoardManufacturer, FitMode, RefreshMode, Rotation from .models.firmware import FirmwareVersion from .models.led_flash import LedFlashConfig +from .partial import ( + ERR_ETAG_MISMATCH, + PARTIAL_FLAG_COMPRESSED, + PartialState, + _generate_etag, + build_partial_logical_stream, + compute_partial_region, + encode_segment_wire, + parse_nack, +) from .protocol import ( CHUNK_SIZE, ENCRYPTED_CHUNK_SIZE, @@ -48,6 +60,8 @@ build_authenticate_step2, build_direct_write_data_command, build_direct_write_end_command, + build_direct_write_end_with_etag, + build_direct_write_partial_start, build_direct_write_start_compressed, build_direct_write_start_uncompressed, build_led_activate_command, @@ -913,6 +927,7 @@ async def upload_image( fit: FitMode = FitMode.CONTAIN, rotate: Rotation = Rotation.ROTATE_0, progress_callback: Callable[[int, int], None] | None = None, + state: PartialState | None = None, ) -> Image.Image: """Upload image to device display. @@ -975,9 +990,32 @@ async def upload_image( fit=fit, rotate=rotate, ) - await self._dispatch_upload(image_data, refresh_mode, compress, compressed_data, progress_callback) + + if state is not None: + partial_outcome = await self._maybe_upload_partial(processed_image, state, progress_callback) + if partial_outcome == "success": + _LOGGER.info("Image upload complete (partial path)") + return processed_image + if partial_outcome == "no_change": + _LOGGER.info("No pixels changed; skipping upload") + return processed_image + if partial_outcome == "fallback_full": + _LOGGER.info("Partial path unavailable or etag mismatch; continuing with full upload") + + upload_refresh_mode = RefreshMode.FULL if state is not None else refresh_mode + full_upload_etag = _generate_etag() if state is not None else None + await self._dispatch_upload( + image_data, + upload_refresh_mode, + compress, + compressed_data, + progress_callback, + new_etag=full_upload_etag, + ) _LOGGER.info("Image upload complete") + if state is not None: + self._update_partial_state(state, processed_image, image_data, full_upload_etag) return processed_image async def upload_prepared_image( @@ -986,6 +1024,7 @@ async def upload_prepared_image( refresh_mode: RefreshMode = RefreshMode.FULL, compress: bool = True, progress_callback: Callable[[int, int], None] | None = None, + state: PartialState | None = None, ) -> None: """Upload pre-computed image data to device. @@ -1004,9 +1043,32 @@ async def upload_prepared_image( Raises: ProtocolError: If upload fails """ - image_data, compressed_data, _ = prepared_data - await self._dispatch_upload(image_data, refresh_mode, compress, compressed_data, progress_callback) + image_data, compressed_data, processed_image = prepared_data + + if state is not None: + partial_outcome = await self._maybe_upload_partial(processed_image, state, progress_callback) + if partial_outcome == "success": + _LOGGER.info("Prepared image upload complete (partial path)") + return + if partial_outcome == "no_change": + _LOGGER.info("No pixels changed; skipping prepared upload") + return + if partial_outcome == "fallback_full": + _LOGGER.info("Partial prepared upload unavailable or etag mismatch; continuing with full upload") + + upload_refresh_mode = RefreshMode.FULL if state is not None else refresh_mode + full_upload_etag = _generate_etag() if state is not None else None + await self._dispatch_upload( + image_data, + upload_refresh_mode, + compress, + compressed_data, + progress_callback, + new_etag=full_upload_etag, + ) _LOGGER.info("Prepared image upload complete") + if state is not None: + self._update_partial_state(state, processed_image, image_data, full_upload_etag) async def _dispatch_upload( self, @@ -1015,6 +1077,7 @@ async def _dispatch_upload( compress: bool, compressed_data: bytes | None, progress_callback: Callable[[int, int], None] | None, + new_etag: int | None = None, ) -> None: """Choose compressed or uncompressed upload protocol and execute it.""" display_cfg = self._config.displays[0] if (self._config and self._config.displays) else None @@ -1031,6 +1094,7 @@ async def _dispatch_upload( compressed_data=compressed_data, uncompressed_size=len(image_data), progress_callback=progress_callback, + new_etag=new_etag, ) else: if compress and not supports_compression: @@ -1040,9 +1104,179 @@ async def _dispatch_upload( else: _LOGGER.info("Compression disabled or no compressed data, using uncompressed protocol") await self._execute_upload( - image_data, refresh_mode, use_compression=False, progress_callback=progress_callback + image_data, + refresh_mode, + use_compression=False, + progress_callback=progress_callback, + new_etag=new_etag, ) + def _update_partial_state( + self, + state: PartialState, + processed_image: Image.Image, + image_data: bytes, + etag: int | None = None, + ) -> None: + """After a successful full upload, refresh state to reflect what's now on the panel. + + Stores the etag committed to the device on 0x72 (or generates one if + absent), then stashes the palette pixels for diffing on the next call. + ``image_data`` is unused but kept for API symmetry. + """ + del image_data + palette_image = processed_image.convert("P") if processed_image.mode != "P" else processed_image + state.etag = _generate_etag() if etag is None else etag + state.last_image = palette_image.tobytes() + state.width, state.height = processed_image.size + state.bytes_per_pixel = 1 + + async def _send_partial_chunks( + self, + remaining: bytes, + stream_bytes: bytes, + state: PartialState, + progress_callback: Callable[[int, int], None] | None = None, + ) -> None: + """Send remaining 0x71 chunks and update upload progress.""" + chunk_size = ENCRYPTED_CHUNK_SIZE if self._session_key is not None else CHUNK_SIZE + total_stream_bytes = len(stream_bytes) + bytes_sent = total_stream_bytes - len(remaining) + offset = 0 + while offset < len(remaining): + chunk = remaining[offset : offset + chunk_size] + await self._write(build_direct_write_data_command(chunk)) + ack = await self._read(self.TIMEOUT_ACK) + nack = parse_nack(ack) + if nack is not None: + opcode, err = nack + state.etag = 0 + state.last_image = None + raise ProtocolError(f"Partial 0x71 NACK: opcode=0x{opcode:02x} err=0x{err:02x}") + validate_ack_response(ack, CommandCode.DIRECT_WRITE_DATA) + offset += len(chunk) + bytes_sent += len(chunk) + if progress_callback is not None: + progress_callback(bytes_sent, total_stream_bytes) + + async def _maybe_upload_partial( + self, + processed_image: Image.Image, + state: PartialState, + progress_callback: Callable[[int, int], None] | None = None, + ) -> str: + """Try a partial upload using the 0x76 single-rectangle protocol.""" + # Resolve all partial-update preconditions in one pass (support checks, + # state validation, diff computation, and region alignment). + region = compute_partial_region(processed_image, state, self._config, self.color_scheme) + if isinstance(region, str): + return region + + display = region.display + _LOGGER.debug( + "Partial path diff: old_etag=0x%08x, image=%dx%d, rect=(%d,%d,%d,%d)", + state.etag, + region.width, + region.height, + region.rx, + region.ry, + region.rw, + region.rh, + ) + + # Build logical stream from the changed rectangle only, then compress + # when it reduces the transfer size. + old_palette_image = region.palette_image.copy() + old_palette_image.frombytes(region.old_palette) + old_rect_bytes = encode_segment_wire( + old_palette_image, + region.rx, + region.ry, + region.rw, + region.rh, + region.color_scheme, + ) + new_rect_bytes = encode_segment_wire( + region.palette_image, + region.rx, + region.ry, + region.rw, + region.rh, + region.color_scheme, + ) + + logical_stream = build_partial_logical_stream(old_rect_bytes, new_rect_bytes) + compressed_stream = zlib.compress(logical_stream, level=6) + use_compression = display.supports_zip and len(compressed_stream) < len(logical_stream) + stream_bytes = compressed_stream if use_compression else logical_stream + + flags = 0 + if use_compression: + flags |= PARTIAL_FLAG_COMPRESSED + + _LOGGER.debug( + "Partial stream: rect=(%d,%d,%d,%d), uncompressed=%d, wire=%d, compressed=%s", + region.rx, + region.ry, + region.rw, + region.rh, + len(logical_stream), + len(stream_bytes), + use_compression, + ) + + new_etag = _generate_etag() + _LOGGER.debug("Partial upload: old_etag=0x%08x new_etag=0x%08x", state.etag, new_etag) + + # Start partial upload (0x76), stream remaining 0x71 chunks, and finish + # with partial refresh. + start_pkt, remaining = build_direct_write_partial_start( + old_etag=state.etag, + new_etag=new_etag, + flags=flags, + x=region.rx, + y=region.ry, + width=region.rw, + height=region.rh, + stream_bytes=stream_bytes, + ) + await self._write(start_pkt) + try: + response = await self._read(self.TIMEOUT_ACK) + nack = parse_nack(response) + if nack is not None: + opcode, err = nack + if opcode == 0x76 and err == ERR_ETAG_MISMATCH: + _LOGGER.info("Partial upload: etag mismatch; falling back to full upload") + state.etag = 0 + state.last_image = None + return "fallback_full" + raise ProtocolError(f"Partial 0x76 NACK: opcode=0x{opcode:02x} err=0x{err:02x}") + validate_ack_response(response, CommandCode.DIRECT_WRITE_PARTIAL_START) + except (BLETimeoutError, InvalidResponseError): + _LOGGER.info("Partial upload start was not acknowledged; falling back to full upload") + return "fallback_full" + + await self._send_partial_chunks(remaining, stream_bytes, state, progress_callback) + + await self._write(build_direct_write_end_command(RefreshMode.PARTIAL.value)) + response = await self._read(self.TIMEOUT_ACK) + validate_ack_response(response, CommandCode.DIRECT_WRITE_END) + + 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)") + if command != CommandCode.DIRECT_WRITE_REFRESH_COMPLETE: + raise ProtocolError(f"Unexpected response waiting for refresh: {command.name} (0x{command:04x})") + + 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" + async def _execute_upload( self, image_data: bytes, @@ -1051,6 +1285,7 @@ async def _execute_upload( compressed_data: bytes | None = None, uncompressed_size: int | None = None, progress_callback: Callable[[int, int], None] | None = None, + new_etag: int | None = None, ) -> None: """Execute image upload using compressed or uncompressed protocol. @@ -1110,7 +1345,11 @@ async def _execute_upload( # 4. Send END (unless device auto-triggered refresh), then wait for 0x73 if not auto_completed: - end_cmd = build_direct_write_end_command(refresh_mode.value) + end_cmd = ( + build_direct_write_end_with_etag(refresh_mode.value, new_etag) + if new_etag is not None + else build_direct_write_end_command(refresh_mode.value) + ) await self._write(end_cmd) # Compressed END triggers decompression + full SPI write to display IC, which diff --git a/src/opendisplay/models/enums.py b/src/opendisplay/models/enums.py index 6e2446e..6f1e8d0 100644 --- a/src/opendisplay/models/enums.py +++ b/src/opendisplay/models/enums.py @@ -9,11 +9,14 @@ class RefreshMode(IntEnum): """Display refresh modes. - Only FULL and FAST are supported by the firmware. + FULL is the normal full-screen update. + FAST is a panel-specific reduced-flash refresh. + PARTIAL requests the panel's true partial-update mode when supported. """ FULL = 0 FAST = 1 + PARTIAL = 2 class ICType(IntEnum): diff --git a/src/opendisplay/partial.py b/src/opendisplay/partial.py new file mode 100644 index 0000000..09ae7ae --- /dev/null +++ b/src/opendisplay/partial.py @@ -0,0 +1,321 @@ +"""Partial rendering support for OpenDisplay BLE devices. + +Provides PartialState (caller-owned mutable holder), bounding-rect helpers, +and the logical stream builder for the 0x76 single-rectangle protocol. + +Serialization format: struct header (magic + version + scalar fields) followed +by a 4-byte big-endian length-prefixed ``last_image`` blob. + +``last_image`` stores raw palette bytes (1 byte per pixel, from +``PIL.Image.tobytes()`` on the dithered palette image). +""" + +from __future__ import annotations + +import os +import struct +from dataclasses import dataclass + +from epaper_dithering import ColorScheme +from PIL import Image + +from .encoding import encode_image +from .models.config import DisplayConfig, GlobalConfig + +# --------------------------------------------------------------------------- +# Wire constants mirrored from the firmware partial-rendering protocol. +# --------------------------------------------------------------------------- + +# NACK error codes returned by the device inside {0xFF, opcode, error, 0x00} +ERR_ETAG_MISMATCH = 0x01 # on 0x76: client must fall back to full transfer +ERR_MIXED_DATA = 0x02 # aborted; etag cleared on device +ERR_RECT_OOB = 0x03 # on 0x76: rectangle out of display bounds +ERR_RECT_ALIGN = 0x04 # on 0x76: x or width not aligned to byte boundary +ERR_PARTIAL_FLAGS = 0x05 # on 0x76: unsupported or reserved flags set +ERR_PARTIAL_STREAM = 0x06 # on 0x71/0x72: stream byte count or content error +ERR_PARTIAL_UNSUPPORTED = 0x07 # on 0x76: partial update unsupported for panel mode + +NACK_PREFIX = 0xFF + +# 0x76 flag bits +PARTIAL_FLAG_COMPRESSED = 0x01 # bit 0: stream is zlib-compressed + +# pixels_per_byte for each bits_per_pixel value +_PIXELS_PER_BYTE: dict[int, int] = {1: 8, 2: 4, 4: 2, 8: 1} + + +def parse_nack(response: bytes) -> tuple[int, int] | None: + """Return (opcode, error_code) if response is a 4-byte {0xFF, op, err, 0x00} NACK. + + Returns None for any other response shape. + """ + if len(response) == 4 and response[0] == NACK_PREFIX and response[3] == 0x00: + return response[1], response[2] + return None + + +@dataclass +class PartialRegion: + """Container for validated partial-diff metadata before upload.""" + + display: DisplayConfig + color_scheme: ColorScheme + width: int + height: int + palette_image: Image.Image + new_palette: bytes + old_palette: bytes + rx: int + ry: int + rw: int + rh: int + + +def compute_partial_region( + processed_image: Image.Image, + state: PartialState, + config: GlobalConfig | None, + color_scheme: ColorScheme, +) -> str | PartialRegion: + """Build partial-region metadata for upload diffing.""" + if config is None or not getattr(config, "displays", None): + return "fallback_full" + + display = config.displays[0] + if not display.partial_update_support: + return "fallback_full" + + if color_scheme in (ColorScheme.BWR, ColorScheme.BWY): + return "fallback_full" + + width, height = processed_image.size + if state.etag == 0 or state.last_image is None or state.width != width or state.height != height: + return "fallback_full" + + palette_image = processed_image.convert("P") if processed_image.mode != "P" else processed_image + new_palette = palette_image.tobytes() + old_palette = state.last_image + if len(old_palette) != len(new_palette): + return "fallback_full" + + bbox = compute_bounding_rect(old_palette, new_palette, width, height) + if bbox is None: + return "no_change" + + bpp = { + ColorScheme.MONO: 1, + ColorScheme.BWRY: 2, + ColorScheme.GRAYSCALE_4: 2, + ColorScheme.BWGBRY: 4, + ColorScheme.GRAYSCALE_16: 4, + }.get(color_scheme, 1) + pixels_per_byte = _PIXELS_PER_BYTE.get(bpp, 8) + + rx, ry, rw, rh = align_rect(*bbox, width, height, pixels_per_byte) + if rw == 0 or rh == 0: + return "fallback_full" + + return PartialRegion( + display=display, + color_scheme=color_scheme, + width=width, + height=height, + palette_image=palette_image, + new_palette=new_palette, + old_palette=old_palette, + rx=rx, + ry=ry, + rw=rw, + rh=rh, + ) + + +# --------------------------------------------------------------------------- +# Bounding-rect helpers +# --------------------------------------------------------------------------- + + +def compute_bounding_rect( + old: bytes, + new: bytes, + width: int, + height: int, +) -> tuple[int, int, int, int] | None: + """Return (x0, y0, x1_excl, y1_excl) of changed pixels, or None if identical.""" + if old == new: + return None + min_x, max_x, min_y, max_y = width, -1, height, -1 + for y in range(height): + row_off = y * width + row_changed = False + for x in range(width): + if old[row_off + x] != new[row_off + x]: + min_x = min(min_x, x) + max_x = max(max_x, x) + row_changed = True + if row_changed: + min_y = min(min_y, y) + max_y = max(max_y, y) + if max_x < 0: + return None + return (min_x, min_y, max_x + 1, max_y + 1) + + +def align_rect( + x0: int, + y0: int, + x1: int, + y1: int, + display_width: int, + _display_height: int, + pixels_per_byte: int, +) -> tuple[int, int, int, int]: + """Expand (x0, y0, x1, y1) to packed-byte boundaries. + + Returns (x, y, width, height) ready to send in 0x76. + """ + aligned_x0 = (x0 // pixels_per_byte) * pixels_per_byte + aligned_x1 = x1 + if aligned_x1 % pixels_per_byte: + aligned_x1 += pixels_per_byte - (aligned_x1 % pixels_per_byte) + if aligned_x1 > display_width: + aligned_x1 = display_width + misalign = (aligned_x1 - aligned_x0) % pixels_per_byte + if misalign: + aligned_x0 = max(0, aligned_x0 - (pixels_per_byte - misalign)) + return (aligned_x0, y0, aligned_x1 - aligned_x0, y1 - y0) + + +def encode_segment_wire( + palette_image: Image.Image, + x: int, + y: int, + w: int, + h: int, + color_scheme: ColorScheme, +) -> bytes: + """Encode a palette rectangle to protocol wire bytes for partial updates.""" + cropped = palette_image.crop((x, y, x + w, y + h)) + pixels = cropped.tobytes() + + if color_scheme == ColorScheme.MONO: + output = bytearray((len(pixels) + 7) // 8) + for i, palette_idx in enumerate(pixels): + if palette_idx > 0: + output[i // 8] |= 1 << (7 - (i % 8)) + return bytes(output) + + if color_scheme in (ColorScheme.BWRY, ColorScheme.GRAYSCALE_4): + output = bytearray((len(pixels) + 3) // 4) + for i, palette_idx in enumerate(pixels): + shift = (3 - (i % 4)) * 2 + output[i // 4] |= (palette_idx & 0x03) << shift + return bytes(output) + + if color_scheme in (ColorScheme.BWGBRY, ColorScheme.GRAYSCALE_16): + output = bytearray((len(pixels) + 1) // 2) + bwgbry_map = {0: 0, 1: 1, 2: 2, 3: 3, 4: 5, 5: 6} + for i, palette_idx in enumerate(pixels): + value = palette_idx & 0x0F + if color_scheme == ColorScheme.BWGBRY: + value = bwgbry_map.get(value, 0) + if i % 2 == 0: + output[i // 2] |= value << 4 + else: + output[i // 2] |= value + return bytes(output) + + return encode_image(cropped, color_scheme) + + +def build_partial_logical_stream( + old_rect_bytes: bytes, + new_rect_bytes: bytes, +) -> bytes: + """Build a plane-major old-then-new partial stream. + + Produces: old_rect + new_rect. + """ + assert len(old_rect_bytes) == len(new_rect_bytes), "old/new rect byte lengths must match" + return old_rect_bytes + new_rect_bytes + + +# --------------------------------------------------------------------------- +# Etag helpers +# --------------------------------------------------------------------------- + + +def _generate_etag() -> int: + """Generate a random non-zero 32-bit etag.""" + while True: + value = int.from_bytes(os.urandom(4), "big") + if value != 0: + return value + + +# --------------------------------------------------------------------------- +# PartialState +# --------------------------------------------------------------------------- + +# Serialization header: b"PDST" magic(4) + version(B) + etag(4BE) + width(4LE) +# + height(4LE) + bpp(4LE) +# Followed by: img_len(4BE) + img_bytes(img_len) +_MAGIC = b"PDST" +_VERSION = 1 +_HEADER_FMT = ">4sBIIII" # magic(4s), version(B), etag(I BE), width(I), height(I), bpp(I) +_HEADER_SIZE = struct.calcsize(_HEADER_FMT) # 4+1+4+4+4+4 = 21 bytes + + +@dataclass +class PartialState: + """Mutable, opaque state tracked by the caller across partial updates. + + Treat fields as opaque; persist only via ``to_bytes`` / ``from_bytes``. + + ``last_image`` holds raw palette bytes (1 byte per pixel) from + ``PIL.Image.tobytes()`` on the dithered palette image. + ``bytes_per_pixel`` is always 1 for library-populated instances. + """ + + etag: int = 0 # last new_etag successfully sent (0 = unknown) + last_image: bytes | None = None # raw palette pixel buffer matching etag + width: int = 0 + height: int = 0 + bytes_per_pixel: int = 0 + + # ------------------------------------------------------------------ + # Serialization + # ------------------------------------------------------------------ + + def to_bytes(self) -> bytes: + """Serialize state to bytes (struct header + length-prefixed image).""" + img = self.last_image or b"" + header = struct.pack( + _HEADER_FMT, + _MAGIC, + _VERSION, + self.etag, + self.width, + self.height, + self.bytes_per_pixel, + ) + img_len_bytes = struct.pack(">I", len(img)) + return header + img_len_bytes + img + + @classmethod + def from_bytes(cls, data: bytes) -> "PartialState": + """Deserialize state from bytes produced by ``to_bytes``.""" + min_len = _HEADER_SIZE + 4 # header + img_len field + if len(data) < min_len: + raise ValueError(f"PartialState data too short: {len(data)} bytes (need {min_len})") + magic, version, etag, width, height, bpp = struct.unpack_from(_HEADER_FMT, data, 0) + if magic != _MAGIC: + raise ValueError(f"PartialState magic mismatch: {magic!r}") + if version != _VERSION: + raise ValueError(f"PartialState version unsupported: {version}") + (img_len,) = struct.unpack_from(">I", data, _HEADER_SIZE) + offset = _HEADER_SIZE + 4 + if len(data) < offset + img_len: + raise ValueError("PartialState image data truncated") + img: bytes | None = data[offset : offset + img_len] if img_len > 0 else None + return cls(etag=etag, last_image=img, width=width, height=height, bytes_per_pixel=bpp) diff --git a/src/opendisplay/protocol/__init__.py b/src/opendisplay/protocol/__init__.py index 3b2f0fc..7b70da5 100644 --- a/src/opendisplay/protocol/__init__.py +++ b/src/opendisplay/protocol/__init__.py @@ -14,6 +14,8 @@ build_authenticate_step2, build_direct_write_data_command, build_direct_write_end_command, + build_direct_write_end_with_etag, + build_direct_write_partial_start, build_direct_write_start_compressed, build_direct_write_start_uncompressed, build_led_activate_command, @@ -50,8 +52,10 @@ "build_write_config_command", "build_direct_write_start_compressed", "build_direct_write_start_uncompressed", + "build_direct_write_partial_start", "build_direct_write_data_command", "build_direct_write_end_command", + "build_direct_write_end_with_etag", "build_led_activate_command", "parse_config_response", "serialize_config", diff --git a/src/opendisplay/protocol/commands.py b/src/opendisplay/protocol/commands.py index 786d1e9..66f01a6 100644 --- a/src/opendisplay/protocol/commands.py +++ b/src/opendisplay/protocol/commands.py @@ -2,6 +2,7 @@ from __future__ import annotations +import struct from enum import IntEnum from ..models.led_flash import LedFlashConfig @@ -31,6 +32,7 @@ class CommandCode(IntEnum): 0x0073 # Device→host: refresh finished (same code as LED_ACTIVATE, different direction) ) DIRECT_WRITE_REFRESH_TIMEOUT = 0x0074 # Device→host: refresh timed out + DIRECT_WRITE_PARTIAL_START = 0x0076 # Start a partial update transfer (stream via 0x71) # Protocol constants @@ -133,6 +135,50 @@ def build_direct_write_start_uncompressed() -> bytes: return CommandCode.DIRECT_WRITE_START.to_bytes(2, byteorder="big") +def build_direct_write_partial_start( + old_etag: int, + new_etag: int, + flags: int, + x: int, + y: int, + width: int, + height: int, + stream_bytes: bytes = b"", +) -> tuple[bytes, bytes]: + """Build 0x76 partial START packet. + + Fixed payload is 17 bytes; optional initial stream bytes are appended up + to MAX_START_PAYLOAD total packet size (including the 2-byte command). + + Wire fixed payload: + flags(1) + old_etag(4BE) + new_etag(4BE) + x(2BE) + y(2BE) + + width(2BE) + height(2BE) + + Returns: + (start_packet, remaining_stream_bytes) — send start_packet as the + 0x76 command, then remaining_stream_bytes via 0x71 DATA chunks. + """ + if not 0 <= flags <= 0xFF: + raise ValueError(f"partial flags out of uint8 range: {flags}") + if not 0 <= old_etag <= 0xFFFFFFFF: + raise ValueError(f"old_etag must be uint32, got {old_etag}") + if not 0 <= new_etag <= 0xFFFFFFFF: + raise ValueError(f"new_etag must be uint32, got {new_etag}") + + fixed = ( + struct.pack(">B", flags) + + struct.pack(">I", old_etag) + + struct.pack(">I", new_etag) + + struct.pack(">HHHH", x, y, width, height) + ) # 1+4+4+2+2+2+2 = 17 bytes + + cmd = CommandCode.DIRECT_WRITE_PARTIAL_START.to_bytes(2, byteorder="big") + max_initial = MAX_START_PAYLOAD - 2 - len(fixed) # 200 - 2 - 17 = 181 bytes + initial = stream_bytes[:max_initial] + remaining = stream_bytes[max_initial:] + return cmd + fixed + initial, remaining + + def build_direct_write_data_command(chunk_data: bytes) -> bytes: """Build command to send image data chunk. @@ -175,6 +221,14 @@ def build_direct_write_end_command(refresh_mode: int = 0) -> bytes: return cmd + refresh +def build_direct_write_end_with_etag(refresh_mode: int, new_etag: int) -> bytes: + """Build 0x72 END with a new_etag tail. Etag presence is by length only.""" + if not 0 <= new_etag <= 0xFFFFFFFF: + raise ValueError(f"new_etag out of uint32 range: {new_etag}") + cmd = CommandCode.DIRECT_WRITE_END.to_bytes(2, byteorder="big") + 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/tests/unit/test_device_partial.py b/tests/unit/test_device_partial.py new file mode 100644 index 0000000..5c7a205 --- /dev/null +++ b/tests/unit/test_device_partial.py @@ -0,0 +1,213 @@ +"""Client-flow tests for streamed partial uploads.""" + +from __future__ import annotations + +import asyncio + +from epaper_dithering import ColorScheme +from PIL import Image + +from opendisplay import OpenDisplayDevice +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 ERR_ETAG_MISMATCH, PARTIAL_FLAG_COMPRESSED, PartialState + + +def _config(partial_update_support: int = 1, transmission_modes: int = 0x00) -> GlobalConfig: + 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=16, + 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=partial_update_support, + 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 _device(config: GlobalConfig | None = None) -> OpenDisplayDevice: + return OpenDisplayDevice( + mac_address="AA:BB:CC:DD:EE:FF", + config=config or _config(), + capabilities=DeviceCapabilities(width=16, height=8, color_scheme=ColorScheme.MONO), + ) + + +def _image(changed: bool = False) -> Image.Image: + img = Image.new("P", (16, 8), 0) + if changed: + img.putpixel((13, 3), 1) + return img + + +def test_no_change_image_skips_transfer(monkeypatch): + device = _device() + state = PartialState(etag=0x01020304, last_image=_image().tobytes(), width=16, height=8, bytes_per_pixel=1) + + async def fail_write(data: bytes) -> None: + raise AssertionError(f"unexpected write: {data!r}") + + monkeypatch.setattr(device, "_write", fail_write) + + outcome = asyncio.run(device._maybe_upload_partial(_image(), state, None)) + + assert outcome == "no_change" + assert state.etag == 0x01020304 + + +def test_valid_partial_never_sends_0x70_and_uses_uncompressed_0x76(monkeypatch): + device = _device() + old = _image() + new = _image(changed=True) + state = PartialState(etag=0x01020304, last_image=old.tobytes(), width=16, height=8, bytes_per_pixel=1) + writes: list[bytes] = [] + responses = [b"\x00\x76", b"\x00\x72", b"\x00\x73"] + + async def capture_write(data: bytes) -> None: + writes.append(data) + + async def read_response(timeout: float) -> bytes: + return responses.pop(0) + + monkeypatch.setattr(device, "_write", capture_write) + monkeypatch.setattr(device, "_read", read_response) + + outcome = asyncio.run(device._maybe_upload_partial(new, state, None)) + + opcodes = [int.from_bytes(w[:2], "big") for w in writes] + assert outcome == "success" + assert 0x70 not in opcodes + assert opcodes == [0x76, 0x72] + assert writes[0][2] & PARTIAL_FLAG_COMPRESSED == 0 + assert int.from_bytes(writes[0][3:7], "big") == 0x01020304 + assert int.from_bytes(writes[0][7:11], "big") == state.etag + assert len(writes[1]) == 3 + + +def test_empty_state_falls_back_to_full(monkeypatch): + device = _device() + state = PartialState() + full_uploads = 0 + refresh_modes: list[RefreshMode] = [] + + async def execute_upload(image_data, refresh_mode, **kwargs) -> None: + nonlocal full_uploads + full_uploads += 1 + refresh_modes.append(refresh_mode) + + monkeypatch.setattr(device, "_execute_upload", execute_upload) + + asyncio.run( + device.upload_prepared_image((b"\x00" * 16, None, _image()), refresh_mode=RefreshMode.PARTIAL, state=state) + ) + + assert full_uploads == 1 + assert refresh_modes == [RefreshMode.FULL] + assert state.etag != 0 + assert state.last_image == _image().tobytes() + + +def test_etag_mismatch_clears_state_and_retries_full_once(monkeypatch): + device = _device() + state = PartialState(etag=0x01020304, last_image=_image().tobytes(), width=16, height=8, bytes_per_pixel=1) + full_uploads = 0 + refresh_modes: list[RefreshMode] = [] + + async def capture_write(data: bytes) -> None: + pass + + async def read_response(timeout: float) -> bytes: + return bytes([0xFF, 0x76, ERR_ETAG_MISMATCH, 0x00]) + + async def execute_upload(image_data, refresh_mode, **kwargs) -> None: + nonlocal full_uploads + full_uploads += 1 + refresh_modes.append(refresh_mode) + + monkeypatch.setattr(device, "_write", capture_write) + monkeypatch.setattr(device, "_read", read_response) + monkeypatch.setattr(device, "_execute_upload", execute_upload) + + asyncio.run(device.upload_prepared_image((b"\x00" * 16, None, _image(changed=True)), state=state)) + + assert full_uploads == 1 + assert refresh_modes == [RefreshMode.FULL] + assert state.etag != 0 + assert state.last_image == _image(changed=True).tobytes() + + +def test_partial_request_uses_partial_even_when_full_compressed_is_smaller(monkeypatch): + device = _device(_config(transmission_modes=0x02)) + old = _image() + new = Image.new("P", (16, 8), 1) + state = PartialState(etag=0x01020304, last_image=old.tobytes(), width=16, height=8, bytes_per_pixel=1) + writes: list[bytes] = [] + responses = [b"\x00\x76", b"\x00\x72", b"\x00\x73"] + + async def capture_write(data: bytes) -> None: + writes.append(data) + + async def read_response(timeout: float) -> bytes: + return responses.pop(0) + + async def fail_full_upload(*args, **kwargs) -> None: + raise AssertionError("partial request unexpectedly fell back to full upload") + + monkeypatch.setattr(device, "_write", capture_write) + monkeypatch.setattr(device, "_read", read_response) + monkeypatch.setattr(device, "_execute_upload", fail_full_upload) + + asyncio.run( + device.upload_prepared_image((b"\xff" * 16, b"\x01", new), refresh_mode=RefreshMode.PARTIAL, state=state) + ) + + opcodes = [int.from_bytes(w[:2], "big") for w in writes] + assert opcodes == [0x76, 0x72] + assert len(writes[-1]) == 3 diff --git a/tests/unit/test_partial.py b/tests/unit/test_partial.py new file mode 100644 index 0000000..c7cd475 --- /dev/null +++ b/tests/unit/test_partial.py @@ -0,0 +1,165 @@ +"""Unit tests for the 0x76 streamed-rectangle partial protocol.""" + +from __future__ import annotations + +import pytest + +from opendisplay.partial import ( + ERR_ETAG_MISMATCH, + ERR_MIXED_DATA, + ERR_RECT_ALIGN, + ERR_RECT_OOB, + PARTIAL_FLAG_COMPRESSED, + PartialState, + _generate_etag, + align_rect, + build_partial_logical_stream, + compute_bounding_rect, + parse_nack, +) +from opendisplay.protocol.commands import ( + MAX_START_PAYLOAD, + build_direct_write_end_with_etag, + build_direct_write_partial_start, +) + + +class TestPartialState: + def test_roundtrip_empty(self): + state = PartialState() + assert PartialState.from_bytes(state.to_bytes()) == state + + def test_roundtrip_populated(self): + state = PartialState( + etag=0xDEADBEEF, + last_image=bytes(range(256)) * 4, + width=480, + height=800, + bytes_per_pixel=1, + ) + + out = PartialState.from_bytes(state.to_bytes()) + + assert out.etag == state.etag + assert out.last_image == state.last_image + assert out.width == state.width + assert out.height == state.height + assert out.bytes_per_pixel == state.bytes_per_pixel + + def test_bad_magic_rejected(self): + with pytest.raises(ValueError, match="magic"): + PartialState.from_bytes(b"XXXX" + b"\x00" * 21) + + def test_truncated_rejected(self): + with pytest.raises(ValueError): + PartialState.from_bytes(b"PDST") + + +class TestParseNack: + def test_known_errors(self): + assert parse_nack(b"\xff\x76\x01\x00") == (0x76, ERR_ETAG_MISMATCH) + assert parse_nack(b"\xff\x70\x02\x00") == (0x70, ERR_MIXED_DATA) + assert parse_nack(b"\xff\x76\x03\x00") == (0x76, ERR_RECT_OOB) + assert parse_nack(b"\xff\x76\x04\x00") == (0x76, ERR_RECT_ALIGN) + + def test_not_nack_returns_none(self): + assert parse_nack(b"\x00\x76") is None + assert parse_nack(b"\xff\x70\x01") is None + assert parse_nack(b"\xff\x70\x01\x00\x00") is None + assert parse_nack(b"\xff\x70\x01\x01") is None + + +class TestGenerateEtag: + def test_nonzero_uint32(self): + for _ in range(100): + value = _generate_etag() + assert 1 <= value <= 0xFFFFFFFF + + +class TestBoundingRect: + def test_no_change_returns_none(self): + buf = bytes(100) + assert compute_bounding_rect(buf, buf, 10, 10) is None + + def test_single_pixel_change(self): + old = bytearray(16 * 8) + new = bytearray(old) + new[3 * 16 + 13] = 1 + + assert compute_bounding_rect(bytes(old), bytes(new), 16, 8) == (13, 3, 14, 4) + + def test_multiple_changes_one_bounding_box(self): + old = bytearray(32 * 20) + new = bytearray(old) + new[2 * 32 + 1] = 1 + new[15 * 32 + 28] = 1 + + assert compute_bounding_rect(bytes(old), bytes(new), 32, 20) == (1, 2, 29, 16) + + +class TestAlignRect: + def test_mono_expands_to_8_pixels(self): + assert align_rect(13, 3, 14, 4, 32, 20, pixels_per_byte=8) == (8, 3, 8, 1) + + def test_2bpp_expands_to_4_pixels(self): + assert align_rect(5, 1, 7, 3, 32, 20, pixels_per_byte=4) == (4, 1, 4, 2) + + def test_4bpp_expands_to_2_pixels(self): + assert align_rect(5, 1, 6, 3, 32, 20, pixels_per_byte=2) == (4, 1, 2, 2) + + def test_right_edge_clamp_keeps_width_aligned(self): + assert align_rect(30, 1, 32, 2, 32, 20, pixels_per_byte=8) == (24, 1, 8, 1) + + +class TestLogicalStream: + def test_old_then_new_plane_major_stream(self): + stream = build_partial_logical_stream(b"abcdef", b"ABCDEF") + assert stream == b"abcdefABCDEF" + + def test_rejects_mismatched_rect_lengths(self): + with pytest.raises(AssertionError, match="old/new rect byte lengths"): + build_partial_logical_stream(b"abcde", b"ABCDEF") + + def test_stream_accounting(self): + old = bytes(range(16)) + new = bytes(range(16, 32)) + stream = build_partial_logical_stream(old, new) + + assert len(stream) == 32 + assert stream[:16] == old + assert stream[16:] == new + + +class TestBuilders: + def test_partial_start_fixed_fields_and_initial_bytes(self): + stream = bytes(range(200)) + packet, remaining = build_direct_write_partial_start( + old_etag=0xDEADBEEF, + new_etag=0x01020304, + flags=PARTIAL_FLAG_COMPRESSED, + x=8, + y=9, + width=16, + height=10, + stream_bytes=stream, + ) + + assert len(packet) == MAX_START_PAYLOAD + assert packet[:2] == b"\x00\x76" + assert packet[2] == PARTIAL_FLAG_COMPRESSED + assert int.from_bytes(packet[3:7], "big") == 0xDEADBEEF + assert int.from_bytes(packet[7:11], "big") == 0x01020304 + assert int.from_bytes(packet[11:13], "big") == 8 + assert int.from_bytes(packet[13:15], "big") == 9 + assert int.from_bytes(packet[15:17], "big") == 16 + assert int.from_bytes(packet[17:19], "big") == 10 + assert packet[19:] == stream[:181] + assert remaining == stream[181:] + + def test_partial_start_allows_zero_etag(self): + packet, _ = build_direct_write_partial_start(0, 0x01020304, 0, 0, 0, 8, 1) + assert packet[3:7] == b"\x00\x00\x00\x00" + + def test_end_with_etag(self): + cmd = build_direct_write_end_with_etag(refresh_mode=2, new_etag=0x01020304) + assert cmd == b"\x00\x72\x02\x01\x02\x03\x04"