forked from arti/vt1200
185 lines
6.7 KiB
Python
185 lines
6.7 KiB
Python
"""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 | None = None) -> 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 -- and then it must be
|
|
exactly ``payload_len`` bytes, or a frame the CRC covers but the length
|
|
field denies would go on the wire. Omit ``payload`` to build the header
|
|
before the body exists and seal it later with :func:`seal_header`.
|
|
"""
|
|
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 None:
|
|
body = b""
|
|
else:
|
|
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 + body)
|
|
return prefix + _TAIL.pack(crc, 0)
|
|
|
|
|
|
def seal_header(header: bytes, payload: bytes = b"") -> bytes:
|
|
"""Return ``header`` with its CRC field recomputed over ``payload``.
|
|
|
|
``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 lose the payload from the CRC.
|
|
|
|
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 must be exactly %d bytes, got %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,
|
|
}
|