vt1200/harness/udpstream/vidsrc.py

"""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
**LSB first** (bit 0 of a byte is the row's leftmost pixel -- the terminal's
scanout order, see :meth:`Rows.pack_row`).  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] **only** (ruling R14): the body bytes are
covered by each datagram's Ethernet FCS, and a payload-wide CRC is far too slow
on the terminal's CPU.  ``pack_header`` is still called with
``payload=rows_1bpp`` as an advisory check that the body is exactly
``payload_len`` bytes -- it does not change the CRC.

Pure standard library; no dependency on any host-side capture stack.
"""

from harness.udpstream.frame import HEADER_SIZE, KIND_STRIP, pack_header

__all__ = [
    "CHUNK", "MAX_ROWS", "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

#: Datagrams in the terminal's RX ring (``RX_DESC`` in vidstream.c).  A frame
#: that needs more cannot be received at all: the ring runs out of descriptors
#: and the payload drops it.
RING_DATAGRAMS = 16

#: Most source rows one frame can carry, from the ring rather than the header
#: field: the first datagram holds the 16-byte header plus a body, the other 15
#: carry body only, so the ring holds
#: ``(RING_DATAGRAMS-1)*CHUNK + (CHUNK-HEADER_SIZE)`` body bytes = 144 rows of
#: 160.  The 24 KB slot alone would allow 153, and the u8 ``rows`` field 255 --
#: neither is reachable, because the ring is what binds.
MAX_ROWS = (((RING_DATAGRAMS - 1) * CHUNK + (CHUNK - HEADER_SIZE))
            // ROW_BYTES)


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 ``i % 8`` of byte ``i // 8`` -- **LSB first**,
        which is how the terminal's scanout reads the framebuffer: ``dwt/
        screenshot.py`` decodes live VRAM with exactly this rule, and MAME's
        renderer does the same (``(vram[bit >> 4] >> (bit & 15)) & 1``).

        The earlier MSB-first convention here was self-consistent -- the encoder
        and ``tile_check.py`` shared it, so every replay check passed -- but it
        mirrors every eight pixels on a real display.  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.
        # Pixel i must land on bit i % 8 of byte i // 8, so reverse each
        # eight-pixel group before the parse: the first character of a group
        # becomes that byte's MSB, and the last becomes its LSB.
        bits = "".join("1" if value else "0" for value in pixels)
        lsb = "".join(bits[i:i + 8][::-1] for i in range(0, ROW_PIXELS, 8))
        return int(lsb, 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.
    ``rows`` is capped by the terminal's ring, not by its one-byte header field:
    :data:`MAX_ROWS` (144) rows already fill all 16 datagrams, and a frame that
    needs a 17th is dropped by the receiver.
    """

    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 <= MAX_ROWS:
            raise ValueError("rows out of range 1..%d: %r (the terminal's ring "
                             "is %d datagrams of %d bytes, so a frame can carry "
                             "at most %d rows of %d bytes; the header's one-byte "
                             "field allows 255 but the ring does not)"
                             % (MAX_ROWS, rows, RING_DATAGRAMS, CHUNK,
                                MAX_ROWS, ROW_BYTES))
        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)]