feat(framing): framing done

refactor(framing): re-name hamming codec to 7,4 to reflect algorithm used

framing.py

@@ -0,0 +1,46 @@
+from header import Onbeat_Header, PROTOCOL_IDENTIFIER
+from reedsolo import RSCodec
+import numpy as np
+
+PROTOCOL_V1_NOFEC = 1
+PROTOCOL_V1_FEC = 2
+RSC = RSCodec(255-223,255)
+
+def decode_packet(packet: list[int]) -> tuple[Onbeat_Header, list[int]]:
+    if len(packet) < 32:
+        raise OverflowError(f"Packet is too small to be valid, len: {len(packet)}")
+    header_raw = packet[0:32]
+    header = Onbeat_Header(0, 0, "", 0, 0)
+    if header.decode(header_raw) != True:
+        raise RuntimeError("invalid CRC")
+    if header.protocol_id != PROTOCOL_IDENTIFIER:
+        raise ValueError(f"Protocol ID mismatch, got{header.protocol_id}, expected {PROTOCOL_IDENTIFIER}")
+    if header.pkt_len > len(packet) - 32:
+        raise OverflowError(f"Header not fully captured, len:{header.pkt_len}, maximal: {len(packet) - 32}")
+    elif header.pkt_len == 0:
+        return header, []
+    payload = packet[32:32+header.pkt_len]
+    if header.protocol_configuration == PROTOCOL_V1_FEC:
+        payload_decoded = np.empty((4, min(len(payload)//4-32, 255)), dtype=int)
+        for i in range(0,4):
+            payload_decoded[i] = RSC.decode(bytearray(payload[i::4]))[0]
+        payload = payload_decoded.flatten(order='F').tolist()
+    elif header.protocol_configuration == PROTOCOL_V1_NOFEC:
+        pass
+    else:
+        raise ValueError(f"Unknown protocol configuration: {header.protocol_configuration}")
+    return header, payload
+
+def encode_packet(payload: list[int], callsign: str, sequence: int, rs_coding: bool = False) -> list[int]:
+    protocol_configuration = PROTOCOL_V1_NOFEC
+    if rs_coding == True:
+        protocol_configuration = PROTOCOL_V1_FEC
+        if len(payload) != 0:
+            for _ in range(0, len(payload)%4):
+                payload.append(0)
+            payload_encoded = np.empty((4, min((len(payload))//4+32, 255)), dtype=int)
+            for i in range(0,4):
+                payload_encoded[i] = RSC.encode(bytearray(payload[i::4]))
+            payload = payload_encoded.flatten(order='F').tolist()
+    header = Onbeat_Header(0, protocol_configuration, callsign, len(payload), sequence)
+    return header.encode() + payload

hamming_7_4_codec.py

@@ -19,7 +19,7 @@ DECODER_MATRIX = np.asarray([(0, 0, 1, 0, 0, 0, 0,),
 
 def number_to_list(number: int, N: int) -> npt.NDArray:
     """Return the last N bits of a number as an MSB-first array"""
-    return np.array([(number >> (N-1-i)) % 2 for i in range(0, N)])
+    return np.array([(number >> (N-1-i)) % 2 for i in range(0, N)],dtype=int)
 
 
 def list_to_number(array: npt.NDArray, N: int) -> int:
@@ -54,7 +54,8 @@ def decode_nibble(data: int) -> int:
     syndrome = (PARITY_MATRIX @ data_asarr) % 2
     error = list_to_number(syndrome, 3)
     if error == 0:
-        return list_to_number(data_asarr, 4)
+        data_decoded = (DECODER_MATRIX @ data_asarr) % 2
+        return list_to_number(data_decoded, 4)
     data_asarr[error - 1] ^= 1
     data_decoded = (DECODER_MATRIX @ data_asarr) % 2
     return list_to_number(data_decoded, 4)
@@ -86,12 +87,10 @@ if __name__ == "__main__":
     data_corrected_2bit = hamming_7_4_decode(data_corrupted_2bit)
     data_corrected_2bit_str = [int.to_bytes(int(i), 1, "big").decode(
         encoding="utf-8", errors="ignore") for i in data_corrected_2bit]
-    print(f"Recovered data from 1-bit errors: {data_corrected_2bit_str}")
+    print(f"Recovered data from 2-bit errors: {data_corrected_2bit_str}")
 
-    data_corrupted_3bit = [data ^ (7 << (idx % 7))
-                           for idx, data in enumerate(data_encoded)]
     # print(f"Corrupt data with 3-bit errors: {data_corrupted_3bit}")
-    data_corrected_3bit = hamming_7_4_decode(data_corrupted_3bit)
-    data_corrected_3bit_str = [int.to_bytes(int(i), 1, "big").decode(
-        encoding="utf-8", errors="ignore") for i in data_corrected_3bit]
-    print(f"Recovered data from 1-bit errors: {data_corrected_3bit_str}")
+    data_corrected_noerr = hamming_7_4_decode(data_encoded)
+    data_corrected_noerr_str = [int.to_bytes(int(i), 1, "big").decode(
+        encoding="utf-8", errors="ignore") for i in data_corrected_noerr]
+    print(f"Recovered data from no errors: {data_corrected_noerr_str}")

header.py

@@ -1,12 +1,12 @@
-from hamming_8_4_codec import hamming_8_4_encode, hamming_8_4_decode
-from crc import Configuration, Crc16
+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 = Configuration(Crc16.IBM_3740)
+ONBEAT_CRC = Calculator(Crc16.IBM_3740)
 
 
 class Onbeat_Header:
-    """This class represents a header for a single packet of PLSTV."""
+    """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:
@@ -20,9 +20,11 @@ class Onbeat_Header:
         self.protocol_configuration = protocol_configuration
 
         call_len = len(callsign.encode())
-        if call_len >= 10:
+        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:
@@ -30,7 +32,7 @@ class Onbeat_Header:
                 f"Maximum allowed packet size is {2 << 16 - 1} got {pkt_len}")
         self.pkt_len = pkt_len
 
-        if pkt_sequence_id >= 2 << 8:
+        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
@@ -39,27 +41,18 @@ class Onbeat_Header:
         header_asints = []
         header_asints += [(PROTOCOL_IDENTIFIER << 4) +
                           self.protocol_configuration]
-        header_asints += [(int.from_bytes(self.callsign.encode())
-                           >> 8*i) % 2 << 8 for i in range(0, 10)]
-        header_asints += [(self.pkt_len >> 8), self.pkt_len % 2 << 8]
+        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]
-        return hamming_8_4_encode(header_asints)
+        header_asints += [header_crc >> 8, header_crc % 256]
+        return hamming_7_4_encode(header_asints)
 
-    def decode(self, header_encoded: list[int]):
-        header_corrected = hamming_8_4_decode(header_encoded)
-        checksum = ONBEAT_CRC.checksum(
-            bytes(header_corrected[14] << 8 + header_corrected[15]))
-        if checksum != 0:
-            raise ValueError(
-                "Checksum of header is non-zero, packet is invalid")
-        self.protocol_id = header_corrected[0] >> 4
-        self.protocol_configuration = header_corrected[0] % 16
-        callsign_numeric = 0
-        for i in range(1, 11):
-            callsign_numeric += header_corrected[i]
-            callsign_numeric <<= 8
-        self.callsign = int.to_bytes(callsign_numeric, byteorder="big").decode()
-        self.pkt_len = header_corrected[11] << 8 + header_corrected[12]
-        self.pkt_sequence_id = header_corrected[13]
+    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])

hier_blocks/ONBEAT_mod.grc

@@ -90,7 +90,7 @@ blocks:
     key: pmt.intern("packet_len")
     offset: '0'
     src: pmt.intern("src")
-    value: pmt.from_long(64)
+    value: pmt.from_long(8)
   states:
     bus_sink: false
     bus_source: false
@@ -252,23 +252,6 @@ blocks:
     coordinate: [272, 504.0]
     rotation: 0
     state: disabled
-- name: data_len
-  id: parameter
-  parameters:
-    alias: ''
-    comment: ''
-    hide: none
-    label: ''
-    short_id: ''
-    type: ''
-    value: '1024'
-  states:
-    bus_sink: false
-    bus_source: false
-    bus_structure: null
-    coordinate: [976, 24.0]
-    rotation: 0
-    state: enabled
 - name: digital_constellation_modulator_0
   id: digital_constellation_modulator
   parameters:
@@ -288,7 +271,7 @@ blocks:
     bus_sink: false
     bus_source: false
     bus_structure: null
-    coordinate: [1088, 296.0]
+    coordinate: [1096, 312.0]
     rotation: 0
     state: enabled
 - name: pad_sink_0
@@ -306,25 +289,7 @@ blocks:
     bus_sink: false
     bus_source: false
     bus_structure: null
-    coordinate: [1352, 320.0]
-    rotation: 0
-    state: enabled
-- name: pad_sink_0_0
-  id: pad_sink
-  parameters:
-    affinity: ''
-    alias: ''
-    comment: ''
-    label: dbg
-    num_streams: '1'
-    optional: 'True'
-    type: byte
-    vlen: '1'
-  states:
-    bus_sink: false
-    bus_source: false
-    bus_structure: null
-    coordinate: [576, 136.0]
+    coordinate: [1352, 336.0]
     rotation: 0
     state: enabled
 - name: pad_source_0
@@ -386,7 +351,6 @@ connections:
 - [blocks_repack_bits_bb_1_0, '0', blocks_tagged_stream_mux_0, '1']
 - [blocks_tagged_stream_mux_0, '0', digital_constellation_modulator_0, '0']
 - [blocks_vector_source_x_0_0, '0', blocks_tagged_stream_mux_0, '0']
-- [blocks_vector_source_x_0_0, '0', pad_sink_0_0, '0']
 - [blocks_vector_source_x_0_0_0, '0', blocks_repack_bits_bb_1_0, '0']
 - [blocks_vector_source_x_0_0_0_0, '0', blocks_repack_bits_bb_1_0_0, '0']
 - [digital_constellation_modulator_0, '0', pad_sink_0, '0']