Ryan Malloy
ec0dfedc50
Implement precision timing recovery functions: - _refine_symbol_boundary(): Scans at 1/32-symbol resolution to find exact chirp boundary by maximizing dechirped SNR - _find_sfd_boundary(): FFT-based correlation with downchirp template to find exact data start position Bug fixes: - Fix _is_downchirp() false positives by comparing both correlations - Fix _estimate_cfo() to return values in [0, N) range The improved sync_from_samples() now produces bins identical to the reference lora_decode_gpu decoder.
625 lines
23 KiB
Python
625 lines
23 KiB
Python
"""LoRa frame synchronization block.
|
|
|
|
Detects preamble, extracts sync word (NETWORKID), locates SFD,
|
|
and outputs aligned data symbols.
|
|
|
|
Frame structure:
|
|
[Preamble: N upchirps at bin 0]
|
|
[Sync Word: 2 upchirps encoding NETWORKID nibbles]
|
|
[SFD: 2.25 downchirps]
|
|
[Data: encoded payload symbols]
|
|
"""
|
|
|
|
import numpy as np
|
|
from numpy.typing import NDArray
|
|
from dataclasses import dataclass
|
|
from enum import Enum, auto
|
|
from typing import Optional, Callable
|
|
|
|
from .networkid import networkid_from_symbols, sync_word_to_networkid
|
|
|
|
|
|
class FrameSyncState(Enum):
|
|
"""State machine states for frame synchronization."""
|
|
SEARCH = auto() # Searching for preamble
|
|
PREAMBLE = auto() # Tracking preamble chirps
|
|
SYNC_WORD = auto() # Capturing sync word symbols
|
|
SFD = auto() # Detecting SFD downchirps
|
|
DATA = auto() # Outputting data symbols
|
|
|
|
|
|
@dataclass
|
|
class SyncResult:
|
|
"""Result from frame synchronization."""
|
|
found: bool # Frame detected
|
|
networkid: int # Extracted NETWORKID
|
|
cfo_bin: float # Carrier frequency offset (bins)
|
|
data_symbols: list[int] # Aligned data symbol bins
|
|
preamble_count: int # Number of preamble symbols detected
|
|
sync_word_raw: tuple[int, int] # Raw sync word symbol values
|
|
|
|
|
|
@dataclass
|
|
class FrameSyncConfig:
|
|
"""Configuration for frame synchronizer."""
|
|
sf: int = 9 # Spreading factor
|
|
sample_rate: float = 250e3 # Input sample rate
|
|
bw: float = 125e3 # LoRa bandwidth
|
|
preamble_min: int = 4 # Minimum preamble symbols to detect
|
|
expected_networkid: Optional[int] = None # Filter by NETWORKID (None = any)
|
|
sfd_threshold: float = 0.5 # SFD detection threshold
|
|
|
|
|
|
class FrameSync:
|
|
"""Frame synchronization for LoRa signals.
|
|
|
|
Performs preamble detection, sync word extraction, SFD detection,
|
|
and symbol alignment for the receiver chain.
|
|
"""
|
|
|
|
def __init__(self, sf: int = 9, sample_rate: float = 250e3,
|
|
bw: float = 125e3, preamble_min: int = 4,
|
|
expected_networkid: Optional[int] = None):
|
|
"""Initialize frame synchronizer.
|
|
|
|
Args:
|
|
sf: Spreading factor (7-12)
|
|
sample_rate: Input sample rate in Hz
|
|
bw: LoRa signal bandwidth in Hz
|
|
preamble_min: Minimum preamble symbols to consider valid
|
|
expected_networkid: Only accept frames with this NETWORKID (None = all)
|
|
"""
|
|
self.config = FrameSyncConfig(
|
|
sf=sf, sample_rate=sample_rate, bw=bw,
|
|
preamble_min=preamble_min, expected_networkid=expected_networkid
|
|
)
|
|
self.sf = sf
|
|
self.N = 1 << sf
|
|
self.sps = int(self.N * sample_rate / bw)
|
|
|
|
# Generate reference chirps
|
|
n = np.arange(self.sps)
|
|
phase_up = 2 * np.pi * (n * n / (2 * self.sps))
|
|
self._upchirp = np.exp(1j * phase_up).astype(np.complex64)
|
|
self._downchirp = np.conj(self._upchirp)
|
|
|
|
# State
|
|
self.reset()
|
|
|
|
def reset(self):
|
|
"""Reset synchronizer state."""
|
|
self._state = FrameSyncState.SEARCH
|
|
self._preamble_bins = []
|
|
self._preamble_count = 0
|
|
self._sync_bins = []
|
|
self._data_bins = []
|
|
self._cfo_estimate = 0.0
|
|
self._sfd_count = 0 # Count SFD downchirps (need 2 full ones)
|
|
|
|
def _dechirp_and_peak(self, samples: NDArray[np.complex64],
|
|
use_downchirp: bool = False) -> tuple[int, float]:
|
|
"""Dechirp samples and find FFT peak.
|
|
|
|
Args:
|
|
samples: One symbol of IQ samples
|
|
use_downchirp: If True, detect downchirp (for SFD)
|
|
|
|
Returns:
|
|
Tuple of (peak_bin, peak_magnitude)
|
|
"""
|
|
if use_downchirp:
|
|
# For downchirp detection, multiply by upchirp
|
|
dechirped = samples[:self.sps] * self._upchirp
|
|
else:
|
|
# For upchirp detection, multiply by downchirp
|
|
dechirped = samples[:self.sps] * self._downchirp
|
|
|
|
spectrum = np.abs(np.fft.fft(dechirped, n=self.N))
|
|
peak_bin = int(np.argmax(spectrum))
|
|
peak_mag = spectrum[peak_bin] / np.mean(spectrum)
|
|
|
|
return peak_bin, peak_mag
|
|
|
|
def _is_preamble_chirp(self, peak_bin: int, peak_mag: float) -> bool:
|
|
"""Check if a chirp looks like part of the preamble.
|
|
|
|
Preamble chirps should have:
|
|
- Strong FFT peak (high SNR)
|
|
- Bin consistent with previous preamble chirps (if any)
|
|
|
|
Real SDR captures can have significant CFO (carrier frequency offset),
|
|
so the preamble bin can appear anywhere in 0..N-1, not just near 0.
|
|
The key insight is that preamble chirps have the SAME bin value
|
|
(modulo small noise) for many consecutive symbols.
|
|
"""
|
|
if peak_mag < 3.0: # Minimum SNR threshold
|
|
return False
|
|
|
|
# If we have a CFO estimate, check against it
|
|
if self._preamble_count > 0:
|
|
expected_bin = int(round(self._cfo_estimate)) % self.N
|
|
# Tight tolerance: must be within 3 bins of expected
|
|
# This distinguishes preamble (bin ~0+CFO) from sync word (bin >= 8+CFO)
|
|
tolerance = 3
|
|
distance = min(abs(peak_bin - expected_bin),
|
|
self.N - abs(peak_bin - expected_bin))
|
|
return distance <= tolerance
|
|
else:
|
|
# First preamble chirp - accept ANY strong signal
|
|
# Real captures have arbitrary CFO, so preamble can appear at any bin
|
|
# We'll validate by checking if subsequent chirps have the same bin
|
|
return True
|
|
|
|
def _is_downchirp(self, samples: NDArray[np.complex64]) -> tuple[bool, float]:
|
|
"""Detect if samples contain a downchirp (SFD).
|
|
|
|
To distinguish downchirps from upchirps with high bin values (near N),
|
|
we compare both correlations. A true downchirp will have stronger
|
|
correlation with the upchirp reference than with the downchirp reference.
|
|
|
|
Returns:
|
|
Tuple of (is_downchirp, peak_magnitude)
|
|
"""
|
|
seg = samples[:self.sps]
|
|
|
|
# Correlation with upchirp (detects downchirps)
|
|
dc_dechirped = seg * self._upchirp
|
|
dc_spectrum = np.abs(np.fft.fft(dc_dechirped, n=self.N))
|
|
dc_peak_mag = np.max(dc_spectrum) / np.mean(dc_spectrum)
|
|
|
|
# Correlation with downchirp (detects upchirps)
|
|
uc_dechirped = seg * self._downchirp
|
|
uc_spectrum = np.abs(np.fft.fft(uc_dechirped, n=self.N))
|
|
uc_peak_mag = np.max(uc_spectrum) / np.mean(uc_spectrum)
|
|
|
|
# True downchirp: dc correlation >> upchirp correlation
|
|
# Upchirp with high bin: both correlations strong, but uc > dc
|
|
is_dc = dc_peak_mag > 5.0 and dc_peak_mag > uc_peak_mag * 1.5
|
|
|
|
return is_dc, dc_peak_mag
|
|
|
|
def _estimate_cfo(self) -> float:
|
|
"""Estimate CFO from preamble bin measurements."""
|
|
if not self._preamble_bins:
|
|
return 0.0
|
|
|
|
# Average the preamble bin values
|
|
bins = np.array(self._preamble_bins)
|
|
|
|
# Handle wraparound (bins near N-1 and 0 are close)
|
|
# Convert to complex unit vectors and average
|
|
angles = 2 * np.pi * bins / self.N
|
|
avg_angle = np.angle(np.mean(np.exp(1j * angles)))
|
|
cfo = avg_angle * self.N / (2 * np.pi)
|
|
|
|
# Ensure result is in [0, N) range
|
|
if cfo < 0:
|
|
cfo += self.N
|
|
return cfo
|
|
|
|
def _refine_symbol_boundary(self, samples: NDArray[np.complex64],
|
|
coarse_start: int, preamble_len: int) -> tuple[int, int]:
|
|
"""Find true chirp boundary by scanning for max dechirp SNR.
|
|
|
|
The coarse preamble start is grid-aligned to symbol boundaries.
|
|
The actual chirp boundary can be anywhere within that window.
|
|
Scans at 1/32-symbol resolution, averaging over multiple preamble symbols.
|
|
|
|
Args:
|
|
samples: IQ samples containing the frame
|
|
coarse_start: Approximate sample where preamble starts
|
|
preamble_len: Number of preamble symbols detected
|
|
|
|
Returns:
|
|
(refined_start, true_bin): Best sample offset and the preamble bin
|
|
measured at that alignment.
|
|
"""
|
|
sps = self.sps
|
|
N = self.N
|
|
|
|
# Number of preamble symbols to average (use middle ones, skip first/last)
|
|
n_avg = min(preamble_len - 2, 6)
|
|
if n_avg < 2:
|
|
# Not enough preamble — fall back to coarse measurement
|
|
end = coarse_start + sps
|
|
if end > len(samples):
|
|
return coarse_start, 0
|
|
seg = samples[coarse_start:end]
|
|
dechirped = seg * self._downchirp
|
|
spec = np.abs(np.fft.fft(dechirped, n=N)) ** 2
|
|
return coarse_start, int(np.argmax(spec))
|
|
|
|
# Scan offsets: 0 to sps-1 at step = sps/32
|
|
step = max(1, sps // 32)
|
|
offsets = range(0, sps, step)
|
|
|
|
best_snr = -np.inf
|
|
best_offset = 0
|
|
best_bin = 0
|
|
|
|
for off in offsets:
|
|
start = coarse_start + off
|
|
snr_sum = 0.0
|
|
bin_sum = 0
|
|
count = 0
|
|
|
|
# Average over middle preamble symbols (skip first one)
|
|
for k in range(1, 1 + n_avg):
|
|
seg_start = start + k * sps
|
|
if seg_start + sps > len(samples):
|
|
break
|
|
seg = samples[seg_start:seg_start + sps]
|
|
dechirped = seg * self._downchirp
|
|
spec = np.abs(np.fft.fft(dechirped, n=N)) ** 2
|
|
|
|
pk = int(np.argmax(spec))
|
|
pk_power = spec[pk]
|
|
|
|
# Noise: mean of all bins except ±2 around peak
|
|
mask = np.ones(N, dtype=bool)
|
|
mask[max(0, pk - 2):min(N, pk + 3)] = False
|
|
noise = np.mean(spec[mask])
|
|
|
|
if noise > 0:
|
|
snr_sum += 10 * np.log10(pk_power / noise)
|
|
bin_sum += pk
|
|
count += 1
|
|
|
|
if count > 0:
|
|
avg_snr = snr_sum / count
|
|
if avg_snr > best_snr:
|
|
best_snr = avg_snr
|
|
best_offset = off
|
|
best_bin = round(bin_sum / count)
|
|
|
|
# Fine-tune: scan ±step around best at 1-sample resolution
|
|
fine_start = max(0, best_offset - step)
|
|
fine_end = min(sps, best_offset + step + 1)
|
|
|
|
for off in range(fine_start, fine_end):
|
|
if off == best_offset:
|
|
continue
|
|
start = coarse_start + off
|
|
snr_sum = 0.0
|
|
bin_sum = 0
|
|
count = 0
|
|
|
|
for k in range(1, 1 + n_avg):
|
|
seg_start = start + k * sps
|
|
if seg_start + sps > len(samples):
|
|
break
|
|
seg = samples[seg_start:seg_start + sps]
|
|
dechirped = seg * self._downchirp
|
|
spec = np.abs(np.fft.fft(dechirped, n=N)) ** 2
|
|
|
|
pk = int(np.argmax(spec))
|
|
pk_power = spec[pk]
|
|
mask = np.ones(N, dtype=bool)
|
|
mask[max(0, pk - 2):min(N, pk + 3)] = False
|
|
noise = np.mean(spec[mask])
|
|
|
|
if noise > 0:
|
|
snr_sum += 10 * np.log10(pk_power / noise)
|
|
bin_sum += pk
|
|
count += 1
|
|
|
|
if count > 0:
|
|
avg_snr = snr_sum / count
|
|
if avg_snr > best_snr:
|
|
best_snr = avg_snr
|
|
best_offset = off
|
|
best_bin = round(bin_sum / count)
|
|
|
|
return coarse_start + best_offset, best_bin
|
|
|
|
def _find_sfd_boundary(self, samples: NDArray[np.complex64],
|
|
search_start: int, search_len: int) -> Optional[int]:
|
|
"""Find exact data-start sample using SFD downchirp correlation.
|
|
|
|
The SFD is 2.25 downchirps immediately preceding data. We correlate
|
|
a one-symbol downchirp template against the expected SFD region and
|
|
find the correlation peak. The peak location plus 2.25 * sps gives
|
|
the exact sample where data begins.
|
|
|
|
Args:
|
|
samples: IQ samples containing the frame
|
|
search_start: Sample position to start searching
|
|
search_len: Number of samples to search
|
|
|
|
Returns:
|
|
data_start sample position, or None if detection fails
|
|
"""
|
|
sps = self.sps
|
|
|
|
# For correlation to find downchirps, we use the downchirp as template
|
|
# _downchirp is conj(_upchirp), which IS the downchirp
|
|
downchirp_template = self._downchirp
|
|
|
|
# Correlation: slide downchirp across the search region
|
|
search_end = min(search_start + search_len, len(samples) - sps)
|
|
if search_start < 0 or search_start >= search_end:
|
|
return None
|
|
|
|
seg_len = search_end - search_start
|
|
if seg_len <= sps:
|
|
return None
|
|
|
|
segment = samples[search_start:search_start + seg_len]
|
|
|
|
# Pad template to segment length for FFT correlation
|
|
padded_template = np.zeros(seg_len, dtype=np.complex64)
|
|
padded_template[:sps] = downchirp_template
|
|
|
|
# FFT-based correlation: corr[k] = sum(segment[n] * conj(template[n-k]))
|
|
# Peak indicates where template best matches the signal
|
|
corr = np.abs(np.fft.ifft(
|
|
np.fft.fft(segment) * np.conj(np.fft.fft(padded_template))
|
|
))
|
|
|
|
# Peak = start of the first full SFD downchirp symbol
|
|
peak_offset = int(np.argmax(corr))
|
|
sfd_start = search_start + peak_offset
|
|
|
|
# Data starts 2.25 symbols after SFD start
|
|
data_start = sfd_start + int(2.25 * sps)
|
|
return data_start
|
|
|
|
def process_symbol(self, samples: NDArray[np.complex64]) -> Optional[SyncResult]:
|
|
"""Process one symbol's worth of samples.
|
|
|
|
This is a streaming interface - call repeatedly with each symbol.
|
|
|
|
Args:
|
|
samples: Complex IQ samples (length >= sps)
|
|
|
|
Returns:
|
|
SyncResult when a complete frame is detected, None otherwise
|
|
"""
|
|
if len(samples) < self.sps:
|
|
return None
|
|
|
|
peak_bin, peak_mag = self._dechirp_and_peak(samples)
|
|
|
|
if self._state == FrameSyncState.SEARCH:
|
|
# Looking for preamble start
|
|
if self._is_preamble_chirp(peak_bin, peak_mag):
|
|
self._preamble_bins.append(peak_bin)
|
|
self._preamble_count = 1
|
|
self._cfo_estimate = peak_bin
|
|
self._state = FrameSyncState.PREAMBLE
|
|
|
|
elif self._state == FrameSyncState.PREAMBLE:
|
|
# Tracking preamble
|
|
if self._is_preamble_chirp(peak_bin, peak_mag):
|
|
self._preamble_bins.append(peak_bin)
|
|
self._preamble_count += 1
|
|
self._cfo_estimate = self._estimate_cfo()
|
|
else:
|
|
# Preamble ended - check if we have enough
|
|
if self._preamble_count >= self.config.preamble_min:
|
|
# This symbol is first sync word
|
|
self._sync_bins = [peak_bin]
|
|
self._state = FrameSyncState.SYNC_WORD
|
|
else:
|
|
# False alarm, reset
|
|
self.reset()
|
|
|
|
elif self._state == FrameSyncState.SYNC_WORD:
|
|
# Capturing sync word (2 symbols)
|
|
self._sync_bins.append(peak_bin)
|
|
if len(self._sync_bins) >= 2:
|
|
self._state = FrameSyncState.SFD
|
|
|
|
elif self._state == FrameSyncState.SFD:
|
|
# Detecting SFD downchirps (2 full + 0.25 fractional)
|
|
is_dc, _ = self._is_downchirp(samples)
|
|
if is_dc:
|
|
self._sfd_count += 1
|
|
# After 2 full downchirps, transition to DATA
|
|
# The 0.25 fractional downchirp is handled in sync_from_samples
|
|
if self._sfd_count >= 2:
|
|
self._state = FrameSyncState.DATA
|
|
else:
|
|
# Not a downchirp after expecting SFD - could be data already
|
|
# (This handles cases where SFD detection fails)
|
|
self._data_bins.append(peak_bin)
|
|
self._state = FrameSyncState.DATA
|
|
|
|
elif self._state == FrameSyncState.DATA:
|
|
self._data_bins.append(peak_bin)
|
|
|
|
return None
|
|
|
|
def sync_from_samples(self, samples: NDArray[np.complex64],
|
|
max_data_symbols: int = 100) -> SyncResult:
|
|
"""Synchronize and extract frame from a block of samples.
|
|
|
|
Timing recovery strategy:
|
|
1. State machine finds coarse preamble position
|
|
2. _refine_symbol_boundary() scans at 1/32-symbol resolution for exact boundary
|
|
3. _find_sfd_boundary() uses FFT correlation to find exact data start
|
|
|
|
Args:
|
|
samples: Complex IQ samples containing a LoRa frame
|
|
max_data_symbols: Maximum data symbols to extract
|
|
|
|
Returns:
|
|
SyncResult with extracted frame data
|
|
"""
|
|
self.reset()
|
|
|
|
sps = self.sps
|
|
n_symbols = len(samples) // sps
|
|
preamble_start_symbol = None
|
|
sfd_start_symbol = None
|
|
|
|
# Phase 1: Find frame structure using state machine (coarse detection)
|
|
for i in range(n_symbols):
|
|
symbol_samples = samples[i * sps:(i + 1) * sps]
|
|
|
|
prev_state = self._state
|
|
self.process_symbol(symbol_samples)
|
|
|
|
# Record when preamble starts
|
|
if prev_state == FrameSyncState.SEARCH and self._state == FrameSyncState.PREAMBLE:
|
|
preamble_start_symbol = i
|
|
|
|
# Record when we enter SFD state (after sync word)
|
|
if prev_state == FrameSyncState.SYNC_WORD and self._state == FrameSyncState.SFD:
|
|
sfd_start_symbol = i
|
|
|
|
# Stop when we enter DATA state
|
|
if self._state == FrameSyncState.DATA:
|
|
break
|
|
|
|
# Phase 2: Refine timing with sub-sample precision
|
|
if preamble_start_symbol is not None and self._preamble_count >= self.config.preamble_min:
|
|
# Clear any bins captured during state machine (they're grid-aligned)
|
|
self._data_bins = []
|
|
|
|
# Step 2a: Refine preamble boundary at 1/32-symbol resolution
|
|
coarse_start = preamble_start_symbol * sps
|
|
refined_start, true_bin = self._refine_symbol_boundary(
|
|
samples, coarse_start, self._preamble_count
|
|
)
|
|
|
|
# Update CFO estimate with the refined measurement
|
|
self._cfo_estimate = float(true_bin)
|
|
|
|
# Step 2b: Find SFD boundary using FFT correlation
|
|
# SFD starts after preamble + 2 sync word symbols
|
|
# Add 2 extra symbols buffer to ensure we're past the sync word
|
|
# (preamble_count may slightly undercount the actual preamble length)
|
|
sfd_search_start = refined_start + int((self._preamble_count + 3) * sps)
|
|
sfd_search_len = 4 * sps # 4-symbol search window
|
|
|
|
data_start = self._find_sfd_boundary(samples, sfd_search_start, sfd_search_len)
|
|
|
|
# Apply timing fine-tune: the SFD correlation may have slight offset
|
|
# due to symbol boundary not being perfectly aligned. Add a small
|
|
# correction to improve bin accuracy (empirically ~25 samples at BW rate)
|
|
if data_start is not None:
|
|
timing_correction = sps // 20 # ~5% of symbol
|
|
data_start += timing_correction
|
|
|
|
if data_start is None:
|
|
# Fallback: use fixed offset from sync word end
|
|
# sync word is 2 symbols after preamble, SFD is 2.25 after that
|
|
if sfd_start_symbol is not None:
|
|
data_start = refined_start + (self._preamble_count + 2 + 2) * sps + sps // 4
|
|
else:
|
|
# Last resort: estimate from preamble length
|
|
data_start = refined_start + (self._preamble_count + 4) * sps + sps // 4
|
|
|
|
# Step 2c: Extract data symbols from the refined position
|
|
for i in range(max_data_symbols):
|
|
start = data_start + i * sps
|
|
end = start + sps
|
|
if end > len(samples):
|
|
break
|
|
symbol_samples = samples[start:end]
|
|
peak_bin, peak_mag = self._dechirp_and_peak(symbol_samples)
|
|
self._data_bins.append(peak_bin)
|
|
|
|
# Build result
|
|
found = len(self._data_bins) > 0 and self._preamble_count >= self.config.preamble_min
|
|
|
|
# Extract NETWORKID from sync word
|
|
sync_raw = (0, 0)
|
|
networkid = 0
|
|
if len(self._sync_bins) >= 2:
|
|
# Sync word bins are relative to preamble bin
|
|
cfo_int = int(round(self._cfo_estimate))
|
|
sync_raw = (self._sync_bins[0], self._sync_bins[1])
|
|
# Convert to deltas from preamble
|
|
d1 = (self._sync_bins[0] - cfo_int) % self.N
|
|
d2 = (self._sync_bins[1] - cfo_int) % self.N
|
|
networkid = sync_word_to_networkid((d1, d2))
|
|
|
|
# Filter by expected NETWORKID if configured
|
|
if found and self.config.expected_networkid is not None:
|
|
if networkid != self.config.expected_networkid:
|
|
found = False
|
|
|
|
return SyncResult(
|
|
found=found,
|
|
networkid=networkid,
|
|
cfo_bin=self._cfo_estimate,
|
|
data_symbols=list(self._data_bins),
|
|
preamble_count=self._preamble_count,
|
|
sync_word_raw=sync_raw,
|
|
)
|
|
|
|
|
|
def frame_sync(sf: int = 9, sample_rate: float = 250e3,
|
|
expected_networkid: Optional[int] = None) -> FrameSync:
|
|
"""Factory function for GNU Radio compatibility."""
|
|
return FrameSync(sf=sf, sample_rate=sample_rate,
|
|
expected_networkid=expected_networkid)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
print("Frame Sync Test")
|
|
print("=" * 50)
|
|
|
|
# Generate a test frame
|
|
sf = 9
|
|
N = 1 << sf
|
|
fs = 125e3
|
|
|
|
# Simple chirp generation
|
|
n = np.arange(N)
|
|
|
|
def upchirp(f_start: int) -> np.ndarray:
|
|
phase = 2 * np.pi * ((f_start * n / N) + (n * n / (2 * N)))
|
|
return np.exp(1j * phase).astype(np.complex64)
|
|
|
|
def downchirp() -> np.ndarray:
|
|
return np.conj(upchirp(0))
|
|
|
|
# Build test frame
|
|
# Preamble (8 upchirps at bin 0)
|
|
preamble = np.tile(upchirp(0), 8)
|
|
|
|
# Sync word for NETWORKID=18 (0x12): nibbles 1, 2 → bins 32, 64
|
|
sync_nibble_hi = 1
|
|
sync_nibble_lo = 2
|
|
sync_bin_1 = sync_nibble_hi * (N // 16) # 32
|
|
sync_bin_2 = sync_nibble_lo * (N // 16) # 64
|
|
sync_word = np.concatenate([upchirp(sync_bin_1), upchirp(sync_bin_2)])
|
|
|
|
# SFD (2.25 downchirps)
|
|
dc = downchirp()
|
|
sfd = np.concatenate([dc, dc, dc[:N // 4]])
|
|
|
|
# Data symbols
|
|
data_bins = [100, 200, 300, 400, 500]
|
|
data = np.concatenate([upchirp(b) for b in data_bins])
|
|
|
|
# Complete frame
|
|
frame = np.concatenate([preamble, sync_word, sfd, data])
|
|
|
|
print(f"\nTest frame (SF{sf}):")
|
|
print(f" Preamble: 8 symbols")
|
|
print(f" Sync word: bins [{sync_bin_1}, {sync_bin_2}] → NETWORKID=18")
|
|
print(f" SFD: 2.25 downchirps")
|
|
print(f" Data: {len(data_bins)} symbols, bins {data_bins}")
|
|
print(f" Total: {len(frame)} samples")
|
|
|
|
# Test synchronizer
|
|
sync = FrameSync(sf=sf, sample_rate=fs)
|
|
result = sync.sync_from_samples(frame)
|
|
|
|
print(f"\nSync result:")
|
|
print(f" found: {result.found}")
|
|
print(f" networkid: {result.networkid}")
|
|
print(f" cfo_bin: {result.cfo_bin:.2f}")
|
|
print(f" preamble_count: {result.preamble_count}")
|
|
print(f" sync_word_raw: {result.sync_word_raw}")
|
|
print(f" data_symbols: {result.data_symbols}")
|
|
|
|
if result.found and result.networkid == 18:
|
|
print("\n✓ Frame sync OK")
|
|
else:
|
|
print("\n✗ Frame sync FAILED")
|