File size: 5,178 Bytes
3280bb4 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 | import argparse
import hashlib
# Protocol Constants from compress_chirp3.py
SYNC_MARKER = 0xBB
PKT_SIZE = 255
TRANSPORT_HDR = 3
DATA_PER_PKT = PKT_SIZE - TRANSPORT_HDR # 252 Bytes
def xor_fec_parity(data_packets):
"""Computes XOR parity byte-by-byte across all data packets."""
parity = bytearray(DATA_PER_PKT)
for pkt in data_packets:
# Extract data segment (excluding transport header)
data_part = pkt[TRANSPORT_HDR:]
for idx in range(min(len(data_part), DATA_PER_PKT)):
parity[idx] ^= data_part[idx]
return bytes(parity)
def pack_payload(payload_bytes, num_data_packets):
"""Encapsulates payload into N-1 data packets and 1 XOR-FEC parity packet."""
total_capacity = num_data_packets * DATA_PER_PKT
# Pad payload if it's smaller than the capacity
if len(payload_bytes) < total_capacity:
payload_bytes = payload_bytes.ljust(total_capacity, b'\x00')
elif len(payload_bytes) > total_capacity:
payload_bytes = payload_bytes[:total_capacity]
data_packets = []
total_packets = num_data_packets + 1
for idx in range(num_data_packets):
chunk = payload_bytes[idx * DATA_PER_PKT : (idx + 1) * DATA_PER_PKT]
header = bytes([SYNC_MARKER, idx, total_packets])
data_packets.append(header + chunk)
# Generate XOR-parity packet
parity_data = xor_fec_parity(data_packets)
parity_header = bytes([SYNC_MARKER, num_data_packets, total_packets])
parity_packet = parity_header + parity_data
return data_packets + [parity_packet]
def run_proof():
print("======================================================================")
print("ZYMATICA | Chirp Packetization & XOR-FEC Transmission Channel Proof")
print("======================================================================\n")
# 1. Prepare raw payload
raw_payload = b"ip zymatica.space | " * 50 # 1000 bytes payload
payload_hash = hashlib.sha256(raw_payload).hexdigest()
print(f"[1] Source Payload Prepared:")
print(f" - Size: {len(raw_payload)} bytes")
print(f" - SHA-256 Checksum: {payload_hash}")
# 2. Pack payload into chirps
num_data_pkts = 4
packets = pack_payload(raw_payload, num_data_pkts)
print(f"\n[2] Packaging Payload into {len(packets)} LoRa Chirp-3 Packets:")
for idx, pkt in enumerate(packets):
ptype = "DATA" if idx < num_data_pkts else "FEC-PARITY"
print(f" - Packet {idx}: Sync=0x{pkt[0]:02X}, Idx={pkt[1]}, Total={pkt[2]}, Size={len(pkt)} bytes ({ptype})")
# 3. Simulate transmission with exactly one lost packet (Packet index 2 is dropped)
dropped_index = 2
print(f"\n[3] Simulating Lossy Channel Transmission...")
print(f" -> WARNING: Packet index {dropped_index} dropped during transit.")
received_packets = [pkt for idx, pkt in enumerate(packets) if idx != dropped_index]
# 4. Perform XOR-FEC Recovery on the receiver
print(f"\n[4] Executing Forward Error Correction (XOR-FEC) Reassembler...")
# Identify which packet is missing
received_indices = {pkt[1] for pkt in received_packets}
total_packets = received_packets[0][2]
missing_index = None
for idx in range(total_packets):
if idx not in received_indices:
missing_index = idx
break
print(f" -> Detected missing packet index: {missing_index}")
# Recover missing packet by XORing all received packets' payloads
recovered_data = bytearray(DATA_PER_PKT)
for pkt in received_packets:
data_part = pkt[TRANSPORT_HDR:]
for idx in range(DATA_PER_PKT):
recovered_data[idx] ^= data_part[idx]
recovered_packet = bytes([SYNC_MARKER, missing_index, total_packets]) + bytes(recovered_data)
print(f" -> Packet index {missing_index} reconstructed successfully.")
# Insert recovered packet back into the buffer
all_reconstructed_packets = list(received_packets)
all_reconstructed_packets.append(recovered_packet)
# Sort by packet index (byte at offset 1)
all_reconstructed_packets.sort(key=lambda x: x[1])
# 5. Reassemble and verify payload
reassembled_payload = bytearray()
for idx in range(num_data_pkts):
reassembled_payload.extend(all_reconstructed_packets[idx][TRANSPORT_HDR:])
# Trim padding if necessary to match original length
reassembled_payload = bytes(reassembled_payload[:len(raw_payload)])
reassembled_hash = hashlib.sha256(reassembled_payload).hexdigest()
print(f"\n[5] Reassembled Payload Checksum Verification:")
print(f" - Original SHA-256: {payload_hash}")
print(f" - Reassembled SHA-256: {reassembled_hash}")
assert payload_hash == reassembled_hash, "Checksum validation failed! Data corrupted."
print("\n[VERIFICATION] Lossless XOR-FEC reconstruction validated. No data loss.")
if __name__ == "__main__":
parser = argparse.ArgumentParser(description="Zymatica LoRa FEC Proof")
parser.add_argument("--test", action="store_true", help="Run test mode")
args = parser.parse_args()
run_proof()
|