László Párkányi
5bf5bfec5d
refactor(framing): re-name hamming codec to 7,4 to reflect algorithm used
59 lines
2.7 KiB
Python
59 lines
2.7 KiB
Python
from hamming_7_4_codec import hamming_7_4_encode, hamming_7_4_decode
|
|
from crc import Calculator, Crc16
|
|
PROTOCOL_IDENTIFIER = 0x0
|
|
# CRC_POLY = (0xed2f << 1) + 1 # from https://users.ece.cmu.edu/~koopman/crc/index.html as "best 16-bit CRC" on 2025-10-24
|
|
ONBEAT_CRC = Calculator(Crc16.IBM_3740)
|
|
|
|
|
|
class Onbeat_Header:
|
|
"""This class represents a header for a single packet of ONBEAT."""
|
|
|
|
def __init__(self, protocol_id, protocol_configuration: int, callsign: str, pkt_len: int, pkt_sequence_id: int):
|
|
if protocol_id >= 2 << 4:
|
|
raise OverflowError(
|
|
f"protocol identifier should be confined to 4 bits, got {protocol_id}")
|
|
self.protocol_id = protocol_id
|
|
|
|
if protocol_configuration >= 2 << 4:
|
|
raise OverflowError(
|
|
f"protocol configuration should be confined to 4 bits, got {protocol_configuration}")
|
|
self.protocol_configuration = protocol_configuration
|
|
|
|
call_len = len(callsign.encode())
|
|
if call_len > 10:
|
|
raise OverflowError(
|
|
f"Callsign must be confined to 10 bytes, got {callsign} with length {call_len} (using UTF-8)")
|
|
for _ in range(call_len, 10):
|
|
callsign += "\0"
|
|
self.callsign = callsign
|
|
|
|
if pkt_len >= 2 << 16:
|
|
raise OverflowError(
|
|
f"Maximum allowed packet size is {2 << 16 - 1} got {pkt_len}")
|
|
self.pkt_len = pkt_len
|
|
|
|
if pkt_sequence_id >= 256:
|
|
raise OverflowError(
|
|
f"Packet sequence ID must be confined to 8 bits, got {pkt_sequence_id}")
|
|
self.pkt_sequence_id = pkt_sequence_id
|
|
|
|
def encode(self) -> list[int]:
|
|
header_asints = []
|
|
header_asints += [(PROTOCOL_IDENTIFIER << 4) +
|
|
self.protocol_configuration]
|
|
header_asints += [ord(c) for c in self.callsign]
|
|
header_asints += [(self.pkt_len >> 8), self.pkt_len % 256]
|
|
header_asints += [self.pkt_sequence_id]
|
|
header_crc = ONBEAT_CRC.checksum(bytes(header_asints))
|
|
header_asints += [header_crc >> 8, header_crc % 256]
|
|
return hamming_7_4_encode(header_asints)
|
|
|
|
def decode(self, header_encoded: list[int]) -> bool:
|
|
header_corrected = hamming_7_4_decode(header_encoded)
|
|
self.protocol_id = int(header_corrected[0] >> 4)
|
|
self.protocol_configuration = int(header_corrected[0] % 16)
|
|
self.callsign = bytes(header_corrected[1:11]).decode(errors="ignore")
|
|
self.pkt_len = int((header_corrected[11] << 8) + header_corrected[12])
|
|
self.pkt_sequence_id = int(header_corrected[13])
|
|
return ONBEAT_CRC.verify(bytearray(header_corrected[0:14]), (header_corrected[14] << 8) + header_corrected[15])
|