"""Binary frame header and CRC for the VT1200 video stream. This module is the single definition of the wire format. The host sender (Python) and the terminal payload (C, ``vidstream.c``) must agree byte for byte, so the layout is spelled out here and must be mirrored by the payload: offset size field ------ ---- ------------------------------------------------------ 0 2 magic, ASCII 'V','S' 2 1 version, 0x01 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 9 1 stretch 10 2 reserved, zero 12 2 crc16, u16 little-endian 14 2 padding to HEADER_SIZE, zero The CRC is CCITT-FALSE -- polynomial 0x1021, initial value 0xFFFF, no input or output reflection, no final XOR -- and it covers **header bytes 0..11 only**, never the payload and never the CRC field itself (ruling R14). That is a deliberate narrowing of an earlier contract. A payload-wide CRC cost the terminal ~139 instructions per payload byte; on the 34010 that made the parser, not the wire or the blitter, the frame-rate ceiling. Nothing is lost by dropping it: every datagram already carries an Ethernet FCS that the LANCE validates before the bytes reach a slot, and ``payload_len`` plus the delivered length catches truncation. What the header CRC still buys is a check that the geometry a frame was *sent* with is the geometry the terminal acts on. So a corrupted **payload** byte does not invalidate a frame here -- that is the chip's job -- while a corrupted **header** byte does. The status body --------------- A ``KIND_STATUS`` frame is the terminal's report to the host, and it is an ordinary frame: the same 16-byte header, the same header-only CRC, with ``rows`` and ``stretch`` zero (a status has no geometry). Its body is 16 bytes: offset size field ------ ---- ------------------------------------------------------ 0 4 frames_rx, u32 little-endian (frames fp_parse accepted) 4 2 crc_err, u16 little-endian (frames it rejected: header CRC, magic, version or length -- since R14 a *body* error is the LANCE's FCS and can never appear here) 6 2 drops, u16 little-endian (frames that arrived and did not reach the screen: CSR0 MISS report windows -- the bit is latched, so two misses in one window count once -- plus frames too large for the ring, plus datagrams the chip flagged bad, which carry no bytes at all) 8 2 csr0, u16 little-endian (the chip's status word when the report was built) 10 2 heap_free_kb, u16 little-endian (the terminal's free heap) 12 1 dict_state, u8: bit 0 ``DICT_READY`` (the 32 KB dictionary is complete), bit 1 ``DICT_BAD`` (at least one chunk was refused). Both can be set: a chunk refused after the dictionary became ready leaves it usable. 13 1 reserved, zero 14 2 decode_worst, u16 little-endian (the worst per-frame decode time since the last status frame that was actually sent, in TMS34010 REFCNT ticks -- *not* microseconds; the unit conversion is still to be measured on the real machine) The out-of-cadence reports ---------------------------- A status is sent every 64 accepted frames, but two early reports exist so the host is never left with silence: * the **first** accepted frame -- it proves the payload is running and carries ``dict_state`` before the first 64 frames have arrived; * **immediately after a refused dictionary chunk** -- one lost chunk makes every later chunk fail and no frame is ever accepted, so the cadence would never fire and a host would report "payload not running" instead of "dictionary incomplete". ``decode_worst`` is reset only when a status frame has actually been handed to the chip, so a failed transmit does not lose the measurement. Both ends put that frame in an Ethernet frame: 6 bytes destination, 6 bytes source, the ethertype below, then the ``VS`` frame. The terminal does not filter on ethertype (its LANCE accepts broadcast and its own physical address), so the value is for the host's benefit: a status frame must not be mistaken for a probe (0x88B5, sendraw.py) or a command (0x88B6, lance_cmd.c). """ import struct __all__ = [ "HEADER_SIZE", "MAGIC", "VERSION", "CRC_OFFSET", "KINDS", "KIND_STRIP", "KIND_TILES_FULL", "KIND_TILES_DELTA", "KIND_DICT_CHUNK", "KIND_STATUS", "DICT_STATE_READY", "DICT_STATE_BAD", "ETHERTYPE", "BROADCAST", "TERMINAL_MAC", "MIN_ETHERNET_PAYLOAD", "STATUS_SIZE", "crc16", "pack_header", "seal_header", "unpack_header", "pack_status", "unpack_status", "ethernet_frame", ] #: Total size of the fixed header every frame starts with. HEADER_SIZE = 16 #: Magic bytes that open every frame: ASCII 'V','S'. MAGIC = b"VS" #: Wire format version. Bump when the layout changes. 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 = 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 = 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 #: stream gets its own so a captured frame cannot be mistaken for either. ETHERTYPE = 0x88B7 #: Destination that always passes the LANCE's address filter: with LADRF zero #: the chip accepts its own physical address and broadcast, nothing else. BROADCAST = b"\xff" * 6 #: The terminal's own physical address, from its Ethernet address ROM #: (bytes 0..5 of the eight at bit 0x6800000; the last two are the checksum). TERMINAL_MAC = b"\x08\x00\x2b\x16\x4b\x5a" #: Bytes of padding an Ethernet frame needs after its 14-byte header. MIN_ETHERNET_PAYLOAD = 46 #: Size of the status body (see the module docstring). STATUS_SIZE = 16 #: ``dict_state`` bits: the upload completed / a chunk was refused. DICT_STATE_READY = 0x01 DICT_STATE_BAD = 0x02 #: Everything before the CRC field, little-endian. _PREFIX = struct.Struct("<2sBBHHBBH") #: The CRC field and the two padding bytes behind it. _TAIL = struct.Struct(" int: """CCITT-FALSE CRC of ``data`` (poly 0x1021, init 0xFFFF, not reflected). No input or output reflection and no final XOR, so ``crc16(b"123456789")`` is the standard check value 0x29B1 and equals ``binascii.crc_hqx(data, 0xFFFF)`` byte for byte. """ crc = 0xFFFF for byte in data: crc = ((crc << 8) & 0xFFFF) ^ _CRC_TABLE[((crc >> 8) ^ byte) & 0xFF] return crc def _checked(name: str, value: int, limit: int) -> int: """Range-check a header field, so a bad value fails here and not on the wire.""" if not isinstance(value, int) or not 0 <= value <= limit: raise ValueError("%s out of range 0..%d: %r" % (name, limit, value)) return value def pack_header(kind: int, frame_no: int, payload_len: int, rows: int, stretch: int, payload: bytes | None = None) -> bytes: """Build the 16-byte header for a frame. The CRC covers header bytes 0..11 only (ruling R14), so the header can be built before the body exists. ``payload`` is kept as an **advisory** consistency check: when it is passed it must be exactly ``payload_len`` bytes, which catches a header that describes a body the caller does not have. Its *contents* are not covered -- the Ethernet FCS is. """ kind = _checked("kind", kind, 0xFF) frame_no = _checked("frame_no", frame_no, 0xFFFF) payload_len = _checked("payload_len", payload_len, 0xFFFF) rows = _checked("rows", rows, 0xFF) stretch = _checked("stretch", stretch, 0xFF) if payload is not None: body = bytes(payload) if len(body) != payload_len: raise ValueError("payload is %d bytes but payload_len says %d" % (len(body), payload_len)) prefix = _PREFIX.pack( MAGIC, VERSION, kind, frame_no, payload_len, rows, stretch, 0, # reserved ) assert len(prefix) == CRC_OFFSET crc = crc16(prefix) return prefix + _TAIL.pack(crc, 0) def seal_header(header: bytes, payload: bytes | None = None) -> bytes: """Return ``header`` with its CRC field recomputed over header bytes 0..11. ``header`` must be exactly :data:`HEADER_SIZE` bytes: a longer buffer is rejected rather than silently truncated, so a caller cannot pass ``header + payload`` here and believe the payload was sealed. ``payload`` is advisory, exactly as in :func:`pack_header`: when it is passed it must match the header's ``payload_len``, and its bytes are not covered by the CRC. Sealing is therefore idempotent and order-independent -- header first or body first, the bytes are the same. """ if len(header) != HEADER_SIZE: raise ValueError("header must be exactly %d bytes, got %d" % (HEADER_SIZE, len(header))) if payload is not None: want = struct.unpack_from(" dict: """Parse a frame header and check its CRC. ``crc_ok`` covers header bytes 0..11 only (ruling R14). ``payload`` is advisory: when it is passed, its length must equal ``payload_len`` (a truncated or over-long body is a length error, not a CRC error), but the bytes themselves are never hashed here. The returned ``crc_ok`` is purely the CRC result -- callers also check ``magic`` and ``version`` (the C receiver checks magic, the version and the length bound before it bothers with the CRC). """ if len(buf) < HEADER_SIZE: raise ValueError("buffer shorter than a %d-byte header: %d" % (HEADER_SIZE, len(buf))) prefix = _PREFIX.unpack_from(buf, 0) magic, version, kind, frame_no = prefix[:4] payload_len, rows, stretch, _reserved = prefix[4:] if payload is not None and len(payload) != payload_len: raise ValueError("payload is %d bytes but the header says %d" % (len(payload), payload_len)) stored_crc, _pad = _TAIL.unpack_from(buf, CRC_OFFSET) return { "magic": magic, "version": version, "kind": kind, "frame_no": frame_no, "payload_len": payload_len, "rows": rows, "stretch": stretch, "crc_ok": crc16(bytes(buf[:CRC_OFFSET])) == stored_crc, } # --- the status body and the Ethernet envelope --------------------------- def pack_status(frames_rx: int, crc_err: int, drops: int, csr0: int, heap_free_kb: int, dict_state: int = 0, decode_worst: int = 0) -> bytes: """Build the 16-byte status body the terminal sends (see the docstring). The field widths are the terminal's: ``frames_rx`` is a 32-bit counter that must not wrap in any run, the rest are 16-bit quantities whose limits (65,535 missed packets, a 64 MB heap) are far above anything a VT1200 reaches, so an out-of-range value is a bug here rather than a wrap on the wire. ``send_status()`` in vidstream.c writes these same offsets. ``dict_state`` is the two-bit set :data:`DICT_STATE_READY` / :data:`DICT_STATE_BAD`; ``decode_worst`` is in REFCNT ticks, not microseconds (the terminal-side comment says why). """ frames_rx = _checked("frames_rx", frames_rx, 0xFFFFFFFF) crc_err = _checked("crc_err", crc_err, 0xFFFF) drops = _checked("drops", drops, 0xFFFF) csr0 = _checked("csr0", csr0, 0xFFFF) heap_free_kb = _checked("heap_free_kb", heap_free_kb, 0xFFFF) dict_state = _checked("dict_state", dict_state, 0xFF) decode_worst = _checked("decode_worst", decode_worst, 0xFFFF) return _STATUS.pack(frames_rx, crc_err, drops, csr0, heap_free_kb, dict_state, 0, decode_worst) def unpack_status(body: bytes) -> dict: """Decode a status body, or raise ValueError if it is not 16 bytes.""" if len(body) != STATUS_SIZE: raise ValueError("a status body is %d bytes, got %d" % (STATUS_SIZE, len(body))) (frames_rx, crc_err, drops, csr0, heap_free_kb, dict_state, _reserved, decode_worst) = _STATUS.unpack_from(body, 0) return { "frames_rx": frames_rx, "crc_err": crc_err, "drops": drops, "csr0": csr0, "heap_free_kb": heap_free_kb, "dict_state": dict_state, "dict_ready": bool(dict_state & DICT_STATE_READY), "dict_bad": bool(dict_state & DICT_STATE_BAD), "decode_worst": decode_worst, } def ethernet_frame(payload: bytes, src: bytes, dst: bytes = BROADCAST, ethertype: int = ETHERTYPE, pad: bool = False) -> bytes: """Put ``payload`` in an Ethernet frame. ``pad`` is deliberately off by default and is the *caller's* decision: Ethernet needs 46 bytes after the 14-byte header, but a frame split across datagrams may only be padded at its very end. Padding a middle chunk would insert bytes between two chunks of one frame, shifting the body the receiver reassembles -- so the chunking layer pads its last datagram and nothing else. """ if len(src) != 6 or len(dst) != 6: raise ValueError("MAC addresses are 6 bytes") data = bytes(dst) + bytes(src) + struct.pack(">H", ethertype) + bytes(payload) if pad and len(data) < 14 + MIN_ETHERNET_PAYLOAD: data += b"\x00" * (14 + MIN_ETHERNET_PAYLOAD - len(data)) return data