diff --git a/src/glanceable/damage.py b/src/glanceable/damage.py new file mode 100644 index 0000000..d8e9393 --- /dev/null +++ b/src/glanceable/damage.py @@ -0,0 +1,229 @@ +"""Damage tracking for surfaces that cannot afford a repaint. + +Device-free by construction: this module knows about ``Surface`` ops and +rectangles, nothing else. It lives above ``surface.py`` and stays there. + +Named ``damage`` rather than ``geometry`` because ``geometry.py`` is the chord +solver and these are different problems -- one is about the shape of the glass, +this is about what changed since the last frame. +""" + +from __future__ import annotations + +from dataclasses import dataclass, field + +#: Merge damage boxes within this many pixels. One background fill plus a few +#: redundant redraws beats two round trips. +DEFAULT_MERGE_GAP = 6 + +#: Above this many regions the per-region overhead has overtaken the saving. +DEFAULT_MAX_REGIONS = 6 + + +@dataclass(frozen=True) +class Box: + """Axis-aligned rectangle. ``x``/``y`` inclusive, ``w``/``h`` extents.""" + + x: int + y: int + w: int + h: int + + @property + def right(self) -> int: + return self.x + self.w + + @property + def bottom(self) -> int: + return self.y + self.h + + @property + def area(self) -> int: + return max(0, self.w) * max(0, self.h) + + @property + def is_empty(self) -> bool: + return self.w <= 0 or self.h <= 0 + + def intersects(self, other: "Box", gap: int = 0) -> bool: + if self.is_empty or other.is_empty: + return False + return ( + self.x - gap < other.right + and other.x - gap < self.right + and self.y - gap < other.bottom + and other.y - gap < self.bottom + ) + + def union(self, other: "Box") -> "Box": + if self.is_empty: + return other + if other.is_empty: + return self + x, y = min(self.x, other.x), min(self.y, other.y) + return Box(x, y, max(self.right, other.right) - x, max(self.bottom, other.bottom) - y) + + def clip(self, bounds: "Box") -> "Box": + x, y = max(self.x, bounds.x), max(self.y, bounds.y) + return Box(x, y, min(self.right, bounds.right) - x, min(self.bottom, bounds.bottom) - y) + + +EMPTY = Box(0, 0, 0, 0) + + +class Op: + """One recorded ``Surface`` call.""" + + def box(self) -> Box: # pragma: no cover - interface + raise NotImplementedError + + def replay(self, surface) -> None: # pragma: no cover - interface + raise NotImplementedError + + +@dataclass(frozen=True) +class FillRect(Op): + x: int + y: int + w: int + h: int + color_index: int + + def box(self) -> Box: + return Box(self.x, self.y, self.w, self.h) + + def replay(self, surface) -> None: + surface.fill_rect(self.x, self.y, self.w, self.h, self.color_index) + + +@dataclass(frozen=True) +class BlitCoverage(Op): + """A coverage blit. + + Equality is by ``digest`` -- the coverage bytes -- so an identical run + re-rendered at the same place compares equal and produces no damage. The + image itself is excluded from comparison but carried for replay. + """ + + x: int + y: int + w: int + h: int + palette_base: int + levels: int + digest: bytes + image: object = field(compare=False, repr=False, default=None) + + def box(self) -> Box: + return Box(self.x, self.y, self.w, self.h) + + def replay(self, surface) -> None: + surface.blit_coverage(self.image, self.x, self.y, self.palette_base, self.levels) + + +@dataclass(frozen=True) +class Region: + """A rectangle to repaint, and the ops inside it in draw order. + + ``needs_fill`` is False for a pure addition onto known background -- the + fill would be a wasted write. + """ + + box: Box + ops: tuple[Op, ...] + needs_fill: bool = True + + +@dataclass(frozen=True) +class DamagePlan: + regions: tuple[Region, ...] = () + first_paint: bool = False + + @property + def is_empty(self) -> bool: + return not self.regions + + @property + def area(self) -> int: + return sum(r.box.area for r in self.regions) + + +def merge_boxes(boxes, gap: int = DEFAULT_MERGE_GAP) -> list[Box]: + """Transitively merge overlapping or near-adjacent boxes to a fixpoint.""" + pending = [b for b in boxes if not b.is_empty] + merged: list[Box] = [] + while pending: + current = pending.pop() + absorbed = True + while absorbed: + absorbed = False + rest = [] + for other in pending: + if current.intersects(other, gap=gap): + current = current.union(other) + absorbed = True + else: + rest.append(other) + pending = rest + merged.append(current) + merged.sort(key=lambda b: (b.y, b.x)) + return merged + + +def plan( + previous: tuple[Op, ...] | None, + current: tuple[Op, ...], + bounds: Box, + *, + merge_gap: int = DEFAULT_MERGE_GAP, + max_regions: int = DEFAULT_MAX_REGIONS, +) -> DamagePlan: + """Minimal repaint turning ``previous`` into ``current``. + + ``previous`` of ``None`` means the panel contents are unknown, so all of + ``bounds`` is damaged. + """ + if previous is None: + keep = tuple(op for op in current if op.box().clip(bounds).area > 0) + return DamagePlan(regions=(Region(bounds, keep),), first_paint=True) + + if previous == current: + return DamagePlan() + + from collections import Counter + + old, new = Counter(previous), Counter(current) + removed, added = old - new, new - old + + # Removed ops leave pixels behind and must be filled. Added ops only need a + # fill where the previous frame actually put something. + fill = [op.box() for op in removed.elements()] + additive = [] + old_boxes = [op.box() for op in previous] + for op in added.elements(): + box = op.box() + (fill if any(box.intersects(b) for b in old_boxes) else additive).append(box) + + fill = [b.clip(bounds) for b in fill] + additive = [b.clip(bounds) for b in additive] + boxes = merge_boxes(fill + additive, gap=merge_gap) + + if len(boxes) > max_regions: + merged = EMPTY + for b in boxes: + merged = merged.union(b) + boxes = [merged] + + if not boxes: + return DamagePlan() + + return DamagePlan( + regions=tuple( + Region( + box, + tuple(op for op in current if op.box().intersects(box)), + needs_fill=any(d.intersects(box) for d in fill), + ) + for box in boxes + ) + ) diff --git a/src/glanceable/retained.py b/src/glanceable/retained.py new file mode 100644 index 0000000..bc140e1 --- /dev/null +++ b/src/glanceable/retained.py @@ -0,0 +1,104 @@ +"""Retained-mode wrapper for any ``Surface``. + +Wraps a surface, buffers the ops for a frame, and on ``present()`` forwards +only what changed since the last frame. Works over ``PILSurface`` and +``LuaSurface`` identically, because it is written against the ABC and knows +nothing about either. + +It exists because of two facts that only bite on real hardware: + +* **Nothing erases.** ``blit_coverage`` writes ink pixels through a mask, so + re-rendering a shorter line leaves the tail of the previous line on the + glass. On ``PILSurface`` you only see this if a test reuses the image; on + device it is guaranteed. +* **There is no back buffer.** ``display.show()`` is a registered no-op in the + Halo firmware, so drawing lands in the buffer the panel is scanning out. A + clear-and-repaint is a full-field luminance transient -- the exact + pre-attentional motion this library exists to avoid. + +The saving is bandwidth, not statements. A 240x18 coverage map at 4 levels is +1,080 bytes; a five-line HUD is 5,400. Repainting one changed line ships 1,080 +instead, which is why this wrapper is load-bearing rather than an +optimisation. +""" + +from __future__ import annotations + +import hashlib + +from PIL import Image + +from .damage import BlitCoverage, Box, FillRect, Op, plan +from .surface import Surface + + +class RetainedSurface(Surface): + """Buffers a frame and forwards only the damaged regions on ``present()``. + + Args: + inner: the surface actually driving pixels. + background_index: palette entry to fill damaged regions with before + redrawing. Must match the background the coverage blits assume -- + entry 0 for ``ramp_palette``. + """ + + def __init__(self, inner: Surface, background_index: int = 0): + self._inner = inner + self._background = background_index + self._pending: list[Op] = [] + self._committed: tuple[Op, ...] | None = None + self.last_plan = None + + @property + def size(self) -> tuple[int, int]: + return self._inner.size + + @property + def bounds(self) -> Box: + w, h = self._inner.size + return Box(0, 0, w, h) + + def invalidate(self) -> None: + """Forget what is on the panel; the next present repaints everything.""" + self._committed = None + + def fill_rect(self, x: int, y: int, w: int, h: int, color_index: int) -> None: + if w <= 0 or h <= 0: + return + self._pending.append(FillRect(x, y, w, h, color_index)) + + def blit_coverage( + self, coverage: Image.Image, x: int, y: int, palette_base: int, levels: int + ) -> None: + digest = hashlib.blake2b(coverage.tobytes(), digest_size=16).digest() + self._pending.append( + BlitCoverage( + x=x, + y=y, + w=coverage.width, + h=coverage.height, + palette_base=palette_base, + levels=levels, + digest=digest, + image=coverage, + ) + ) + + def present(self) -> None: + current = tuple(self._pending) + self._pending.clear() + + damage = plan(self._committed, current, self.bounds) + self.last_plan = damage + + for region in damage.regions: + box = region.box.clip(self.bounds) + if box.is_empty: + continue + if region.needs_fill: + self._inner.fill_rect(box.x, box.y, box.w, box.h, self._background) + for op in region.ops: + op.replay(self._inner) + + self._committed = current + self._inner.present() diff --git a/src/glanceable/surface.py b/src/glanceable/surface.py index 0ce68cc..53a8bdd 100644 --- a/src/glanceable/surface.py +++ b/src/glanceable/surface.py @@ -172,22 +172,70 @@ class SpriteSurface(Surface): bytes, so they wrap at 256. Which codes are safe to use is the application's business -- the SDK does not reserve a range. - NOTE: field shapes are checked against the published brilliant_msg 7.0.0 - classes, but this has NOT been run on hardware. Treat the wire format as - unconfirmed until it has been round-tripped on a physical Halo. + Field shapes are verified field-for-field against brilliant_msg 7.1.1: + SpritePayload matches TxSprite (width, height, num_colors, palette_data, + pixel_data, compress) and SpriteCoords matches TxSpriteCoords (code, x, y, + offset), both in declaration order, so asdict() splats cleanly. msg_code is + applied by BrilliantMsg.send_message() at send time and correctly absent + from both. + + Halo is supported: brilliant_msg ships device-side sprite.lua whose + set_palette() branches on frame.HARDWARE_VERSION, using integer palette + indices 0-15 on Halo against colour names on Frame, then renders through + frame.display.bitmap. + + Pixels stay one byte per pixel here. TxSprite.pack() does the bit packing + (_pack_1bit/_pack_2bit/_pack_4bit); pre-packing would double-encode. + + STILL UNCONFIRMED, and not resolvable by reading the SDK: the x/y origin. + sprite_coords.lua only parses the fields -- what they mean is decided by + the app-side Lua that calls frame.display.bitmap, which is 1-based on Halo. + TxSpriteCoords documents x as 1..640, Frame's panel. Needs a device. """ def __init__( self, width: int, height: int, palette: list[int], base_code: int = 0x20 ): self._size = (width, height) - self._palette = bytes(palette) - self._num_colors = max(2, len(palette) // 3) + + # TxSprite.pack() buckets bpp by num_colors (<=2 -> 1bpp, <=4 -> 2bpp, + # else 4bpp) and the device-side sprite.lua slices the palette as + # exactly num_colors*3 bytes, treating everything after as pixel data. + # An unrounded count would desynchronise that slice. + supplied = max(2, len(palette) // 3) + self._num_colors = 2 if supplied <= 2 else 4 if supplied <= 4 else 16 + if supplied > 16: + raise ValueError( + f"palette holds {supplied} colours; the sprite format caps at 16" + ) + + # Truncate to the declared count. sprite.lua reads the palette as + # string.sub(data, 8, 8 + num_colors*3 - 1) and takes the remainder as + # pixels, so a longer palette shifts every pixel and corrupts the frame. + # brilliant_msg's own from_indexed_png_bytes truncates the same way. + self._palette = bytes(palette[: self._num_colors * 3]).ljust( + self._num_colors * 3, b"\x00" + ) + self._base_code = base_code + self._code_seq = 0 self.ops: list[SpriteOp] = [] def _next_code(self) -> int: - return (self._base_code + len(self.ops)) & 0xFF + """Per-sprite identifier, cycling within the byte range above base. + + Previously derived from len(self.ops), which never resets because + present() does not clear the log -- so codes wrapped past 0xFF and + collided with sprites still live on the device. + """ + span = 0x100 - self._base_code + code = self._base_code + (self._code_seq % span) + self._code_seq += 1 + return code + + def reset_codes(self) -> None: + """Restart code allocation. Call when the device display is cleared.""" + self._code_seq = 0 def _emit(self, w: int, h: int, pixels: bytes, x: int, y: int) -> None: self.ops.append( @@ -229,3 +277,4 @@ def blit_coverage( def present(self) -> None: pass + diff --git a/tests/test_retained.py b/tests/test_retained.py new file mode 100644 index 0000000..3b1749c --- /dev/null +++ b/tests/test_retained.py @@ -0,0 +1,178 @@ +"""Sprite wire-format and retained-mode tests. + +The wire-format tests encode facts verified against brilliant_msg 7.1.1 source +(tx_sprite.py, tx_sprite_coords.py, lua/sprite.lua). They are the guardrail +that stops a future edit from silently desynchronising the device parser. +""" + +import dataclasses + +import pytest +from PIL import Image + +from glanceable.retained import RetainedSurface +from glanceable.surface import ( + PILSurface, + SpriteCoords, + SpritePayload, + SpriteSurface, + Surface, +) +from glanceable.typography import ramp_palette + +PALETTE = ramp_palette(4) + + +def coverage(w, h, value=255): + return Image.new("L", (w, h), value) + + +# -- wire format against brilliant_msg 7.1.1 ------------------------------ + + +def test_payload_fields_match_txsprite_exactly(): + """TxSprite(**asdict(payload)) must construct without translation.""" + assert [f.name for f in dataclasses.fields(SpritePayload)] == [ + "width", + "height", + "num_colors", + "palette_data", + "pixel_data", + "compress", + ] + + +def test_coords_fields_match_txspritecoords_exactly(): + assert [f.name for f in dataclasses.fields(SpriteCoords)] == [ + "code", + "x", + "y", + "offset", + ] + + +def test_compress_defaults_false(): + """The compressed flag is header byte 5; sprite.lua reads it positionally.""" + assert SpritePayload(1, 1, 2, b"", b"").compress is False + + +def test_palette_length_always_matches_the_declared_colour_count(): + """sprite.lua slices exactly num_colors*3 bytes and treats the rest as + pixels, so any mismatch shifts every pixel in the frame.""" + for levels in (2, 4, 16): + s = SpriteSurface(256, 256, ramp_palette(levels)) + assert len(s._palette) == s._num_colors * 3 + + +def test_palette_is_padded_up_to_the_bucket_not_left_short(): + """A 5-colour ramp rounds up to the 16-colour format; the wire palette must + still be 48 bytes or the device parser reads pixels as palette.""" + s = SpriteSurface(256, 256, ramp_palette(5)) + assert s._num_colors == 16 + assert len(s._palette) == 48 + + +def test_num_colors_is_rounded_to_a_format_the_packer_supports(): + """TxSprite.pack() buckets bpp as <=2, <=4, else 4bpp.""" + assert SpriteSurface(64, 64, ramp_palette(2))._num_colors == 2 + assert SpriteSurface(64, 64, ramp_palette(4))._num_colors == 4 + assert SpriteSurface(64, 64, ramp_palette(16))._num_colors == 16 + + +def test_oversized_palette_is_rejected(): + with pytest.raises(ValueError, match="caps at 16"): + SpriteSurface(64, 64, [0] * (17 * 3)) + + +def test_sprite_codes_do_not_collide_before_wrapping(): + """Previously derived from len(ops), which never resets, so codes ran past + 0xFF and collided with sprites still live on the device.""" + s = SpriteSurface(256, 256, PALETTE, base_code=0x20) + codes = [s._next_code() for _ in range(0x100 - 0x20)] + assert len(set(codes)) == len(codes) + + +def test_sprite_codes_wrap_back_to_base_not_to_zero(): + s = SpriteSurface(256, 256, PALETTE, base_code=0x20) + codes = [s._next_code() for _ in range(0x100 - 0x20 + 1)] + assert codes[-1] == 0x20 + + +def test_reset_codes_restarts_allocation(): + s = SpriteSurface(256, 256, PALETTE) + first = s._next_code() + s._next_code() + s.reset_codes() + assert s._next_code() == first + + +def test_pixel_data_stays_one_byte_per_pixel(): + """TxSprite.pack() does the bit packing; pre-packing would double-encode.""" + s = SpriteSurface(256, 256, PALETTE) + s.fill_rect(0, 0, 4, 3, 1) + assert len(s.ops[0].payload.pixel_data) == 12 + + +# -- retained mode over both backends ------------------------------------- + + +def test_retained_works_over_the_sprite_backend(): + inner = SpriteSurface(256, 256, PALETTE) + s = RetainedSurface(inner) + cov = coverage(120, 18) + + s.blit_coverage(cov, 20, 100, 0, 4) + s.present() + after_first = len(inner.ops) + + s.blit_coverage(cov, 20, 100, 0, 4) + s.present() + + assert len(inner.ops) == after_first, "an unchanged frame must emit no sprites" + + +def test_retained_erases_the_previous_extent_on_the_sprite_backend(): + inner = SpriteSurface(256, 256, PALETTE) + s = RetainedSurface(inner) + + s.blit_coverage(coverage(200, 18), 20, 100, 0, 4) + s.present() + inner.ops.clear() + + s.blit_coverage(coverage(80, 18), 20, 100, 0, 4) + s.present() + + assert any(op.width >= 200 for op in inner.ops), "old extent must be covered" + + +def test_retained_only_touches_the_changed_line(): + inner = SpriteSurface(256, 256, PALETTE) + s = RetainedSurface(inner) + for i in range(5): + s.blit_coverage(coverage(200, 18), 20, 40 + i * 20, 0, 4) + s.present() + inner.ops.clear() + + for i in range(5): + s.blit_coverage(coverage(200, 18, 128 if i == 2 else 255), 20, 40 + i * 20, 0, 4) + s.present() + + assert len(inner.ops) < 5 + + +def test_both_backends_agree_on_op_geometry(): + """The surface-agreement property: identical calls, identical extents.""" + pil, spr = PILSurface(256, 256, PALETTE), SpriteSurface(256, 256, PALETTE) + for surface in (pil, spr): + s = RetainedSurface(surface) + s.fill_rect(10, 10, 40, 20, 1) + s.blit_coverage(coverage(64, 18), 12, 40, 0, 4) + s.present() + + assert [(o[0], o[1], o[2], o[3]) for o in pil.ops] == [ + (o.x, o.y, o.width, o.height) for o in spr.ops + ] + + +def test_retained_surface_satisfies_the_abc(): + assert isinstance(RetainedSurface(PILSurface(64, 64, PALETTE)), Surface)