feat(tiles): deterministic dictionary builder and quantiser

harness/udpstream/tests/test_tilecost.py

@@ -0,0 +1,184 @@
+"""Tests for the tile dictionary builder and the frame quantiser.
+
+The plan's two tests come first, verbatim.  The extras pin down what the plan
+text leaves implicit but the later tasks depend on:
+
+* the tile layout -- 32 rows of four bytes, row-major, MSB first -- so a
+  dictionary entry is exactly the bytes the terminal's ``PIXBLT B,XY`` reads
+  from ``dict + index*128``;
+* the index frame's size.  The plan says ``quantise`` returns 1,280 index bytes
+  (a 40x32 screen), but its own test feeds a 128-row frame, which is only 160
+  tiles.  A buffer shorter than the screen is therefore the *top* of the screen
+  and the rows it does not cover quantise the all-zero tile; that convention is
+  pinned here rather than left to the caller to discover;
+* the Hamming tie-break: equal distance resolves to the lowest index;
+* validation, so a mis-sized frame is an error here and not a mis-painted
+  screen later.
+"""
+import random
+
+import pytest
+
+from harness.udpstream.tilecost import build_dictionary, quantise
+
+#: One source row of the 1280-px screen, as ``vidsrc.Rows.pack_row`` packs it.
+ROW_BYTES = 160
+
+#: A 32-px tile is four bytes wide, so a screen row is 40 tiles.
+COLS = 40
+
+#: Dictionary entry size: 32 rows of four bytes.
+TILE_BYTES = 128
+
+
+def _frame(seed):
+    r = random.Random(seed)
+    return bytes(r.randrange(256) for _ in range(128 * 1280 // 8))
+
+
+def test_dictionary_is_deterministic_and_indexable():
+    frames = [_frame(i) for i in range(4)]
+    a = build_dictionary(frames, k=8, seed=7)
+    b = build_dictionary(frames, k=8, seed=7)
+    assert a == b and len(a) == 8 * 128
+
+
+def test_quantise_returns_1280_indices_within_range():
+    frames = [_frame(i) for i in range(4)]
+    d = build_dictionary(frames, k=8, seed=7)
+    idx = quantise(frames[0], d)
+    assert len(idx) == 1280 and max(idx) < 8
+
+
+# --- layout --------------------------------------------------------------
+
+def _screen_of(patterns, seed=0):
+    """Build a 1280x1024 frame whose 32x32 tiles are drawn from ``patterns``."""
+    r = random.Random(seed)
+    screen = bytearray(ROW_BYTES * 1024)
+    for ty in range(32):
+        for tx in range(COLS):
+            tile = patterns[r.randrange(len(patterns))]
+            for row in range(32):
+                off = (ty * 32 + row) * ROW_BYTES + tx * 4
+                screen[off:off + 4] = tile[row * 4:row * 4 + 4]
+    return bytes(screen)
+
+
+def _render(indices, dictionary):
+    """Blit an index frame the way the terminal will: one block per index."""
+    screen = bytearray(ROW_BYTES * 1024)
+    for ty in range(32):
+        for tx in range(COLS):
+            base = indices[ty * COLS + tx] * TILE_BYTES
+            for row in range(32):
+                off = (ty * 32 + row) * ROW_BYTES + tx * 4
+                screen[off:off + 4] = dictionary[base + row * 4:
+                                                 base + row * 4 + 4]
+    return bytes(screen)
+
+
+def test_quantise_picks_the_byte_block_the_blitter_reads():
+    """Tile (tx, ty) is rows ty*32..+31, bytes tx*4..tx*4+4 of each row."""
+    frame = bytearray(ROW_BYTES * 32)           # one tile row: 40 tiles
+    tiles = []
+    for tx in range(COLS):
+        tile = bytearray()
+        for row in range(32):
+            block = bytes((tx * 7 + row * 3 + c) & 0xFF for c in range(4))
+            frame[row * ROW_BYTES + tx * 4:row * ROW_BYTES + tx * 4 + 4] = block
+            tile += block
+        tiles.append(bytes(tile))
+
+    idx = quantise(bytes(frame), b"".join(tiles))
+    # Every tile has distance 0 to its own dictionary entry, so the first row of
+    # the index frame names them in order -- and nothing else could.
+    assert idx[:COLS] == bytes(range(COLS))
+
+
+def test_rows_below_a_short_frame_quantise_the_zero_tile():
+    """A 32-row frame is the top of the screen; everything below is black."""
+    white = bytes([0xFF]) * TILE_BYTES
+    black = bytes(TILE_BYTES)
+    dictionary = white + black                  # index 0 white, 1 black
+    frame = bytes([0xFF]) * (ROW_BYTES * 32)    # one all-white tile row
+
+    idx = quantise(frame, dictionary)
+    assert len(idx) == 1280
+    assert idx[:COLS] == bytes([0]) * COLS
+    assert idx[COLS:] == bytes([1]) * (1280 - COLS)
+
+
+def test_equal_distance_resolves_to_the_lowest_index():
+    white = bytes([0xFF]) * TILE_BYTES
+    black = bytes(TILE_BYTES)
+    dictionary = white + black + black          # entries 1 and 2 are identical
+    frame = bytes(ROW_BYTES * 32)               # all black
+
+    idx = quantise(frame, dictionary)
+    assert set(idx) == {1}
+
+    # The same tie with the duplicate *first*: an exact match at index 0 must
+    # stop the search, not be overtaken by a later exact match.
+    dictionary = black + white + black
+    idx = quantise(frame, dictionary)
+    assert set(idx) == {0}
+
+
+def test_a_dictionary_of_the_frames_own_tiles_reconstructs_the_frame():
+    rng = random.Random(3)
+    patterns = [bytes(rng.randrange(256) for _ in range(TILE_BYTES))
+                for _ in range(5)]
+    frames = [_screen_of(patterns, seed=s) for s in range(3)]
+
+    d = build_dictionary(frames, k=5, seed=1)
+    assert len(d) == 5 * TILE_BYTES
+    for frame in frames:
+        assert _render(quantise(frame, d), d) == frame
+
+
+def test_build_dictionary_does_not_depend_on_frame_order():
+    """Candidates are named by content, so frame order cannot matter."""
+    frames = [_frame(i) for i in range(4)]
+    assert (build_dictionary(frames, k=4, seed=2)
+            == build_dictionary(list(reversed(frames)), k=4, seed=2))
+
+
+def test_build_dictionary_of_a_single_repeated_tile_is_that_tile():
+    """The centroid is data, not a seeded sample: one distinct tile stays."""
+    pattern = bytes((i * 37 + 11) & 0xFF for i in range(TILE_BYTES))
+    frame = _screen_of([pattern], seed=0)
+
+    assert build_dictionary([frame], k=1, seed=0) == pattern
+
+
+# --- validation ----------------------------------------------------------
+
+@pytest.mark.parametrize("k", [0, -1, 257, 1000])
+def test_build_dictionary_rejects_k_that_is_not_an_index_byte(k):
+    with pytest.raises(ValueError):
+        build_dictionary([_frame(0)], k=k)
+
+
+def test_build_dictionary_rejects_no_frames():
+    with pytest.raises(ValueError):
+        build_dictionary([], k=8)
+
+
+def test_frames_must_be_whole_160_byte_rows_of_whole_32_row_tiles():
+    d = bytes(TILE_BYTES)
+    with pytest.raises(ValueError):
+        quantise(bytes(ROW_BYTES * 32 + 1), d)   # half a row
+    with pytest.raises(ValueError):
+        quantise(bytes(ROW_BYTES * 33), d)       # half a tile row
+    with pytest.raises(ValueError):
+        quantise(bytes(ROW_BYTES * 1056), d)     # below the 1024-row screen
+    with pytest.raises(ValueError):
+        build_dictionary([bytes(ROW_BYTES * 31)], k=1)
+
+
+def test_the_dictionary_is_a_whole_number_of_tiles_of_at_most_256():
+    frame = bytes(ROW_BYTES * 32)
+    for bad in (b"", bytes(TILE_BYTES + 1), bytes(257 * TILE_BYTES)):
+        with pytest.raises(ValueError):
+            quantise(frame, bad)

harness/udpstream/tilecost.py

@@ -0,0 +1,302 @@
+"""Build a 32x32 1-bpp tile dictionary and quantise frames into tile indices.
+
+Scheme C replaces row strips with a fixed dictionary of ``k`` 32x32 tiles plus
+one index byte per tile, so a full screen is 1,280 index bytes on the wire
+instead of 20,480 bytes of rows.  This module is the offline half: it builds the
+dictionary from the clip's frames and turns a frame into indices.  The
+terminal's half is one ``PIXBLT B,XY`` per tile from ``dict + index*128`` with
+``SPTCH=32`` and ``DYDX=(32,32)``; see the design's §8.3.
+
+Geometry
+--------
+
+The screen is 1280x1024, one bit per pixel.  A source row is therefore 160 bytes
+(:data:`harness.udpstream.vidsrc.ROW_BYTES`) and a 32x32 tile is 32 rows of
+**four** bytes -- 32 px is two 16-bit words -- for 128 bytes: row-major, two
+bytes per word, MSB first within each byte, the same bit order
+:meth:`harness.udpstream.vidsrc.Rows.pack_row` produces.  The grid is 40 columns
+by 32 rows: 1,280 tiles.
+
+A frame buffer is whole 160-byte rows and whole 32-row tile bands.  A buffer
+shorter than the 1024-row screen is the **top** of the screen: :func:`quantise`
+still returns the full 40x32 index frame the terminal's ``TILES_FULL`` expects,
+and the rows the buffer does not cover are quantised from the all-zero (black)
+tile.  That is the only reading that satisfies both the wire contract (a full
+frame is always 1,280 indices) and the plan's 128-row test frame, which covers
+160 of the 1,280 tiles; it is pinned by a test rather than left implicit.
+
+The dictionary
+--------------
+
+:func:`build_dictionary` runs Lloyd's algorithm over every tile of every
+supplied frame with Hamming distance as the metric -- the right metric here,
+because a displaced wire bit is one wrong pixel.  It is deterministic end to
+end:
+
+* only **distinct** tiles are candidates (repeats add nothing to a vote), and
+  candidates are named by content, so the result does not depend on the order
+  the frames are supplied in;
+* the initial centroids are the ``k`` candidates whose ``blake2b(seed, tile)``
+  digest sorts lowest.  That is a seeded choice, but unlike
+  :meth:`random.Random.sample` it is stable across Python versions, and the
+  dictionary is meant to be committed once and reproduced;
+* a tile joins the nearest centroid, ties to the **lowest** index;
+* a centroid is the per-pixel majority of its members, which is the Hamming
+  centroid that minimises total distance.  A pixel with exactly half its members
+  set stays 0: ties are black;
+* an empty cluster keeps its previous centroid;
+* iteration stops when the centroids stop moving, and in any case after
+  :data:`ITERATIONS` (8) passes;
+* the centroids are finally sorted as byte strings, so the dictionary's *order*
+  is canonical as well as its contents.
+
+With fewer distinct candidate tiles than ``k`` the remaining entries are the
+all-zero tile: unused slots are black, never invented.  A short frame is not
+padded here -- padding the training set with the 896 unseen black tile rows
+would outvote the picture -- so build the dictionary from frames that cover
+what the clip actually shows (the full-screen frames it will send).
+
+Cost: one quantise is ``tiles x k`` Hamming distances (327,680 for a full screen
+at ``k=256``), but repeated tiles share one answer, which is what makes a frame
+of flat panels cheap.  The dictionary is built offline once per clip.
+"""
+
+import hashlib
+
+from harness.udpstream.vidsrc import ROW_BYTES, SCREEN_ROWS
+
+__all__ = [
+    "ITERATIONS", "MAX_TILES", "SCREEN_TILE_ROWS", "TILES_PER_SCREEN",
+    "TILE_BYTES", "TILE_COLS", "TILE_PIXELS", "TILE_ROW_BYTES",
+    "build_dictionary", "quantise",
+]
+
+#: Side of a tile, in pixels (and rows).
+TILE_PIXELS = 32
+
+#: Bytes per tile row: 32 px packed one bit per pixel.
+TILE_ROW_BYTES = TILE_PIXELS // 8                       # 4
+
+#: A whole tile: 32 rows of four bytes, row-major.
+TILE_BYTES = TILE_PIXELS * TILE_ROW_BYTES               # 128
+
+#: Tiles across one 160-byte source row.
+TILE_COLS = ROW_BYTES // TILE_ROW_BYTES                 # 40
+
+#: Tile bands down the screen.
+SCREEN_TILE_ROWS = SCREEN_ROWS // TILE_PIXELS           # 32
+
+#: Indices in one full-screen ``TILES_FULL`` frame.
+TILES_PER_SCREEN = TILE_COLS * SCREEN_TILE_ROWS         # 1,280
+
+#: Most dictionary entries: an index travels as one byte.
+MAX_TILES = 256
+
+#: Most Lloyd passes, whether or not the centroids have settled.
+ITERATIONS = 8
+
+_TILE_BITS = TILE_BYTES * 8
+_ALL = (1 << _TILE_BITS) - 1
+
+
+def _tile_rows(frame):
+    """Return how many 32-row tile bands ``frame`` holds, or raise ValueError.
+
+    The buffer must be whole 160-byte rows and whole 32-row bands, and at most
+    the 1024-row screen: a partial row or band has no tile to name, and rows
+    below the screen would address memory the terminal does not have.
+    """
+    try:
+        length = len(frame)
+    except TypeError:
+        raise TypeError("a frame is bytes-like, got %s"
+                        % type(frame).__name__)
+    if length % ROW_BYTES:
+        raise ValueError("a frame is whole %d-byte rows, got %d bytes"
+                         % (ROW_BYTES, length))
+    rows = length // ROW_BYTES
+    if rows % TILE_PIXELS:
+        raise ValueError("a frame is whole %d-row tiles, got %d rows"
+                         % (TILE_PIXELS, rows))
+    if not 1 <= rows <= SCREEN_ROWS:
+        raise ValueError("a frame is 1..%d rows of %d bytes, got %d"
+                         % (SCREEN_ROWS, ROW_BYTES, rows))
+    return rows // TILE_PIXELS
+
+
+def _frame_tiles(frame):
+    """Return ``frame``'s tiles as ints, row-major over the tile grid.
+
+    Tile (tx, ty) is rows ``ty*32``..``+31``, bytes ``tx*4``..``tx*4+4`` of each
+    row, concatenated in that order -- exactly the 128 bytes the blitter reads
+    from ``dict + index*128``.  An int is the cheapest Hamming carrier:
+    ``(a ^ b).bit_count()`` is one C call.
+    """
+    tiles = []
+    for ty in range(_tile_rows(frame)):
+        top = ty * TILE_PIXELS * ROW_BYTES
+        for tx in range(TILE_COLS):
+            left = tx * TILE_ROW_BYTES
+            tiles.append(int.from_bytes(
+                b"".join(frame[top + row * ROW_BYTES + left:
+                               top + row * ROW_BYTES + left + TILE_ROW_BYTES]
+                         for row in range(TILE_PIXELS)), "big"))
+    return tiles
+
+
+def _dictionary_entries(dictionary):
+    """Return the dictionary as ints, or raise ValueError on a bad length."""
+    try:
+        length = len(dictionary)
+    except TypeError:
+        raise TypeError("a dictionary is bytes-like, got %s"
+                        % type(dictionary).__name__)
+    if length % TILE_BYTES:
+        raise ValueError("a dictionary is whole %d-byte tiles, got %d bytes"
+                         % (TILE_BYTES, length))
+    count = length // TILE_BYTES
+    if not 1 <= count <= MAX_TILES:
+        raise ValueError("a dictionary is 1..%d tiles of %d bytes (an index is "
+                         "one byte), got %d"
+                         % (MAX_TILES, TILE_BYTES, count))
+    return [int.from_bytes(dictionary[i * TILE_BYTES:(i + 1) * TILE_BYTES],
+                           "big")
+            for i in range(count)]
+
+
+def _nearest(tile, entries):
+    """Index of the entry with the smallest Hamming distance; ties lower."""
+    best = 0
+    best_distance = (tile ^ entries[0]).bit_count()
+    if best_distance == 0:
+        return 0
+    for index in range(1, len(entries)):
+        distance = (tile ^ entries[index]).bit_count()
+        if distance == 0:
+            return index        # nothing can beat zero, and later ties lose
+        if distance < best_distance:
+            best, best_distance = index, distance
+    return best
+
+
+def _majority(tiles):
+    """Per-bit majority of ``tiles`` as one int; a tie (half set) stays 0.
+
+    Counts are accumulated bit-sliced: plane *j* holds bit *j* of the count at
+    every pixel position, so one addition is ``~log2(n)`` big-int operations and
+    the threshold is a single borrow-propagating subtraction.  Counting pixels
+    one at a time would be 1,024 Python operations per member.
+    """
+    planes = []
+    for tile in tiles:
+        carry = tile
+        shift = 0
+        while carry:
+            if shift == len(planes):
+                planes.append(0)
+            carry, planes[shift] = planes[shift] & carry, planes[shift] ^ carry
+            shift += 1
+
+    need = len(tiles) // 2 + 1          # strict majority
+    borrow = 0                          # borrow out of count - need
+    for shift in range(max(len(planes), need.bit_length())):
+        plane = planes[shift] if shift < len(planes) else 0
+        subtrahend = _ALL if (need >> shift) & 1 else 0
+        borrow = ((~plane & _ALL) & (subtrahend | borrow)) | (subtrahend
+                                                              & borrow)
+    # A pixel whose subtraction ended without a borrow counts at least ``need``.
+    return ~borrow & _ALL
+
+
+def _initial_centroids(candidates, k, seed):
+    """Choose ``k`` distinct candidates as a deterministic function of content.
+
+    With at most ``k`` candidates every one is kept (the rest of the dictionary
+    is padded black).  Otherwise the choice is the ``k`` lowest
+    ``blake2b(seed, tile)`` digests.  Seeded, but stable across Python versions
+    and independent of frame order, which matters because the dictionary is
+    committed output.  ``random.Random.sample`` guarantees neither.
+    """
+    if len(candidates) <= k:
+        return candidates + [0] * (k - len(candidates))
+    prefix = str(seed).encode("ascii")
+
+    def key(tile):
+        digest = hashlib.blake2b(prefix)
+        digest.update(tile.to_bytes(TILE_BYTES, "big"))
+        return digest.digest()
+
+    return sorted(candidates, key=key)[:k]
+
+
+def build_dictionary(frames, k=256, seed=0):
+    """Return ``k`` tiles of :data:`TILE_BYTES` bytes, deterministically ordered.
+
+    ``frames`` are packed 1-bpp buffers as :func:`quantise` takes them; every
+    tile of every frame feeds the k-means.  ``k`` defaults to 256, the most an
+    index byte can name.  ``seed`` chooses the initial centroids; the same
+    ``seed``, frames and ``k`` always give byte-identical output.
+    """
+    if not isinstance(k, int) or isinstance(k, bool) or not 1 <= k <= MAX_TILES:
+        raise ValueError("k is 1..%d (an index is one byte), got %r"
+                         % (MAX_TILES, k))
+    frames = list(frames)
+    if not frames:
+        raise ValueError("build_dictionary needs at least one frame")
+
+    tiles = []
+    for frame in frames:
+        tiles.extend(_frame_tiles(frame))
+
+    seen = set()
+    candidates = []
+    for tile in tiles:
+        if tile not in seen:
+            seen.add(tile)
+            candidates.append(tile)
+
+    centroids = _initial_centroids(candidates, k, seed)
+    for _ in range(ITERATIONS):
+        buckets = [[] for _ in range(k)]
+        nearest = {}
+        for tile in tiles:
+            index = nearest.get(tile)
+            if index is None:
+                index = _nearest(tile, centroids)
+                nearest[tile] = index
+            buckets[index].append(tile)
+        updated = [centroids[i] if not bucket else _majority(bucket)
+                   for i, bucket in enumerate(buckets)]
+        if updated == centroids:
+            break
+        centroids = updated
+
+    centroids.sort()
+    return b"".join(tile.to_bytes(TILE_BYTES, "big") for tile in centroids)
+
+
+def quantise(frame, dictionary):
+    """Return the frame's :data:`TILES_PER_SCREEN` indices, row-major.
+
+    Each index is the dictionary entry with the smallest Hamming distance to
+    that 32x32 tile, ties to the lowest index, so the terminal can paint the
+    whole screen from the returned bytes alone.  A frame shorter than 1024 rows
+    is the top of the screen; the rest quantises the all-zero tile.
+    """
+    entries = _dictionary_entries(dictionary)
+    tiles = _frame_tiles(frame)
+
+    nearest = {}
+    indices = bytearray(TILES_PER_SCREEN)
+    for position, tile in enumerate(tiles):
+        index = nearest.get(tile)
+        if index is None:
+            index = _nearest(tile, entries)
+            nearest[tile] = index
+        indices[position] = index
+
+    black = nearest.get(0)
+    if black is None:
+        black = _nearest(0, entries)
+    indices[len(tiles):] = bytes([black]) * (TILES_PER_SCREEN - len(tiles))
+    return bytes(indices)