feat(stream): 1-bpp row packing and strip frame encoder

harness/udpstream/tests/test_vidsrc.py

@@ -0,0 +1,138 @@
+"""Tests for the strip encoder: 1-bpp row packing and frame chunking.
+
+The plan's three tests come first, verbatim.  The extras pin down what the plan
+text leaves implicit but the terminal depends on: the frame is split at 1450
+bytes with only the last packet short, and a payload whose length disagrees with
+``rows`` is rejected here instead of on the wire.
+
+Note the third plan test checks ``crc_ok`` *with the payload*: the header CRC
+covers header[0:12] + payload, so a header-only seal would fail every frame on
+the receiving side.
+"""
+import pytest
+
+from harness.udpstream.frame import HEADER_SIZE, KIND_STRIP, unpack_header
+from harness.udpstream.vidsrc import Rows, StripSource
+
+#: Largest UDP payload the transport promises per packet.
+CHUNK = 1450
+
+
+def test_pack_row_is_msb_first_and_160_bytes():
+    px = [0] * 1280
+    px[0] = 1
+    px[7] = 1
+    px[8] = 1
+    b = Rows.pack_row(px)
+    assert len(b) == 160
+    # Pixel 0 -> bit 7, pixel 7 -> bit 0, pixel 8 -> bit 7 of byte 1.
+    # NOTE: the plan/brief write this as 0b11000001, which is 0xC1 and encodes
+    # pixels 0, 1 and 7 -- three pixels from a two-pixel segment of the row, so
+    # no packing rule can produce it from px[0] and px[7].  MSB-first, as the
+    # same brief's prose states and as the terminal's bit-addressed row needs,
+    # is 0b10000001.  See the task-2 report.
+    assert b[0] == 0b10000001 and b[1] == 0b10000000
+
+
+def test_strip_source_rejects_stretch_that_does_not_divide_1024():
+    with pytest.raises(ValueError):
+        StripSource(rows=100, stretch=7)
+
+
+def test_strip_frame_round_trips_through_the_header():
+    src = StripSource(rows=8, stretch=128)
+    rows = bytes([0xAA]) * (8 * 160)
+    pkts = src.encode(frame_no=3, rows_1bpp=rows)
+    head = unpack_header(pkts[0][:16])
+    assert head["kind"] == 1 and head["rows"] == 8 and head["stretch"] == 128
+    assert head["payload_len"] == 8 * 160
+    assert b"".join(pkts)[16:] == rows
+    assert unpack_header(pkts[0][:16], b"".join(pkts)[16:])["crc_ok"] is True
+
+
+# --- packing -------------------------------------------------------------
+
+def test_pack_row_puts_every_pixel_where_the_terminal_reads_it():
+    """Pixel i is bit (7 - i % 8) of byte i // 8, i.e. MSB first, and no bit leaks."""
+    px = [1 if i % 3 == 0 else 0 for i in range(1280)]
+    b = Rows.pack_row(px)
+    assert len(b) == 160
+    for i, v in enumerate(px):
+        assert (b[i // 8] >> (7 - i % 8)) & 1 == v, i
+
+
+def test_pack_row_rejects_a_row_that_is_not_1280_pixels():
+    with pytest.raises(ValueError):
+        Rows.pack_row([1] * 1279)
+    with pytest.raises(ValueError):
+        Rows.pack_row([1] * 1281)
+
+
+# --- framing -------------------------------------------------------------
+
+def test_encode_splits_the_frame_at_1450_bytes_and_only_the_last_is_short():
+    """A full-screen frame is 20,496 B including the header: 15 packets."""
+    src = StripSource(rows=128, stretch=8)
+    rows = bytes([0x5A]) * (128 * 160)
+    pkts = src.encode(frame_no=1, rows_1bpp=rows)
+
+    assert all(len(p) <= CHUNK for p in pkts)
+    assert all(len(p) == CHUNK for p in pkts[:-1])
+    assert len(pkts) == 15
+    assert sum(len(p) for p in pkts) == HEADER_SIZE + 128 * 160
+    assert b"".join(pkts)[HEADER_SIZE:] == rows
+
+
+def test_a_small_strip_is_a_single_packet():
+    src = StripSource(rows=8, stretch=128)
+    rows = bytes([0x11]) * (8 * 160)
+    pkts = src.encode(frame_no=2, rows_1bpp=rows)
+    assert len(pkts) == 1
+    assert len(pkts[0]) == HEADER_SIZE + 8 * 160
+    assert unpack_header(pkts[0][:HEADER_SIZE], pkts[0][HEADER_SIZE:])["crc_ok"] is True
+
+
+def test_every_packet_carries_the_same_valid_header():
+    """The header travels once, in the first packet, and the body follows it."""
+    src = StripSource(rows=8, stretch=128)
+    rows = bytes([0x0F]) * (8 * 160)
+    pkts = src.encode(frame_no=0x1234, rows_1bpp=rows)
+
+    head = unpack_header(pkts[0][:HEADER_SIZE])
+    assert head["magic"] == b"VS"
+    assert head["kind"] == KIND_STRIP
+    assert head["frame_no"] == 0x1234
+    assert head["payload_len"] == 8 * 160
+    assert head["rows"] == 8 and head["stretch"] == 128
+    assert unpack_header(pkts[0][:HEADER_SIZE], b"".join(pkts)[HEADER_SIZE:])["crc_ok"] is True
+    # A header-only check must not validate: that would mean the CRC missed the body.
+    assert head["crc_ok"] is False
+
+
+def test_encode_rejects_a_payload_that_is_not_rows_times_160():
+    src = StripSource(rows=8, stretch=128)
+    with pytest.raises(ValueError):
+        src.encode(frame_no=0, rows_1bpp=bytes(8 * 160 - 1))
+    with pytest.raises(ValueError):
+        src.encode(frame_no=0, rows_1bpp=bytes(8 * 160 + 1))
+
+
+def test_encode_rejects_a_frame_number_that_does_not_fit_u16():
+    src = StripSource(rows=8, stretch=128)
+    with pytest.raises(ValueError):
+        src.encode(frame_no=0x10000, rows_1bpp=bytes(8 * 160))
+
+
+# --- geometry validation -------------------------------------------------
+
+@pytest.mark.parametrize("stretch", [0, 7, 1023, 1025, 2048])
+def test_strip_source_rejects_impossible_stretch(stretch):
+    with pytest.raises(ValueError):
+        StripSource(rows=8, stretch=stretch)
+
+
+def test_strip_source_rejects_rows_that_do_not_fit_the_header():
+    with pytest.raises(ValueError):
+        StripSource(rows=0, stretch=128)
+    with pytest.raises(ValueError):
+        StripSource(rows=256, stretch=4)

harness/udpstream/vidsrc.py

@@ -0,0 +1,101 @@
+"""Turn 1-bpp source rows into framed strip packets for the VT1200 stream.
+
+A *strip* frame is *N* unique rows of 1280 pixels, each row 160 bytes packed MSB
+first.  The terminal blits row *n* to destination row ``n * stretch`` with the
+blitter's vertical stretch, so each source row is reused ``stretch`` times and
+the whole screen is repainted from *N* = 1024 / stretch rows.  That is the whole
+scheme: 128 rows is a full-screen frame at 20,480 B of payload, 8 rows is a
+16th of the screen at 1,280 B.
+
+The wire form is the 16-byte header from :mod:`harness.udpstream.frame` followed
+by the packed rows, split into datagrams of at most :data:`CHUNK` bytes.  The
+header travels **once, in the first packet**, and the body continues immediately
+after it -- the terminal reassembles the datagrams in order and then checks the
+header against the whole payload, exactly as the C receiver does:
+
+    [ header | body ... ][ body ... ][ body ]
+
+The header CRC covers header[0:12] **plus the payload**, so ``pack_header`` is
+called with ``payload=rows_1bpp``.  A header-only CRC would satisfy every local
+test that only unpacks the header and would still be rejected by the terminal
+for every single frame.
+
+Pure standard library; no dependency on any host-side capture stack.
+"""
+
+from harness.udpstream.frame import KIND_STRIP, pack_header
+
+__all__ = [
+    "CHUNK", "ROW_BYTES", "ROW_PIXELS", "SCREEN_ROWS", "Rows", "StripSource",
+]
+
+#: Width of the terminal's line buffer, in pixels, and the row it packs to.
+ROW_PIXELS = 1280
+ROW_BYTES = ROW_PIXELS // 8                     # 160
+
+#: Scanlines the blitter paints; ``stretch`` divides this.
+SCREEN_ROWS = 1024
+
+#: Largest datagram this transport sends, header included.
+CHUNK = 1450
+
+
+class Rows:
+    """Pack one 1-bpp source row into the bytes the blitter reads."""
+
+    @staticmethod
+    def pack_row(pixels: list[int]) -> bytes:
+        """Return the :data:`ROW_BYTES` packed bytes for ``pixels``.
+
+        Pixel *i* becomes bit ``7 - i % 8`` of byte ``i // 8`` -- MSB first,
+        which is the bit order of the terminal's line buffer.  A truthy pixel is
+        a set bit, so ``True``, ``1`` and any other nonzero value all mean ink.
+        """
+        if len(pixels) != ROW_PIXELS:
+            raise ValueError("a row is %d pixels, got %d"
+                             % (ROW_PIXELS, len(pixels)))
+        # "1000000..." -> the integer those bits denote, big-endian bytes.
+        # Equivalent to setting bit 7 - i % 8 of byte i // 8 for each pixel i,
+        # but the join+parse is C-speed per character and measured ~1.5x a
+        # per-pixel shift loop on this box (65-120 us vs 100-150 us per row).
+        bits = "".join("1" if value else "0" for value in pixels)
+        return int(bits, 2).to_bytes(ROW_BYTES, "big")
+
+
+class StripSource:
+    """Frame encoder for one strip geometry.
+
+    ``rows`` is how many unique source rows a frame carries; ``stretch`` is how
+    many destination scanlines each of them paints, so it must divide
+    :data:`SCREEN_ROWS` or the last row would stop short of the screen bottom.
+    Both must fit their one-byte header fields.
+    """
+
+    def __init__(self, rows: int, stretch: int):
+        if not isinstance(stretch, int) or not 1 <= stretch <= SCREEN_ROWS:
+            raise ValueError("stretch out of range 1..%d: %r"
+                             % (SCREEN_ROWS, stretch))
+        if SCREEN_ROWS % stretch:
+            raise ValueError("stretch %d does not divide %d; a row would paint "
+                             "a fractional number of scanlines"
+                             % (stretch, SCREEN_ROWS))
+        if not isinstance(rows, int) or not 1 <= rows <= 0xFF:
+            raise ValueError("rows out of range 1..255: %r" % (rows,))
+        self.rows = rows
+        self.stretch = stretch
+
+    def encode(self, frame_no: int, rows_1bpp: bytes) -> list[bytes]:
+        """Return one frame as packets: header plus body, chunked at CHUNK.
+
+        ``rows_1bpp`` must already be packed (``Rows.pack_row`` per row) and
+        exactly ``rows * 160`` bytes, or the header would describe a body the
+        terminal does not receive.
+        """
+        body = bytes(rows_1bpp)
+        expected = self.rows * ROW_BYTES
+        if len(body) != expected:
+            raise ValueError("%d rows are %d bytes, got %d"
+                             % (self.rows, expected, len(body)))
+        frame = pack_header(KIND_STRIP, frame_no, len(body), self.rows,
+                            self.stretch, payload=body) + body
+        return [frame[i:i + CHUNK] for i in range(0, len(frame), CHUNK)]