opencode.energia.arti.ee/vt1200
1
0 Fork 0
forked from arti/vt1200
Code Pull Requests Activity

feat(stream): frame header and CRC16 shared by both ends

Browse Source
This commit is contained in:
opencode.energia.arti.ee committed 2026-10-30 22:15:18 +00:00
1 parent be1e8b28a7
commit 16a52bb853
2 files changed
+292
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

harness/udpstream/frame.py
+168
View File
@@ -0,0 +1,168 @@
"""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 (KIND_STRIP = 1, KIND_STATUS = 5)
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 plus the
payload**, never the CRC field itself.
``unpack_header(buf)`` with no payload therefore checks a bare header (the CRC
of an empty payload). ``unpack_header(buf, payload)`` checks a whole frame,
which is what the receiving side does. A payload that is not passed is simply
not covered, so a header sealed over a payload will *not* validate without it.
"""
import struct
__all__ = [
"HEADER_SIZE", "MAGIC", "VERSION", "CRC_OFFSET",
"KIND_STRIP", "KIND_STATUS",
"crc16", "pack_header", "seal_header", "unpack_header",
]
#: 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) plus the payload.
CRC_OFFSET = 12
#: A strip of packed 1-bpp source rows.
KIND_STRIP = 1
#: A status report travelling the other way.
KIND_STATUS = 5
#: Everything before the CRC field, little-endian.
_PREFIX = struct.Struct("<2sBBHHBBH")
#: The CRC field and the two padding bytes behind it.
_TAIL = struct.Struct("<HH")
def _make_crc_table():
"""Build the 256-entry CCITT-FALSE table from the plain bit loop.
Keeping the loop here means the polynomial, the init value and the shift
direction are stated exactly once, in the form ``vidstream.c`` repeats.
The table is only how the host gets through a 12 KB frame in ~2 ms instead
of ~28 ms, which a 30 fps sender cannot afford.
"""
table = []
for byte in range(256):
crc = byte << 8
for _ in range(8):
if crc & 0x8000:
crc = ((crc << 1) ^ 0x1021) & 0xFFFF
else:
crc = (crc << 1) & 0xFFFF
table.append(crc)
return tuple(table)
_CRC_TABLE = _make_crc_table()
def crc16(data: bytes) -> 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 = b"") -> bytes:
"""Build the 16-byte header for a frame.
The CRC is computed over header bytes 0..11 plus ``payload``. Pass the
payload here to seal a complete frame in one call; with the default empty
payload the header carries the CRC of a header-only frame (use
:func:`seal_header` to add the CRC later -- that is what the chunked sender
does, since it builds the header before splitting the payload).
"""
prefix = _PREFIX.pack(
MAGIC,
VERSION,
_checked("kind", kind, 0xFF),
_checked("frame_no", frame_no, 0xFFFF),
_checked("payload_len", payload_len, 0xFFFF),
_checked("rows", rows, 0xFF),
_checked("stretch", stretch, 0xFF),
0, # reserved
)
assert len(prefix) == CRC_OFFSET
crc = crc16(prefix + bytes(payload))
return prefix + _TAIL.pack(crc, 0)
def seal_header(header: bytes, payload: bytes = b"") -> bytes:
"""Return ``header`` with its CRC field recomputed over ``payload``.
Byte-for-byte identical to ``pack_header(..., payload=payload)`` for a
header that came from :func:`pack_header`; kept separate so a caller that
assembles a frame in pieces (header, then payload) can still seal it.
"""
if len(header) < HEADER_SIZE:
raise ValueError("header shorter than %d bytes: %d"
% (HEADER_SIZE, len(header)))
crc = crc16(bytes(header[:CRC_OFFSET]) + bytes(payload))
return bytes(header[:CRC_OFFSET]) + _TAIL.pack(crc, 0)
def unpack_header(buf: bytes, payload: bytes = b"") -> dict:
"""Parse a frame header and check its CRC.
``payload`` is the frame body that follows the header; the CRC is computed
over ``header[0:12] + payload``. The returned ``crc_ok`` is purely the CRC
result -- callers also check ``magic`` and ``version`` (the C receiver
checks magic 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:]
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]) + bytes(payload)) == stored_crc,
}
harness/udpstream/tests/test_frame.py
+124
View File
@@ -0,0 +1,124 @@
"""Tests for the frame header and CRC shared by the host sender and the payload.
The plan's two tests come first, verbatim. The rest pin down the parts of the
format the plan text is ambiguous about, because both ends must agree on them:
the CCITT-FALSE polynomial, the CRC covering header[0:12] *and* the payload, and
the exact byte offsets the C side will mirror.
"""
import struct
import pytest
from harness.udpstream.frame import (CRC_OFFSET, HEADER_SIZE, KIND_STATUS,
KIND_STRIP, MAGIC, VERSION, crc16,
pack_header, seal_header, unpack_header)
def test_header_is_16_bytes_and_round_trips():
h = pack_header(KIND_STRIP, frame_no=7, payload_len=12800, rows=80, stretch=12)
assert len(h) == HEADER_SIZE
d = unpack_header(h)
assert d["kind"] == KIND_STRIP and d["frame_no"] == 7
assert d["payload_len"] == 12800 and d["rows"] == 80 and d["stretch"] == 12
assert d["crc_ok"] is True
def test_crc_detects_a_single_bit_flip():
b = bytearray(pack_header(KIND_STRIP, 1, 10, 8, 128))
b[5] ^= 0x01
assert unpack_header(bytes(b))["crc_ok"] is False
# --- the CRC itself ------------------------------------------------------
def test_crc16_is_ccitt_false():
"""The canonical CCITT-FALSE check value; catches a wrong poly/init/reflect."""
assert crc16(b"123456789") == 0x29B1
assert crc16(b"") == 0xFFFF
def test_crc16_matches_an_independent_implementation():
"""Cross-check the table against the stdlib, which is CCITT with init 0xFFFF.
One byte per call exercises every table entry; the longer inputs exercise
the running state, including the sizes the stream actually uses.
"""
import binascii
for b in range(256):
assert crc16(bytes([b])) == binascii.crc_hqx(bytes([b]), 0xFFFF), b
for n in (0, 1, 12, 160, 12816, 20496):
data = bytes((i * 37 + 11) & 0xFF for i in range(n))
assert crc16(data) == binascii.crc_hqx(data, 0xFFFF), n
def test_crc16_spans_the_payload():
payload = bytes(range(256)) * 2
h = pack_header(KIND_STRIP, frame_no=1, payload_len=len(payload), rows=8,
stretch=128, payload=payload)
assert unpack_header(h, payload)["crc_ok"] is True
# The same header checked without its payload must not validate: that is the
# difference between "header covers the payload" and "header only".
assert unpack_header(h)["crc_ok"] is False
# And a payload bit flip is caught.
bad = bytearray(payload)
bad[100] ^= 0x80
assert unpack_header(h, bytes(bad))["crc_ok"] is False
def test_seal_header_after_the_payload_is_known():
"""The chunked sender builds the header first, then seals it (Task 2/6 path)."""
payload = b"\x01\x02\x03\x04"
h = pack_header(KIND_STRIP, 3, len(payload), 8, 128)
sealed = seal_header(h, payload)
assert sealed[:CRC_OFFSET] == h[:CRC_OFFSET]
assert unpack_header(sealed, payload)["crc_ok"] is True
# Sealing twice is idempotent.
assert seal_header(sealed, payload) == sealed
# And it produces the same bytes as the one-shot call.
assert sealed == pack_header(KIND_STRIP, 3, len(payload), 8, 128,
payload=payload)
# --- the byte layout the C guest will mirror -----------------------------
def test_byte_offsets_and_endianness():
h = pack_header(KIND_STATUS, frame_no=0x1234, payload_len=0x5678, rows=0x9A,
stretch=0xBC)
assert h[0:2] == MAGIC == b"VS"
assert h[2] == VERSION == 0x01
assert h[3] == KIND_STATUS == 5
assert h[4:6] == b"\x34\x12" # frame_no, little-endian
assert h[6:8] == b"\x78\x56" # payload_len, little-endian
assert h[8] == 0x9A # rows
assert h[9] == 0xBC # stretch
assert h[10:12] == b"\x00\x00" # reserved
assert h[12:14] == struct.pack("<H", crc16(h[0:12]))
assert h[14:16] == b"\x00\x00" # padding to HEADER_SIZE
def test_unpack_reports_magic_and_version():
d = unpack_header(pack_header(KIND_STRIP, 0, 0, 0, 1))
assert d["magic"] == b"VS" and d["version"] == VERSION
d = unpack_header(b"XX" + pack_header(KIND_STRIP, 0, 0, 0, 1)[2:])
assert d["magic"] == b"XX"
# --- malformed input -----------------------------------------------------
def test_short_buffer_is_rejected():
with pytest.raises(ValueError):
unpack_header(b"V" * (HEADER_SIZE - 1))
@pytest.mark.parametrize("kwargs", [
dict(kind=256), dict(kind=-1),
dict(frame_no=0x10000), dict(frame_no=-1),
dict(payload_len=0x10000), dict(payload_len=-1),
dict(rows=256), dict(rows=-1),
dict(stretch=256), dict(stretch=-1),
])
def test_out_of_range_fields_raise(kwargs):
base = dict(kind=KIND_STRIP, frame_no=1, payload_len=10, rows=8, stretch=128)
base.update(kwargs)
with pytest.raises(ValueError):
pack_header(**base)