Binary Struct Records With Struct Module

import struct
from dataclasses import dataclass

MAGIC = b"SNSR"
HEADER = struct.Struct("<4sBBHI")
MAX_PAYLOAD = 4096


@dataclass(frozen=True)
class Packet:
    kind: int
    timestamp: int
    payload: bytes


def encode_packet(packet: Packet) -> bytes:
    if len(packet.payload) > MAX_PAYLOAD:
        raise ValueError("payload exceeds maximum size")
    header = HEADER.pack(MAGIC, 1, packet.kind, len(packet.payload), packet.timestamp)
    return header + packet.payload


def decode_packets(buffer: bytes) -> list[Packet]:
    packets: list[Packet] = []
    offset = 0
    while offset < len(buffer):
        if len(buffer) - offset < HEADER.size:
            raise ValueError(f"truncated header at offset {offset}")
        magic, version, kind, length, timestamp = HEADER.unpack_from(buffer, offset)
        if magic != MAGIC:
            raise ValueError(f"bad magic at offset {offset}")
        if version != 1:
            raise ValueError(f"unsupported version {version}")
        if length > MAX_PAYLOAD:
            raise ValueError(f"payload length {length} exceeds limit")
        offset += HEADER.size
        end = offset + length
        if end > len(buffer):
            raise ValueError(f"truncated payload at offset {offset}")
        packets.append(
            Packet(kind=kind, timestamp=timestamp, payload=buffer[offset:end])
        )
        offset = end
    return packets


def main() -> None:
    stream = (
        encode_packet(Packet(kind=1, timestamp=1_700_000_000, payload=b"temp=21.5"))
        + encode_packet(
            Packet(kind=2, timestamp=1_700_000_060, payload=b"humidity=48")
        )
    )
    print(f"encoded {len(stream)} bytes")
    for packet in decode_packets(stream):
        print(packet.kind, packet.timestamp, packet.payload.decode("ascii"))

    try:
        decode_packets(stream[:-3])
    except ValueError as error:
        print("rejected:", error)


if __name__ == "__main__":
    main()