Ryan Malloy
c839d225a8
GNU Radio Out-of-Tree module providing: - Complete TX chain: PHYEncode → FrameGen → CSSMod - Complete RX chain: CSSDemod → FrameSync → PHYDecode - NETWORKID extraction/encoding (0-255 range) - All SF (7-12) and CR (4/5-4/8) combinations - Loopback tested with 24/24 configurations passing Key features: - Fractional SFD (2.25 downchirp) handling - Gray encode/decode with proper inverse operations - gr-lora_sdr compatible decode modes - GRC block definitions and example flowgraphs - Comprehensive documentation Discovered RYLR998 sync word mapping: sync_bin_1 = (NETWORKID >> 4) * 8 sync_bin_2 = (NETWORKID & 0x0F) * 8
319 lines
11 KiB
Python
319 lines
11 KiB
Python
"""LoRa PHY layer encoder: Whiten → Hamming FEC → interleave → Gray encode.
|
|
|
|
Complete transmit chain for encoding payload bytes into CSS symbol bins.
|
|
|
|
TX chain order:
|
|
Payload bytes → CRC append → Nibble split → Whiten → Hamming FEC
|
|
→ Interleave → Gray encode → Symbol bins
|
|
"""
|
|
|
|
from dataclasses import dataclass
|
|
from typing import Optional
|
|
|
|
|
|
# Whitening sequence (same as RX - XOR is its own inverse)
|
|
WHITEN_SEQ = bytes([
|
|
0xFF, 0xFE, 0xFC, 0xF8, 0xF0, 0xE1, 0xC2, 0x85,
|
|
0x0B, 0x17, 0x2F, 0x5E, 0xBC, 0x78, 0xF1, 0xE3,
|
|
0xC6, 0x8D, 0x1A, 0x34, 0x68, 0xD0, 0xA0, 0x40,
|
|
0x80, 0x01, 0x02, 0x04, 0x08, 0x11, 0x23, 0x47,
|
|
0x8E, 0x1C, 0x38, 0x71, 0xE2, 0xC4, 0x89, 0x12,
|
|
0x25, 0x4B, 0x97, 0x2E, 0x5C, 0xB8, 0x70, 0xE0,
|
|
0xC0, 0x81, 0x03, 0x06, 0x0C, 0x19, 0x32, 0x64,
|
|
0xC9, 0x92, 0x24, 0x49, 0x93, 0x26, 0x4D, 0x9B,
|
|
0x37, 0x6E, 0xDC, 0xB9, 0x72, 0xE4, 0xC8, 0x90,
|
|
0x20, 0x41, 0x82, 0x05, 0x0A, 0x15, 0x2B, 0x56,
|
|
0xAD, 0x5B, 0xB6, 0x6D, 0xDA, 0xB5, 0x6B, 0xD6,
|
|
0xAC, 0x59, 0xB2, 0x65, 0xCB, 0x96, 0x2C, 0x58,
|
|
0xB0, 0x61, 0xC3, 0x87, 0x0F, 0x1F, 0x3E, 0x7D,
|
|
0xFA, 0xF4, 0xE8, 0xD1, 0xA2, 0x44, 0x88, 0x10,
|
|
0x21, 0x43, 0x86, 0x0D, 0x1B, 0x36, 0x6C, 0xD8,
|
|
0xB1, 0x63, 0xC7, 0x8F, 0x1E, 0x3C, 0x79, 0xF3,
|
|
0xE7, 0xCE, 0x9C, 0x39, 0x73, 0xE6, 0xCC, 0x98,
|
|
0x31, 0x62, 0xC5, 0x8B, 0x16, 0x2D, 0x5A, 0xB4,
|
|
0x69, 0xD2, 0xA4, 0x48, 0x91, 0x22, 0x45, 0x8A,
|
|
0x14, 0x29, 0x52, 0xA5, 0x4A, 0x95, 0x2A, 0x54,
|
|
0xA9, 0x53, 0xA7, 0x4E, 0x9D, 0x3B, 0x77, 0xEE,
|
|
0xDD, 0xBB, 0x76, 0xEC, 0xD9, 0xB3, 0x67, 0xCF,
|
|
0x9E, 0x3D, 0x7B, 0xF7, 0xEF, 0xDF, 0xBE, 0x7C,
|
|
0xF9, 0xF2, 0xE5, 0xCA, 0x94, 0x28, 0x51, 0xA3,
|
|
0x46, 0x8C, 0x18, 0x30, 0x60, 0xC1, 0x83, 0x07,
|
|
0x0E, 0x1D, 0x3A, 0x75, 0xEB, 0xD7, 0xAE, 0x5D,
|
|
0xBA, 0x74, 0xE9, 0xD3, 0xA6, 0x4C, 0x99, 0x33,
|
|
0x66, 0xCD, 0x9A, 0x35, 0x6A, 0xD4, 0xA8, 0x50,
|
|
0xA1, 0x42, 0x84, 0x09, 0x13, 0x27, 0x4F, 0x9F,
|
|
0x3F, 0x7F, 0xFF, 0xFE, 0xFC, 0xF8, 0xF0, 0xE1,
|
|
0xC2, 0x85, 0x0B, 0x17, 0x2F, 0x5E, 0xBC, 0x78,
|
|
0xF1, 0xE3, 0xC6, 0x8D, 0x1A, 0x34, 0x68, 0xD0,
|
|
])
|
|
|
|
|
|
class PHYEncode:
|
|
"""LoRa PHY layer encoder block.
|
|
|
|
Encodes payload bytes into CSS symbol bins.
|
|
"""
|
|
|
|
def __init__(self, sf: int = 9, cr: int = 1, has_crc: bool = True,
|
|
ldro: bool = False, implicit_header: bool = False):
|
|
"""Initialize PHY encoder.
|
|
|
|
Args:
|
|
sf: Spreading factor (7-12)
|
|
cr: Coding rate 1-4 (CR 4/5 through CR 4/8)
|
|
has_crc: Append CRC to payload
|
|
ldro: Low Data Rate Optimization
|
|
implicit_header: Omit header (for implicit mode)
|
|
"""
|
|
if not 7 <= sf <= 12:
|
|
raise ValueError(f"SF must be 7-12, got {sf}")
|
|
if not 1 <= cr <= 4:
|
|
raise ValueError(f"CR must be 1-4, got {cr}")
|
|
|
|
self.sf = sf
|
|
self.N = 1 << sf
|
|
self.cr = cr
|
|
self.has_crc = has_crc
|
|
self.ldro = ldro
|
|
self.implicit_header = implicit_header
|
|
|
|
def crc16(self, data: bytes) -> int:
|
|
"""Standard CRC-16/CCITT."""
|
|
crc = 0x0000
|
|
for byte in data:
|
|
crc ^= byte << 8
|
|
for _ in range(8):
|
|
if crc & 0x8000:
|
|
crc = (crc << 1) ^ 0x1021
|
|
else:
|
|
crc <<= 1
|
|
crc &= 0xFFFF
|
|
return crc
|
|
|
|
def whiten(self, nibbles: list[int], offset: int = 0) -> list[int]:
|
|
"""Apply whitening to payload nibbles."""
|
|
result = []
|
|
for i in range(0, len(nibbles) - 1, 2):
|
|
seq_byte = WHITEN_SEQ[(offset + i // 2) % len(WHITEN_SEQ)]
|
|
low = nibbles[i] ^ (seq_byte & 0x0F)
|
|
high = nibbles[i + 1] ^ ((seq_byte >> 4) & 0x0F)
|
|
result.append(low)
|
|
result.append(high)
|
|
if len(nibbles) % 2 == 1:
|
|
idx = len(nibbles) - 1
|
|
seq_byte = WHITEN_SEQ[(offset + idx // 2) % len(WHITEN_SEQ)]
|
|
result.append(nibbles[idx] ^ (seq_byte & 0x0F))
|
|
return result
|
|
|
|
def hamming_encode(self, nibble: int, cr: int) -> int:
|
|
"""Encode a 4-bit nibble with Hamming FEC.
|
|
|
|
Codeword layout (Semtech/gr-lora_sdr convention):
|
|
[d0 d1 d2 d3 p0 p1 p2 p_ext] — data at MSB, parity at LSB
|
|
"""
|
|
d0 = (nibble >> 3) & 1
|
|
d1 = (nibble >> 2) & 1
|
|
d2 = (nibble >> 1) & 1
|
|
d3 = nibble & 1
|
|
|
|
if cr == 4:
|
|
# Extended Hamming(8,4)
|
|
p0 = d0 ^ d1 ^ d2
|
|
p1 = d1 ^ d2 ^ d3
|
|
p2 = d0 ^ d1 ^ d3
|
|
p_ext = d0 ^ d1 ^ d2 ^ d3 ^ p0 ^ p1 ^ p2
|
|
return ((d0 << 7) | (d1 << 6) | (d2 << 5) | (d3 << 4) |
|
|
(p0 << 3) | (p1 << 2) | (p2 << 1) | p_ext)
|
|
|
|
elif cr == 3:
|
|
# Hamming(7,4)
|
|
p0 = d0 ^ d1 ^ d2
|
|
p1 = d1 ^ d2 ^ d3
|
|
p2 = d0 ^ d1 ^ d3
|
|
return ((d0 << 6) | (d1 << 5) | (d2 << 4) | (d3 << 3) |
|
|
(p0 << 2) | (p1 << 1) | p2)
|
|
|
|
elif cr == 2:
|
|
# Partial parity(6,4)
|
|
p0 = d0 ^ d1 ^ d2
|
|
p1 = d1 ^ d2 ^ d3
|
|
return ((d0 << 5) | (d1 << 4) | (d2 << 3) | (d3 << 2) |
|
|
(p0 << 1) | p1)
|
|
|
|
else:
|
|
# Even parity(5,4)
|
|
p0 = d0 ^ d1 ^ d2 ^ d3
|
|
return (d0 << 4) | (d1 << 3) | (d2 << 2) | (d3 << 1) | p0
|
|
|
|
def interleave(self, codewords: list[int], sf: int, cr: int,
|
|
reduced_rate: bool) -> list[int]:
|
|
"""Interleave codewords into symbols (inverse of deinterleave)."""
|
|
cw_len = cr + 4
|
|
sf_app = sf - 2 if reduced_rate else sf
|
|
|
|
symbols = []
|
|
|
|
for grp_start in range(0, len(codewords), sf_app):
|
|
grp = codewords[grp_start:grp_start + sf_app]
|
|
if len(grp) < sf_app:
|
|
grp = grp + [0] * (sf_app - len(grp))
|
|
|
|
# Build deinterleaved bit matrix
|
|
deinter = []
|
|
for cw in grp:
|
|
row = []
|
|
for bit_pos in range(cw_len - 1, -1, -1):
|
|
row.append((cw >> bit_pos) & 1)
|
|
deinter.append(row)
|
|
|
|
# Interleave: inverse of gr-lora_sdr deinterleave
|
|
inter = [[0] * sf_app for _ in range(cw_len)]
|
|
for i in range(cw_len):
|
|
for j in range(sf_app):
|
|
inter[i][j] = deinter[(i - j - 1) % sf_app][i]
|
|
|
|
# Read out symbols
|
|
for row in inter:
|
|
sym = 0
|
|
for bit in row:
|
|
sym = (sym << 1) | bit
|
|
symbols.append(sym)
|
|
|
|
return symbols
|
|
|
|
def gray_encode(self, symbols: list[int]) -> list[int]:
|
|
"""Binary to Gray code: g = b ^ (b >> 1)."""
|
|
return [s ^ (s >> 1) for s in symbols]
|
|
|
|
def _xor_bits(self, val: int, mask: int) -> int:
|
|
"""XOR bits of val selected by mask."""
|
|
x = val & mask
|
|
result = 0
|
|
while x:
|
|
result ^= x & 1
|
|
x >>= 1
|
|
return result
|
|
|
|
def make_header(self, payload_len: int, cr: int, has_crc: bool) -> list[int]:
|
|
"""Build 5 header nibbles with checksum."""
|
|
nib0 = (payload_len >> 4) & 0x0F
|
|
nib1 = payload_len & 0x0F
|
|
nib2 = ((cr & 0x07) << 1) | (1 if has_crc else 0)
|
|
|
|
d = (nib0 << 8) | (nib1 << 4) | nib2
|
|
c4 = self._xor_bits(d, 0b111110010000)
|
|
c3 = self._xor_bits(d, 0b110111000000)
|
|
c2 = self._xor_bits(d, 0b101100111000)
|
|
c1 = self._xor_bits(d, 0b100010100110)
|
|
c0 = self._xor_bits(d, 0b011001010101)
|
|
check = (c4 << 4) | (c3 << 3) | (c2 << 2) | (c1 << 1) | c0
|
|
nib3 = (check >> 4) & 0x01
|
|
nib4 = check & 0x0F
|
|
|
|
return [nib0, nib1, nib2, nib3, nib4]
|
|
|
|
def encode(self, payload: bytes) -> list[int]:
|
|
"""Encode payload bytes into CSS symbol bins.
|
|
|
|
Args:
|
|
payload: Raw payload bytes
|
|
|
|
Returns:
|
|
List of symbol bin values ready for CSS modulation
|
|
"""
|
|
sf = self.sf
|
|
cr = self.cr
|
|
has_crc = self.has_crc
|
|
|
|
# Payload to nibbles (low nibble first per byte)
|
|
payload_nibbles = []
|
|
for b in payload:
|
|
payload_nibbles.append(b & 0x0F)
|
|
payload_nibbles.append((b >> 4) & 0x0F)
|
|
|
|
# CRC
|
|
if has_crc:
|
|
crc_val = self.crc16(payload)
|
|
payload_nibbles.append(crc_val & 0x0F)
|
|
payload_nibbles.append((crc_val >> 4) & 0x0F)
|
|
payload_nibbles.append((crc_val >> 8) & 0x0F)
|
|
payload_nibbles.append((crc_val >> 12) & 0x0F)
|
|
|
|
# Whiten payload
|
|
whitened = self.whiten(payload_nibbles, offset=0)
|
|
|
|
# Header nibbles (not whitened)
|
|
header_nibs = self.make_header(len(payload), cr, has_crc)
|
|
|
|
# FEC encode
|
|
# Header block at reduced rate produces sf-2 codewords (all at CR 4/8):
|
|
# First 5 are header fields, remaining are first payload nibbles
|
|
n_extra = max(0, sf - 2 - 5)
|
|
header_cw = [self.hamming_encode(n, 4) for n in header_nibs]
|
|
extra_cw = [self.hamming_encode(n, 4) for n in whitened[:n_extra]]
|
|
header_block_cw = header_cw + extra_cw
|
|
|
|
# Remaining payload at specified CR
|
|
remaining_whitened = whitened[n_extra:]
|
|
payload_cw = [self.hamming_encode(n, cr) for n in remaining_whitened]
|
|
|
|
# Interleave
|
|
header_symbols = self.interleave(header_block_cw, sf, 4, reduced_rate=True)
|
|
payload_symbols = self.interleave(payload_cw, sf, cr,
|
|
reduced_rate=self.ldro)
|
|
|
|
# Gray encode
|
|
# Header uses reduced rate: multiply by 4 to place in upper bits
|
|
header_gray = [g * 4 for g in self.gray_encode(header_symbols)]
|
|
|
|
if self.ldro:
|
|
payload_gray = [g * 4 for g in self.gray_encode(payload_symbols)]
|
|
else:
|
|
payload_gray = self.gray_encode(payload_symbols)
|
|
|
|
return header_gray + payload_gray
|
|
|
|
|
|
def phy_encode(sf: int = 9, cr: int = 1, has_crc: bool = True,
|
|
ldro: bool = False, implicit_header: bool = False) -> PHYEncode:
|
|
"""Factory function for GNU Radio compatibility."""
|
|
return PHYEncode(sf=sf, cr=cr, has_crc=has_crc, ldro=ldro,
|
|
implicit_header=implicit_header)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
print("PHY Encode Test")
|
|
print("=" * 50)
|
|
|
|
# Test encoding
|
|
payload = b"Hello"
|
|
sf = 9
|
|
cr = 1
|
|
|
|
encoder = PHYEncode(sf=sf, cr=cr, has_crc=True)
|
|
bins = encoder.encode(payload)
|
|
|
|
print(f"\nEncoding '{payload.decode()}' (SF{sf}, CR4/{cr+4}):")
|
|
print(f" Payload: {len(payload)} bytes")
|
|
print(f" Symbols: {len(bins)} bins")
|
|
print(f" First 10 bins: {bins[:10]}")
|
|
|
|
# Verify round-trip with decoder
|
|
from .phy_decode import PHYDecode
|
|
decoder = PHYDecode(sf=sf, cr=cr, has_crc=True)
|
|
|
|
# For synthetic test, CFO is 0 and we need to add 1 back
|
|
# (decoder subtracts CFO + 1)
|
|
adjusted_bins = [(b + 1) % (1 << sf) for b in bins]
|
|
frame = decoder.decode(adjusted_bins, cfo_bin=0)
|
|
|
|
print(f"\nRound-trip decode:")
|
|
print(f" header_ok: {frame.header_ok}")
|
|
print(f" crc_ok: {frame.crc_ok}")
|
|
print(f" payload: {frame.payload!r}")
|
|
|
|
if frame.payload == payload and frame.crc_ok:
|
|
print("\n✓ Encode/decode round-trip OK")
|
|
else:
|
|
print("\n✗ Round-trip FAILED")
|