QUELLCODE / PY
protocol.py
src/hardware_box/hid/protocol.py
1# OpenAss: nur privater, experimenteller Gebrauch; auf eigene Verantwortung.2# Niemals Unternehmensrechner, Produktivsysteme, Industrieanlagen oder3# sicherheitskritische Systeme anschließen oder bedienen.4# Haftungshinweis und gesetzliche Grenzen: ../../../DISCLAIMER.txt.5"""HB1 wire framing. CRC-16/CCITT-FALSE (poly1021, initFFFF, no reflection).67CRC covers the ASCII body before the literal `` *``. Every request and reply8has a sequence number, a CRC and LF. Acknowledgements mean firmware accepted9and emitted a report, not that the application performed the intended task.10"""1112from dataclasses import dataclass13import re1415MAX_FRAME_BYTES = 160 # firmware permits 159 bytes before LF16MAX_SEQUENCE = 0xFFFFFFFF17_BODY = re.compile(rb"HB1 (0|[1-9][0-9]*) ([A-Z][A-Z0-9_]*)(?: ([A-Za-z0-9_-]+(?: [A-Za-z0-9_-]+)*))?")181920class ProtocolError(RuntimeError):21 """Malformed, incomplete, corrupted or mismatched serial response."""222324def crc16_ccitt_false(data: bytes) -> int:25 crc = 0xFFFF26 for byte in data:27 crc ^= byte << 828 for _ in range(8):29 crc = ((crc << 1) ^ 0x1021) & 0xFFFF if crc & 0x8000 else (crc << 1) & 0xFFFF30 return crc313233@dataclass(frozen=True)34class Frame:35 sequence: int36 operation: str37 arguments: tuple[str, ...] = ()383940def encode_frame(sequence: int, operation: str, arguments: tuple[object, ...] = ()) -> bytes:41 if type(sequence) is not int or not 0 <= sequence <= MAX_SEQUENCE:42 raise ProtocolError("sequence must be uint32")43 if not isinstance(operation, str) or not re.fullmatch(r"[A-Z][A-Z0-9_]*", operation):44 raise ProtocolError("invalid operation")45 tokens = []46 for arg in arguments:47 if type(arg) is int:48 token = str(arg)49 elif isinstance(arg, str) and re.fullmatch(r"[A-Za-z0-9_-]+", arg):50 token = arg51 else:52 raise ProtocolError("arguments must be integers or single ASCII tokens")53 tokens.append(token)54 body = f"HB1 {sequence} {operation}" + (" " + " ".join(tokens) if tokens else "")55 raw = body.encode("ascii")56 result = raw + f" *{crc16_ccitt_false(raw):04X}\n".encode("ascii")57 if len(result) > MAX_FRAME_BYTES:58 raise ProtocolError("frame exceeds length limit")59 return result606162def decode_frame(data: bytes) -> Frame:63 if not isinstance(data, bytes) or not 12 <= len(data) <= MAX_FRAME_BYTES or not data.endswith(b"\n"):64 raise ProtocolError("incomplete or overlong frame")65 parts = data[:-1].rsplit(b" *", 1)66 if len(parts) != 2 or not re.fullmatch(rb"[0-9A-F]{4}", parts[1]):67 raise ProtocolError("invalid CRC field")68 body, crc = parts69 if crc16_ccitt_false(body) != int(crc, 16):70 raise ProtocolError("CRC mismatch")71 match = _BODY.fullmatch(body)72 if match is None or int(match[1]) > MAX_SEQUENCE:73 raise ProtocolError("invalid frame body")74 return Frame(int(match[1]), match[2].decode("ascii"), tuple((match[3] or b"").decode("ascii").split()))757677def require_ok(frame: Frame, sequence: int) -> tuple[str, ...]:78 if frame.sequence != sequence:79 raise ProtocolError("acknowledgement sequence mismatch")80 if frame.operation == "ERR":81 raise ProtocolError("firmware refused command: " + " ".join(frame.arguments))82 if frame.operation != "OK":83 raise ProtocolError("expected OK acknowledgement")84 return frame.arguments