"""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. Delta frames ------------ The same module carries Task 5's changed-tile list, because it is the other half of the tile wire format and the two layouts share the dictionary: KIND_TILES_FULL (2) 1,280 dictionary indices, row-major, one byte each KIND_TILES_DELTA (3) a u16 count, then `count` u16 positions and `count` one-byte dictionary entries A position is a row-major tile index 0..1,279, so it needs **two** bytes while one dictionary entry fits in one -- the entry space is 256, the tile grid is 1,280. :func:`delta_frame` writes the payload, :func:`unpack_delta` reads it back with the same rules the C decoder enforces (exact length, positions on the grid and strictly ascending), and :func:`keyframe_due` is the cadence rule: a ``TILES_FULL`` whenever the host has no previous frame, every :data:`KEYFRAME_EVERY` frames, and as soon as the terminal's status frame reports a drop. Positions travel separately from values (rather than interleaved as pairs) for one reason: a receiver can validate the whole position list before it touches a pixel, so a frame with one bad position paints **nothing**. Interleaved pairs invite a decoder that paints as it reads, which is the failure the harness's off-grid negative control exists to catch. Pure standard library. """ import struct from harness.udpstream.frame import (HEADER_SIZE, KIND_DICT_CHUNK, MAGIC, VERSION, pack_header, unpack_header) from harness.udpstream.tilecost import TILES_PER_SCREEN __all__ = [ "DELTA_HDR", "KEYFRAME_EVERY", "MAX_BODY", "MAX_DICT_BYTES", "MAX_PAYLOAD", "OFFSET_SIZE", "delta_frame", "dict_chunks", "keyframe_due", "payload_body", "payload_offset", "unpack_delta", ] #: 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 #: A delta payload opens with the u16 entry count (see frame_parse.h). DELTA_HDR = 2 #: Send a full frame (``TILES_FULL``) this often, whether or not anything was #: lost. Ruling R2 of the plan: 60 frames is ~2 s at the sender's 30 fps, so a #: desynchronised screen can never stay wrong for longer than that. KEYFRAME_EVERY = 60 _OFFSET = struct.Struct(" 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:] # --- Task 5: changed-tile lists ------------------------------------------- def _indices(name: str, frame) -> bytes: """Range-check one full-screen index frame and return it as ``bytes``.""" if not isinstance(frame, (bytes, bytearray, memoryview)): raise TypeError("%s is bytes-like, got %s" % (name, type(frame).__name__)) data = bytes(frame) if len(data) != TILES_PER_SCREEN: raise ValueError("%s is one index byte per tile: %d bytes, got %d" % (name, TILES_PER_SCREEN, len(data))) return data def delta_frame(prev: bytes, cur: bytes) -> bytes | None: """Return the ``KIND_TILES_DELTA`` payload from ``prev`` to ``cur``. ``None`` when the frames are identical: there is nothing to put on the wire, and an empty list is not a frame the receiver should have to accept. Otherwise the payload is u16 count | count * u16 position | count * u8 dictionary entry with positions in ascending order (a scan of the row-major grid produces them that way) and each entry the *new* value for that tile. Everything is recomputed from the two frames -- nothing is carried over from a previous delta -- so a delta lost in transit cannot corrupt the next one; it is the periodic keyframe that repairs the screen. """ before = _indices("prev", prev) after = _indices("cur", cur) positions = [i for i in range(TILES_PER_SCREEN) if before[i] != after[i]] if not positions: return None values = bytes(after[i] for i in positions) return (_COUNT.pack(len(positions)) + b"".join(_COUNT.pack(i) for i in positions) + values) def unpack_delta(payload: bytes) -> list[tuple[int, int]]: """Decode a ``KIND_TILES_DELTA`` payload, or raise ``ValueError``. The rules are the C decoder's (``paint_delta`` in ``vidstream.c``), and they are the reason a delta cannot corrupt the screen: * the payload holds exactly ``2 + 3 * count`` bytes, so the list cannot run past the frame the header described; * ``count`` is at least 1 and at most :data:`TILES_PER_SCREEN`; * every position is on the 40 x 32 grid, so no blit is ever aimed past the last tile; * positions are strictly ascending, so no tile is named twice and a reordered list is a refusal rather than a different picture. Each value is a byte and therefore already a dictionary entry, so the "entry index < 256" rule needs no check on this side. """ if not isinstance(payload, (bytes, bytearray, memoryview)): raise TypeError("a delta payload is bytes-like, got %s" % type(payload).__name__) body = bytes(payload) if len(body) < DELTA_HDR: raise ValueError("a delta payload opens with a %d-byte count, got %d" % (DELTA_HDR, len(body))) count = _COUNT.unpack_from(body, 0)[0] if not 1 <= count <= TILES_PER_SCREEN: raise ValueError("a delta lists 1..%d tiles, got %d" % (TILES_PER_SCREEN, count)) if len(body) != DELTA_HDR + 3 * count: raise ValueError("a %d-entry delta is %d bytes, got %d" % (count, DELTA_HDR + 3 * count, len(body))) positions = [] previous = -1 for k in range(count): position = _COUNT.unpack_from(body, DELTA_HDR + 2 * k)[0] if position >= TILES_PER_SCREEN: raise ValueError("tile position %d is off the %d-tile grid" % (position, TILES_PER_SCREEN)) if position <= previous: raise ValueError("tile positions must be strictly ascending: " "%d follows %d" % (position, previous)) previous = position positions.append(position) values = body[DELTA_HDR + 2 * count:] return list(zip(positions, values)) def keyframe_due(frame_no: int, drops: int = 0, seen_drops: int = 0, every: int = KEYFRAME_EVERY) -> bool: """True when the host must send a ``TILES_FULL`` instead of a delta. Two triggers, and both are "the terminal may not have what the host thinks it has": * ``frame_no`` is a multiple of ``every`` -- a fixed cadence that bounds how long a desynchronised screen can stay wrong (ruling R2). Frame 0 is in this set because there is no previous frame to delta against; * ``drops`` rose since ``seen_drops``: the status frame's counter is cumulative, so any increase means at least one frame never reached the screen and the host's idea of the screen is stale. ``crc_err`` is deliberately not a trigger. A refused delta *does* leave the terminal's screen behind the host's -- the host has moved to the next frame and the terminal never painted that one -- so the reason is not that nothing was desynchronised. It is that a fixed 60-frame cadence already bounds how long that desync lasts, and a refused frame carries no screen state a keyframe could recover sooner than the schedule does. ``drops`` is different: it counts frames the terminal *accepted* and then lost, which is the counter the terminal raises for the same repair. """ if (not isinstance(frame_no, int) or isinstance(frame_no, bool) or frame_no < 0): raise ValueError("frame_no is a non-negative int, got %r" % (frame_no,)) if not isinstance(every, int) or isinstance(every, bool) or every < 1: raise ValueError("every is a positive int, got %r" % (every,)) return frame_no % every == 0 or drops > seen_drops