feat(badapple): tile delta format and movie packer

bad-apple/pipeline/movie.py

@@ -0,0 +1,189 @@
+"""The on-RAM movie format: header, index, and per-frame tile deltas."""
+from __future__ import annotations
+
+import os
+import struct
+from dataclasses import dataclass, field
+from typing import Sequence
+
+MAGIC = b"BAPL"
+VERSION = 1
+HEADER_SIZE = 36
+INDEX_ENTRY_SIZE = 6
+
+#: The data must stop below the payload's rdebug origin at byte 0x318000.
+MAX_DATA_BYTES = 0x308000
+
+
+class MovieFormatError(Exception):
+    pass
+
+
+@dataclass(frozen=True)
+class Rect:
+    x0: int
+    y0: int
+    w: int
+    h: int
+
+
+@dataclass(frozen=True)
+class TileRecord:
+    tile_index: int
+    bg: int
+    rects: list[Rect] = field(default_factory=list)
+
+
+@dataclass(frozen=True)
+class FrameRecord:
+    ticks: int
+    flags: int
+    tiles: list[TileRecord] = field(default_factory=list)
+
+
+@dataclass(frozen=True)
+class MovieHeader:
+    width: int
+    height: int
+    scale: int
+    tile: int
+    tiles_x: int
+    tiles_y: int
+    frame_count: int
+    source_fps: int
+    display_hz: int
+
+    def pack(self) -> bytes:
+        index_offset = HEADER_SIZE
+        data_offset = _align4(index_offset + INDEX_ENTRY_SIZE * self.frame_count)
+        return struct.pack(
+            "<4s7H3I3H",
+            MAGIC, VERSION, self.width, self.height, self.scale, self.tile,
+            self.tiles_x, self.tiles_y, self.frame_count,
+            index_offset, data_offset,
+            self.source_fps, self.display_hz, 0)
+
+
+def _align4(value: int) -> int:
+    return (value + 3) & ~3
+
+
+def _pack_rect(rect: Rect) -> bytes:
+    if not (0 <= rect.x0 < 32 and 0 <= rect.y0 < 32
+            and 1 <= rect.w <= 32 and 1 <= rect.h <= 32):
+        raise MovieFormatError("rect outside a 32x32 tile: %r" % (rect,))
+    value = (rect.x0 | (rect.y0 << 5) | ((rect.w - 1) << 10)
+             | ((rect.h - 1) << 15))
+    return value.to_bytes(3, "little")
+
+
+def _unpack_rect(data: bytes, offset: int) -> tuple[Rect, int]:
+    if offset + 3 > len(data):
+        raise MovieFormatError("truncated rect")
+    value = data[offset] | (data[offset + 1] << 8) | (data[offset + 2] << 16)
+    rect = Rect(
+        x0=value & 0x1F,
+        y0=(value >> 5) & 0x1F,
+        w=((value >> 10) & 0x1F) + 1,
+        h=((value >> 15) & 0x1F) + 1,
+    )
+    return rect, offset + 3
+
+
+def encode_frame(record: FrameRecord) -> bytes:
+    if len(record.tiles) > 0xFFFF:
+        raise MovieFormatError("too many dirty tiles in one frame")
+    out = bytearray(struct.pack("<H", len(record.tiles)))
+    for tile in record.tiles:
+        if not (0 <= tile.tile_index <= 0xFFFF):
+            raise MovieFormatError("tile index out of range")
+        if len(tile.rects) > 0x7FFF:
+            raise MovieFormatError("too many rects in one tile")
+        out += struct.pack("<HH", tile.tile_index,
+                           ((tile.bg & 1) << 15) | len(tile.rects))
+        for rect in tile.rects:
+            out += _pack_rect(rect)
+    return bytes(out)
+
+
+def decode_frame(data: bytes, offset: int) -> tuple[FrameRecord, int]:
+    if offset < 0 or offset + 2 > len(data):
+        raise MovieFormatError("truncated frame count")
+    count = data[offset] | (data[offset + 1] << 8)
+    offset += 2
+    tiles: list[TileRecord] = []
+    for _ in range(count):
+        if offset + 4 > len(data):
+            raise MovieFormatError("truncated tile header")
+        tile_index = data[offset] | (data[offset + 1] << 8)
+        packed = data[offset + 2] | (data[offset + 3] << 8)
+        offset += 4
+        bg = (packed >> 15) & 1
+        rect_count = packed & 0x7FFF
+        if offset + rect_count * 3 > len(data):
+            raise MovieFormatError("truncated tile rects")
+        rects: list[Rect] = []
+        for _ in range(rect_count):
+            rect, offset = _unpack_rect(data, offset)
+            rects.append(rect)
+        tiles.append(TileRecord(tile_index, bg, rects))
+    return FrameRecord(1, 0, tiles), offset
+
+
+def write_movie(path: str, frames: Sequence[FrameRecord],
+                header: MovieHeader) -> int:
+    """Write the movie; return its size.  Refuses anything over the cap."""
+    records = [encode_frame(frame) for frame in frames]
+    count = len(records)
+    index_offset = HEADER_SIZE
+    data_offset = _align4(index_offset + INDEX_ENTRY_SIZE * count)
+    total = data_offset + sum(len(record) for record in records)
+    if total > MAX_DATA_BYTES:
+        raise MovieFormatError(
+            "movie is %d bytes, over the %d-byte data cap"
+            % (total, MAX_DATA_BYTES))
+
+    out = bytearray(total)
+    out[0:HEADER_SIZE] = MovieHeader(
+        header.width, header.height, header.scale, header.tile,
+        header.tiles_x, header.tiles_y, count,
+        header.source_fps, header.display_hz).pack()
+    offset = data_offset
+    for i, (frame, record) in enumerate(zip(frames, records)):
+        struct.pack_into("<IBB", out, index_offset + i * INDEX_ENTRY_SIZE,
+                         offset, frame.ticks & 0xFF, frame.flags & 0xFF)
+        out[offset:offset + len(record)] = record
+        offset += len(record)
+
+    tmp = path + ".tmp"
+    with open(tmp, "wb") as handle:
+        handle.write(out)
+    os.replace(tmp, path)
+    return total
+
+
+def read_movie(path: str) -> tuple[MovieHeader, list[FrameRecord]]:
+    with open(path, "rb") as handle:
+        data = handle.read()
+    if len(data) < HEADER_SIZE:
+        raise MovieFormatError("truncated header")
+    fields = struct.unpack("<4s7H3I3H", data[:HEADER_SIZE])
+    (magic, version, width, height, scale, tile, tiles_x, tiles_y,
+     frame_count, index_offset, data_offset, source_fps, display_hz,
+     _reserved) = fields
+    if magic != MAGIC or version != VERSION:
+        raise MovieFormatError("bad magic or version")
+    header = MovieHeader(width, height, scale, tile, tiles_x, tiles_y,
+                         frame_count, source_fps, display_hz)
+    if index_offset + frame_count * INDEX_ENTRY_SIZE > len(data):
+        raise MovieFormatError("truncated index")
+
+    frames: list[FrameRecord] = []
+    for i in range(frame_count):
+        offset, ticks, flags = struct.unpack_from(
+            "<IBB", data, index_offset + i * INDEX_ENTRY_SIZE)
+        if offset < data_offset or offset > len(data):
+            raise MovieFormatError("frame offset outside data area")
+        record, _ = decode_frame(data, offset)
+        frames.append(FrameRecord(ticks, flags, record.tiles))
+    return header, frames

bad-apple/pipeline/tilepack.py

@@ -0,0 +1,81 @@
+"""Turn binary frames into per-tile background + rectangle deltas."""
+from __future__ import annotations
+
+import numpy as np
+
+from .movie import FrameRecord, Rect, TileRecord
+
+
+def _row_runs(row: np.ndarray) -> list[tuple[int, int]]:
+    """Inclusive (start, end) runs of True in one row."""
+    padded = np.concatenate(([0], row.astype(np.int8), [0]))
+    diff = np.diff(padded)
+    starts = np.where(diff == 1)[0]
+    ends = np.where(diff == -1)[0] - 1
+    return list(zip(starts.tolist(), ends.tolist()))
+
+
+def decompose_tile(tile: np.ndarray) -> tuple[int, list[Rect]]:
+    """Majority background plus rectangles of the minority colour."""
+    tile = tile.astype(bool)
+    if tile.ndim != 2:
+        raise ValueError("a tile must be 2-D")
+    height, width = tile.shape
+    black = int(tile.sum())
+    bg = 1 if black > tile.size // 2 else 0
+    fg = tile if bg == 0 else ~tile
+
+    active: dict[tuple[int, int], int] = {}
+    rects: list[Rect] = []
+    for y in range(height):
+        runs = set(_row_runs(fg[y]))
+        for key in list(active):
+            if key not in runs:
+                y0 = active.pop(key)
+                rects.append(Rect(key[0], y0, key[1] - key[0] + 1, y - y0))
+        for key in runs:
+            if key not in active:
+                active[key] = y
+    for key, y0 in active.items():
+        rects.append(Rect(key[0], y0, key[1] - key[0] + 1, height - y0))
+    rects.sort(key=lambda r: (r.y0, r.x0))
+    return bg, rects
+
+
+def diff_tiles(prev: np.ndarray, cur: np.ndarray,
+               tile: int = 32) -> list[TileRecord]:
+    """The tiles whose pixels changed, with the new tile's rectangle list."""
+    if prev.shape != cur.shape:
+        raise ValueError("prev and cur must have the same shape")
+    height, width = cur.shape
+    if height % tile or width % tile:
+        raise ValueError("frame dimensions must be a multiple of the tile")
+    tiles_x = width // tile
+    out: list[TileRecord] = []
+    for ty in range(height // tile):
+        for tx in range(tiles_x):
+            y = ty * tile
+            x = tx * tile
+            old = prev[y:y + tile, x:x + tile]
+            new = cur[y:y + tile, x:x + tile]
+            if np.array_equal(old, new):
+                continue
+            bg, rects = decompose_tile(new)
+            out.append(TileRecord(ty * tiles_x + tx, bg, rects))
+    return out
+
+
+def apply_record(canvas: np.ndarray, record: FrameRecord,
+                 tile: int = 32) -> None:
+    """Apply one frame record to a logical canvas in place."""
+    height, width = canvas.shape
+    tiles_x = width // tile
+    for rec in record.tiles:
+        ty, tx = divmod(rec.tile_index, tiles_x)
+        y = ty * tile
+        x = tx * tile
+        block = canvas[y:y + tile, x:x + tile]
+        block[:, :] = bool(rec.bg)
+        for rect in rec.rects:
+            block[rect.y0:rect.y0 + rect.h,
+                  rect.x0:rect.x0 + rect.w] = not bool(rec.bg)

bad-apple/tests/test_movie.py

@@ -0,0 +1,54 @@
+import numpy as np
+
+from pipeline.movie import (
+    Rect,
+    TileRecord,
+    FrameRecord,
+    MovieHeader,
+    MovieFormatError,
+    encode_frame,
+    decode_frame,
+    write_movie,
+    read_movie,
+)
+
+
+def test_rect_full_tile_bias_roundtrip():
+    rec = FrameRecord(1, 0, [TileRecord(0, 0, [Rect(0, 0, 32, 32)])])
+    back, off = decode_frame(encode_frame(rec), 0)
+    assert back.tiles[0].rects[0] == Rect(0, 0, 32, 32)
+    assert off == len(encode_frame(rec))
+
+
+def test_more_than_127_rects_roundtrip():
+    rects = [Rect(i % 32, i // 32, 1, 1) for i in range(200)]
+    rec = FrameRecord(1, 0, [TileRecord(3, 0, rects)])
+    back, _ = decode_frame(encode_frame(rec), 0)
+    assert len(back.tiles[0].rects) == 200
+
+
+def test_truncated_record_raises():
+    data = encode_frame(
+        FrameRecord(1, 0, [TileRecord(0, 0, [Rect(0, 0, 4, 4)])]))
+    try:
+        decode_frame(data[:-1], 0)
+    except MovieFormatError:
+        return
+    assert False, "truncated record was accepted"
+
+
+def test_write_movie_refuses_over_cap(tmp_path):
+    frames = [FrameRecord(1, 0, []) for _ in range(10)]
+    header = MovieHeader(640, 512, 2, 32, 20, 16, 10, 60, 72)
+    import pipeline.movie as m
+
+    old = m.MAX_DATA_BYTES
+    m.MAX_DATA_BYTES = 8
+    try:
+        try:
+            write_movie(str(tmp_path / "x.bin"), frames, header)
+        except MovieFormatError:
+            return
+        assert False, "cap not enforced"
+    finally:
+        m.MAX_DATA_BYTES = old

bad-apple/tests/test_tilepack.py

@@ -0,0 +1,40 @@
+import numpy as np
+
+from pipeline.movie import Rect, TileRecord, FrameRecord
+from pipeline.tilepack import decompose_tile, diff_tiles, apply_record
+
+
+def test_decompose_solid_tiles():
+    black = np.ones((32, 32), dtype=bool)
+    bg, rects = decompose_tile(black)
+    assert bg == 1 and rects == []
+    white = np.zeros((32, 32), dtype=bool)
+    bg, rects = decompose_tile(white)
+    assert bg == 0 and rects == []
+
+
+def test_decompose_block_is_one_rect():
+    tile = np.zeros((32, 32), dtype=bool)
+    tile[4:6, 7:10] = True
+    bg, rects = decompose_tile(tile)
+    assert bg == 0
+    assert rects == [Rect(7, 4, 3, 2)]
+
+
+def test_diff_only_reports_changed_tiles():
+    a = np.zeros((64, 64), dtype=bool)
+    b = a.copy()
+    b[0:32, 0:32] = True
+    recs = diff_tiles(a, b, tile=32)
+    assert len(recs) == 1 and recs[0].tile_index == 0
+
+
+def test_apply_record_reconstructs():
+    a = np.zeros((32, 32), dtype=bool)
+    b = a.copy()
+    b[4:9, 3:12] = True
+    recs = diff_tiles(a, b, tile=32)
+    canvas = a.copy()
+    for rec in recs:
+        apply_record(canvas, FrameRecord(1, 0, [rec]), tile=32)
+    assert np.array_equal(canvas, b)