feat(tiles): chunked dictionary upload protocol

harness/udpstream/frame.py

@@ -8,7 +8,8 @@ byte, so the layout is spelled out here and must be mirrored by the payload:
     ------  ----  ------------------------------------------------------
          0     2  magic, ASCII 'V','S'
          2     1  version, 0x01
-         3     1  kind (KIND_STRIP = 1, KIND_STATUS = 5)
+         3     1  kind (see the KINDS table below; the payload's
+                    ``frame_parse.h`` mirrors the same numbers)
          4     2  frame_no, u16 little-endian
          6     2  payload_len, u16 little-endian
          8     1  rows
@@ -65,7 +66,8 @@ import struct
 
 __all__ = [
     "HEADER_SIZE", "MAGIC", "VERSION", "CRC_OFFSET",
-    "KIND_STRIP", "KIND_STATUS",
+    "KINDS", "KIND_STRIP", "KIND_TILES_FULL", "KIND_TILES_DELTA",
+    "KIND_DICT_CHUNK", "KIND_STATUS",
     "ETHERTYPE", "BROADCAST", "TERMINAL_MAC", "MIN_ETHERNET_PAYLOAD",
     "STATUS_SIZE",
     "crc16", "pack_header", "seal_header", "unpack_header",
@@ -84,11 +86,32 @@ VERSION = 0x01
 #: The CRC covers header bytes [0, CRC_OFFSET) and nothing else.
 CRC_OFFSET = 12
 
+#: Every frame kind this protocol defines, in one table.  The terminal
+#: payload's ``frame_parse.h`` mirrors these **numbers**; neither side may
+#: renumber one without the other.  The constants below are named views of
+#: this table, not a second definition of the wire format.
+KINDS = {
+    "STRIP": 1,         # packed 1-bpp source rows (the original stream)
+    "TILES_FULL": 2,    # a full 40x32 screen of dictionary indices
+    "TILES_DELTA": 3,   # only the tiles that changed
+    "DICT_CHUNK": 4,    # a slice of the 32x32 tile dictionary
+    "STATUS": 5,        # the terminal's report, travelling the other way
+}
+
 #: A strip of packed 1-bpp source rows.
-KIND_STRIP = 1
+KIND_STRIP = KINDS["STRIP"]
+
+#: A full-screen frame of dictionary indices.
+KIND_TILES_FULL = KINDS["TILES_FULL"]
+
+#: A frame carrying only changed tiles.
+KIND_TILES_DELTA = KINDS["TILES_DELTA"]
+
+#: One slice of the dictionary upload (see :mod:`harness.udpstream.tiledict`).
+KIND_DICT_CHUNK = KINDS["DICT_CHUNK"]
 
 #: A status report travelling the other way.
-KIND_STATUS = 5
+KIND_STATUS = KINDS["STATUS"]
 
 #: Ethernet type the stream's frames carry.  0x88B5 is the probe frame
 #: sendraw.py sends and 0x88B6 the command frame lance_cmd.c uses; the video

harness/udpstream/tests/test_tiledict.py

@@ -0,0 +1,103 @@
+"""Tests for the chunked dictionary upload protocol.
+
+The plan's test comes first, verbatim.  The extras pin down what the plan text
+leaves implicit but the terminal's receiver (Task 3) depends on:
+
+* the 32 KB dictionary splits into exactly 23 independently framed packets
+  (payload of at most 1450 bytes: a 2-byte offset plus at most 1448 bytes of
+  dictionary);
+* every packet is an ordinary ``KIND_DICT_CHUNK`` frame -- 16-byte header, the
+  header-only CRC, no geometry -- so a receiver can tell it from a strip or a
+  status frame and drop a corrupted header without touching the payload;
+* the offset is the dictionary position of the first body byte, so a gap or a
+  reordered packet is visible from the offsets alone;
+* ``offset_base`` shifts every offset, which is how a dictionary can be placed
+  at a non-zero position in the receiver's buffer.
+
+Reassembly is asserted byte for byte: concatenating the bodies in arrival order
+must give back the input, and the last offset plus the last body length must be
+the end of the dictionary.
+"""
+import pytest
+
+from harness.udpstream.frame import (HEADER_SIZE, KIND_DICT_CHUNK,
+                                     unpack_header)
+from harness.udpstream.tiledict import (MAX_PAYLOAD, dict_chunks, payload_body,
+                                        payload_offset)
+
+#: A whole dictionary: 256 entries of 128 bytes.
+DICTIONARY = bytes(range(256)) * 128
+
+
+def test_chunks_cover_the_dictionary_once_in_order():
+    d = bytes(range(256)) * 128          # 32 KB
+    pk = dict_chunks(d)
+    assert b"".join(payload_body(p) for p in pk) == d
+    offs = [payload_offset(p) for p in pk]
+    assert offs == sorted(offs) and offs[0] == 0
+    assert offs[-1] + len(payload_body(pk[-1])) == len(d)
+
+
+def test_dictionary_splits_into_23_framed_packets():
+    pk = dict_chunks(DICTIONARY)
+    assert len(pk) == 23
+    for packet in pk:
+        header = unpack_header(packet)
+        assert header["magic"] == b"VS"
+        assert header["kind"] == KIND_DICT_CHUNK
+        assert header["crc_ok"] is True
+        assert 0 < header["payload_len"] <= MAX_PAYLOAD
+        assert len(packet) == HEADER_SIZE + header["payload_len"]
+
+
+def test_offsets_name_each_body_position():
+    pk = dict_chunks(DICTIONARY)
+    position = 0
+    for packet in pk:
+        assert payload_offset(packet) == position
+        position += len(payload_body(packet))
+    assert position == len(DICTIONARY)
+
+
+def test_offset_base_shifts_every_offset():
+    d = bytes(3000)
+    base = 100 * 128                     # entry 100 of a dictionary
+    pk = dict_chunks(d, offset_base=base)
+    positions = [payload_offset(p) for p in pk]
+    assert positions[0] == base
+    assert b"".join(payload_body(p) for p in pk) == d
+    assert positions[-1] + len(payload_body(pk[-1])) == base + len(d)
+
+
+def test_body_never_crosses_the_1450_byte_payload_budget():
+    pk = dict_chunks(DICTIONARY)
+    assert max(unpack_header(p)["payload_len"] for p in pk) == MAX_PAYLOAD
+    assert all(len(payload_body(p)) <= MAX_PAYLOAD - 2 for p in pk)
+
+
+def test_dictionary_that_would_overflow_the_u16_offset_is_rejected():
+    with pytest.raises(ValueError):
+        dict_chunks(DICTIONARY, offset_base=0x10000 - len(DICTIONARY) + 1)
+    with pytest.raises(ValueError):
+        dict_chunks(DICTIONARY, offset_base=-1)
+
+
+def test_empty_dictionary_is_rejected():
+    with pytest.raises(ValueError):
+        dict_chunks(b"")
+
+
+def test_payload_helpers_reject_a_frame_of_another_kind():
+    from harness.udpstream.frame import KIND_STRIP, pack_header
+    strip = pack_header(KIND_STRIP, 1, 2, 0, 0, payload=b"\x00\x00")
+    with pytest.raises(ValueError):
+        payload_offset(strip)
+    with pytest.raises(ValueError):
+        payload_body(strip)
+
+
+def test_payload_helpers_reject_a_truncated_payload():
+    packet = bytearray(dict_chunks(bytes(100))[0])
+    header = unpack_header(packet)
+    with pytest.raises(ValueError):
+        payload_body(bytes(packet[:HEADER_SIZE + header["payload_len"] - 1]))

harness/udpstream/tiledict.py

@@ -0,0 +1,130 @@
+"""Chunked upload of the 32x32 tile dictionary.
+
+The dictionary is 256 entries of 128 bytes -- 32 KB -- and it does not fit in
+one datagram.  :func:`dict_chunks` splits it into independently framed
+``KIND_DICT_CHUNK`` packets, each carrying at most :data:`MAX_PAYLOAD` payload
+bytes: a 2-byte little-endian **offset** followed by a slice of the dictionary.
+
+    [ 16-byte header ][ offset u16 ][ dictionary bytes ... ]
+
+The returned packets are frame buffers (header plus payload).  Wrapping one in
+an Ethernet frame is the caller's job, exactly as for
+:meth:`harness.udpstream.vidsrc.StripSource.encode`, and the helpers below take
+that frame buffer, not the padded Ethernet envelope.
+
+The offset is the dictionary position of the first body byte, so a receiver
+can detect a gap, a duplicate or a reordered packet without keeping a stream
+state machine (Task 3 requires ascending offsets with no gaps).  A 32 KB
+dictionary needs 23 packets: 22 of 1,448 body bytes (31,856 bytes) and a
+912-byte tail.
+
+Every packet has its own frame header and therefore its own header CRC; the
+CRC covers header bytes 0..11 only (ruling R14), exactly as for a strip frame.
+There is no body checksum -- the Ethernet FCS of each datagram is the check for
+the payload bytes.
+
+``rows`` and ``stretch`` are zero because a dictionary chunk has no geometry,
+and ``frame_no`` is zero because an upload is not one of the video's numbered
+frames.  The ``frame_no`` field is not what orders the chunks; the offset is.
+
+The payload's ``frame_parse.h`` mirrors :data:`harness.udpstream.frame.KINDS`,
+which is where ``KIND_DICT_CHUNK`` is defined.
+
+Pure standard library.
+"""
+
+import struct
+
+from harness.udpstream.frame import (HEADER_SIZE, KIND_DICT_CHUNK, MAGIC,
+                                     VERSION, pack_header, unpack_header)
+
+__all__ = [
+    "MAX_BODY", "MAX_DICT_BYTES", "MAX_PAYLOAD", "OFFSET_SIZE",
+    "dict_chunks", "payload_body", "payload_offset",
+]
+
+#: Largest chunk payload in one packet, offset included.  The same 1450-byte
+#: budget as a strip datagram's chunk (:data:`harness.udpstream.vidsrc.CHUNK`).
+MAX_PAYLOAD = 1450
+
+#: The offset field: a u16 little-endian dictionary position.
+OFFSET_SIZE = 2
+
+#: Most dictionary bytes one packet can carry after its offset.
+MAX_BODY = MAX_PAYLOAD - OFFSET_SIZE                 # 1,448
+
+#: A u16 offset can name positions 0..65,535, so a dictionary ending at
+#: offset 65,536 is still addressable: its last chunk's offset fits.
+MAX_DICT_BYTES = 0x10000
+
+_OFFSET = struct.Struct("<H")
+
+
+def _dict_payload(packet: bytes) -> bytes:
+    """Return a dictionary chunk's payload, or raise ValueError.
+
+    Validates what the helpers rely on: a whole ``VS`` frame of the right
+    version, the dictionary-chunk kind, a length that matches the packet, and
+    room for the offset.  The header CRC is *not* rejected here -- a caller
+    that wants to count bad headers can read ``unpack_header`` itself -- but a
+    truncated or foreign packet is an error rather than a silent misread.
+    """
+    try:
+        header = unpack_header(packet)
+    except ValueError as exc:
+        raise ValueError("not a dictionary chunk: %s" % exc)
+    if header["magic"] != MAGIC or header["version"] != VERSION:
+        raise ValueError("not a VS frame (magic %r, version %d)"
+                         % (header["magic"], header["version"]))
+    if header["kind"] != KIND_DICT_CHUNK:
+        raise ValueError("kind %d is not a dictionary chunk (%d)"
+                         % (header["kind"], KIND_DICT_CHUNK))
+    if len(packet) != HEADER_SIZE + header["payload_len"]:
+        raise ValueError("packet is %d bytes but the header describes %d"
+                         % (len(packet), HEADER_SIZE + header["payload_len"]))
+    if header["payload_len"] < OFFSET_SIZE:
+        raise ValueError("a dictionary chunk payload is at least %d bytes, "
+                         "got %d" % (OFFSET_SIZE, header["payload_len"]))
+    return packet[HEADER_SIZE:HEADER_SIZE + header["payload_len"]]
+
+
+def dict_chunks(dictionary: bytes, offset_base: int = 0) -> list[bytes]:
+    """Split ``dictionary`` into framed ``KIND_DICT_CHUNK`` packets.
+
+    ``offset_base`` is added to every offset, for placing the upload at a
+    non-zero position in the receiver's buffer.  The dictionary must be
+    non-empty and must end within the u16 offset space; those are errors here
+    rather than a wrap on the wire.
+    """
+    if not isinstance(dictionary, (bytes, bytearray, memoryview)):
+        raise TypeError("a dictionary is bytes-like, got %s"
+                        % type(dictionary).__name__)
+    data = bytes(dictionary)
+    if not data:
+        raise ValueError("a dictionary upload needs at least one byte")
+    if (not isinstance(offset_base, int) or isinstance(offset_base, bool)
+            or not 0 <= offset_base <= 0xFFFF):
+        raise ValueError("offset_base is 0..65535, got %r" % (offset_base,))
+    end = offset_base + len(data)
+    if end > MAX_DICT_BYTES:
+        raise ValueError("a %d-byte dictionary at offset %d ends at %d, past "
+                         "the u16 offset space (%d)"
+                         % (len(data), offset_base, end, MAX_DICT_BYTES))
+
+    packets = []
+    for start in range(0, len(data), MAX_BODY):
+        body = data[start:start + MAX_BODY]
+        payload = _OFFSET.pack(offset_base + start) + body
+        packets.append(pack_header(KIND_DICT_CHUNK, 0, len(payload), 0, 0,
+                                   payload=payload) + payload)
+    return packets
+
+
+def payload_offset(packet: bytes) -> int:
+    """Dictionary position of the first body byte in ``packet``."""
+    return _OFFSET.unpack_from(_dict_payload(packet), 0)[0]
+
+
+def payload_body(packet: bytes) -> bytes:
+    """The dictionary bytes ``packet`` carries, offset excluded."""
+    return _dict_payload(packet)[OFFSET_SIZE:]