class LoRaDecoder:
"""LoRa packet decoder.
Combines synchronization, symbol demodulation, and payload decoding. The
decoder is constructed once with the radio/tunable parameters, then used
in one of several ways:
* ``decode(iq)`` -- decode one packet in an IQ buffer
* ``decode_file(path)``-- decode all packets in an IQ recording
* ``decode_iq(iq)`` -- decode all packets in an in-memory buffer
* ``decode_stream(chunk)`` + ``flush()`` -- feed live IQ chunks
``decode`` dispatches on ``settings.sync_algorithm``: ``'xhonneux'`` runs
the Xhonneux et al. 2021 3-stage sync at the Nyquist rate (resampling
``fs -> bw`` internally); ``'gr-lora-sdr'`` runs the gr-lora_sdr
``frame_sync`` port at the native rate (no resampling).
Parameters
----------
sf : int
Spreading factor (7-12).
bw : float
Bandwidth (Hz).
fs : float
Sampling rate (Hz) of the input IQ.
fc : float
Center (carrier) frequency (Hz).
preamble_len : int
Number of preamble upchirps.
implicit_header : bool
True = implicit header (no header transmitted), False = explicit.
crc_enabled : bool
Payload CRC present. Read from the header in explicit mode; forced
in implicit mode.
code_rate : int
Code rate (1-4 -> 4/5..4/8). Read from the header in explicit mode;
forced in implicit mode.
sync_word : int
LoRa sync word (used by the gr-lora-sdr sync).
payload_len : int or None
Payload length in bytes; required when ``implicit_header`` is True.
settings : DecoderSettings or None
Internal/advanced parameters (sync algorithm, decode mode, gates);
see :class:`DecoderSettings`.
"""
def __init__(self, sf, bw, fs, fc,
preamble_len=8, implicit_header=False,
crc_enabled=True, code_rate=1,
sync_word=0x12, payload_len=None,
settings=None):
"""LoRa packet decoder.
Parameters
----------
sf : int
Spreading factor (7-12).
bw : float
Bandwidth (Hz).
fs : float
Sampling frequency (Hz) of the input IQ.
fc : float
Center frequency (Hz).
preamble_len : int
Number of preamble upchirps (default 8).
implicit_header : bool
True = implicit header (no header transmitted), False = explicit.
crc_enabled : bool
Payload CRC present. Read from the header in explicit mode;
forced in implicit mode.
code_rate : int
Code rate (1-4 -> 4/5..4/8). Read from the header in explicit
mode; forced in implicit mode.
sync_word : int
LoRa sync word (default 0x12); used by the gr-lora-sdr sync.
payload_len : int or None
Payload length in bytes; required when ``implicit_header`` is
True (no header to carry it).
settings : DecoderSettings or None
Internal/advanced parameters (sync algorithm, gates, Chase...).
"""
if sf < 7 or sf > 12:
raise ValueError(f'Invalid sf={sf}: must be in 7..12')
if bw <= 0:
raise ValueError(f'Invalid bw={bw}: must be positive')
if fs <= 0:
raise ValueError(f'Invalid fs={fs}: must be positive')
if fc <= 0:
raise ValueError(f'Invalid fc={fc}: must be positive')
if implicit_header and payload_len is None:
raise ValueError('payload_len is required for implicit header')
if settings is not None and \
settings.sync_algorithm not in ('xhonneux', 'gr-lora-sdr'):
raise ValueError(
f'Invalid sync_algorithm={settings.sync_algorithm!r}: '
"must be 'xhonneux' or 'gr-lora-sdr'")
if settings is not None and \
settings.decode_mode not in ('hard', 'chase'):
raise ValueError(
f'Invalid decode_mode={settings.decode_mode!r}: '
"must be 'hard' or 'chase'")
self.sf = sf
self.bw = bw
self.fs = fs
self.fc = fc
self.preamble_len = preamble_len
self.implicit_header = bool(implicit_header)
self.crc_enabled = bool(crc_enabled)
self.code_rate = int(code_rate)
self.sync_word = sync_word
self.payload_len = payload_len
self.settings = settings if settings is not None else DecoderSettings()
self.N = 2 ** sf
self.upchirp, self.downchirp = generate_chirps(self.N)
self._stream_active = False
@property
def preamble_syms(self):
return (self.preamble_len + self.settings.N_netid
+ self.settings.N_sfd_down)
# =====================================================================
# Offline decoding
# =====================================================================
def estimate_carrier_offset(self, source, chunk_samples=2 ** 16):
"""Estimate the coarse carrier offset of a recording, in Hz.
Scans the native-rate stream for a preamble run (consecutive symbol
windows dechirping to the same bin), then resolves the offset from the
preamble upchirp and the first SFD downchirp:
f_up = f_offset + f_STO (upchirp dechirped with downchirp)
f_down = f_offset - f_STO (downchirp dechirped with upchirp)
f_offset = (f_up + f_down) / 2
The timing term cancels, which a plain dechirp peak cannot do (it is
biased by up to a full bandwidth by the STO). Working at ``fs`` rather
than ``bw`` makes the estimate unambiguous over the whole +-fs/2 span
instead of +-bw/2.
Parameters
----------
source : str, os.PathLike or ndarray
Recording path or an in-memory complex baseband buffer at ``fs``.
chunk_samples : int
Read granularity when ``source`` is a path.
Returns
-------
float or None
Offset in Hz (subtract it via ``carrier_offset_hz``), or None when
no preamble was found.
"""
from scipy.signal import resample_poly
up, down = generate_chirps(self.N)
q = int(round(self.fs / self.bw))
NSYM = self.N * q
down_up = resample_poly(down.astype(np.complex128), q, 1)
up_up = resample_poly(up.astype(np.complex128), q, 1)
if isinstance(source, (str, bytes)) or hasattr(source, '__fspath__'):
chunks = iter_iq_chunks(source, chunk_samples)
else:
chunks = [np.asarray(source, dtype=np.complex128)]
def bin_to_hz(k):
return (k - NSYM if k > NSYM / 2 else k) * self.fs / NSYM
sfd_at = self.preamble_len + self.settings.N_netid
min_run = min(4, self.preamble_len)
cands = []
run_len = 0
run_bins = []
run_ratios = []
since_start = 0
idx = 0
buf = np.zeros(0, dtype=np.complex128)
for chunk in chunks:
buf = np.concatenate([buf, np.asarray(chunk, dtype=np.complex128)])
while len(buf) >= NSYM:
w = buf[:NSYM]
buf = buf[NSYM:]
Y = np.abs(np.fft.fft(w * down_up))
b = int(np.argmax(Y))
r = float(Y[b] / (np.median(Y) + 1e-12))
strong = r >= self.settings.strong_ratio
if run_len and since_start == sfd_at and run_len >= min_run:
# This window is D1: dechirp it with the upchirp so the
# timing term cancels against the upchirp estimate.
b_dn = int(np.argmax(np.abs(np.fft.fft(w * up_up))))
b_up = int(np.median(run_bins))
cands.append((run_len, float(np.mean(run_ratios)),
(bin_to_hz(b_up) + bin_to_hz(b_dn)) / 2.0))
run_len = 0
if run_len and strong and abs(b - run_bins[-1]) <= q:
run_bins.append(b)
run_ratios.append(r)
run_len += 1
elif run_len and since_start < sfd_at:
pass # inside the net-id/SFD gap; keep waiting for D1
elif strong:
run_len = 1
run_bins = [b]
run_ratios = [r]
since_start = 0
else:
run_len = 0
if run_len:
since_start += 1
idx += 1
if not cands:
return None
# Several preambles (and the odd noise run) yield several estimates.
# A real signal has them all within a fraction of a bandwidth of each
# other; noise runs scatter. Keep the largest cluster and take its
# median rather than trusting the longest single run.
tol = self.bw / 8.0
best = None
for _rl, _r, f0 in cands:
group = [c for c in cands if abs(c[2] - f0) <= tol]
key = (len(group), sum(c[1] for c in group))
if best is None or key > best[0]:
best = (key, group)
group = best[1]
return float(np.median([c[2] for c in group]))
def decode_file(self, path, carrier_offset_hz=0.0, chunk_samples=2 ** 16):
"""Decode every packet in an IQ recording.
The file is read in chunks (see :func:`softlora.io.iter_iq_chunks`).
Both sync algorithms stream with bounded memory: ``xhonneux`` decimates
``fs -> bw`` and runs the streaming decoder; ``gr-lora-sdr`` scans the
native-rate stream and decodes each detected frame.
Returns
-------
list of Packet
One Packet per successfully decoded packet. Candidates whose
frame never synchronized (e.g. false-positive preamble detections
on noise) are dropped.
"""
if carrier_offset_hz == 'auto':
carrier_offset_hz = self.estimate_carrier_offset(
path, chunk_samples) or 0.0
if self.settings.sync_algorithm == 'gr-lora-sdr':
return self._decode_gr_frames(iter_iq_chunks(path, chunk_samples),
carrier_offset_hz)
self.reset()
self._stream_init()
found = []
pos = 0
for chunk in iter_iq_chunks(path, chunk_samples):
if carrier_offset_hz:
# Rotate with an absolute sample index so the phase is
# continuous across chunk boundaries.
n = np.arange(pos, pos + len(chunk))
chunk = np.asarray(chunk, dtype=np.complex128) * np.exp(
-1j * 2 * np.pi * carrier_offset_hz * n / self.fs)
pos += len(chunk)
chunk = self._resampler.process(chunk)
if len(chunk):
self._buf = np.concatenate([
self._buf, np.asarray(chunk, dtype=np.complex64)
])
found += self._scan(flush=False)
found += self._scan(flush=True)
self.reset()
return found
def decode_iq(self, iq, carrier_offset_hz=0.0):
"""Scan an in-memory IQ buffer and decode every packet in it.
Dispatches on the configured sync algorithm: ``xhonneux`` uses the
streaming decimator (``fs -> bw``); ``gr-lora-sdr`` scans the buffer at
the native rate. Positions in the returned Packets are absolute
within ``iq``.
Returns
-------
list of Packet
"""
if carrier_offset_hz == 'auto':
carrier_offset_hz = self.estimate_carrier_offset(iq) or 0.0
if self.settings.sync_algorithm == 'gr-lora-sdr':
return self._decode_gr_frames(
[np.asarray(iq, dtype=np.complex128)], carrier_offset_hz)
self.reset()
self._stream_init()
iq = np.asarray(iq, dtype=np.complex128)
if carrier_offset_hz:
n = np.arange(len(iq))
iq = iq * np.exp(-1j * 2 * np.pi * carrier_offset_hz * n / self.fs)
rs = self._resampler.process(iq)
self._buf = np.asarray(rs, dtype=np.complex64)
found = self._scan(flush=True)
self.reset()
return found
def _decode_gr_frames(self, chunks, carrier_offset_hz=0.0):
"""Stream IQ chunks and decode every frame with gr-lora-sdr.
``chunks`` is an iterable of complex baseband chunks at ``fs``.
Runs the dechirp preamble run-detection incrementally over the stream
with bounded memory (via :class:`_GrFrameStream`): each detected run is
decoded from a buffered window, then the scanned prefix is dropped.
Returns a list of Packets, one per detected frame. Runs whose frame
never synchronizes (false positives from the coarse scan) are not
returned.
"""
scanner = _GrFrameStream(self, carrier_offset_hz=carrier_offset_hz)
pkts = []
for chunk in chunks:
pkts += scanner.process(chunk)
pkts += scanner.flush()
return pkts
def decode(self, iq, carrier_offset_hz=0.0, max_payload_bytes=255):
"""Decode one packet with the configured synchronization algorithm.
Dispatches on ``settings.sync_algorithm``:
* ``'xhonneux'`` -- the Xhonneux paper 3-stage sync (Nyquist rate;
resamples ``fs -> bw`` internally).
* ``'gr-lora-sdr'`` -- the gr-lora_sdr ``frame_sync`` port (native
rate ``fs``; no resampling).
Parameters
----------
iq : ndarray
Complex baseband at ``fs`` (the whole recording or a window).
carrier_offset_hz : float
Optional coarse carrier offset applied to ``iq`` before sync.
max_payload_bytes : int
Only used by the gr-lora-sdr sync (symbols output before the
header feedback is known).
Returns
-------
Packet
``Packet.ok`` is True when the payload decoded without error.
"""
if self.settings.sync_algorithm == 'gr-lora-sdr':
return self._decode_gr(iq, carrier_offset_hz=carrier_offset_hz,
max_payload_bytes=max_payload_bytes)
return self._decode_xhonneux(iq, resample=True,
carrier_offset_hz=carrier_offset_hz)
def _align_score(self, sig, start, num_syms=16):
"""Peak-to-median sum of the first dechirped symbols at ``start``.
At fs = BW a one-sample timing error is a one-bin symbol error, so the
sub-sample residual dropped when ``payload_start`` is rounded matters.
This scores candidate sample alignments so the sharpest is kept.
"""
total = 0.0
for k in range(num_syms):
w = sig[start + k * self.N:start + (k + 1) * self.N]
if len(w) < self.N:
break
Y = np.abs(np.fft.fft(w * self.downchirp)) ** 2
total += float(Y.max() / (np.median(Y) + 1e-30))
return total
def _demod_at(self, sig, start_real, num_syms, pad=64):
"""Demodulate ``num_syms`` symbols starting at a fractional sample.
The Eq. 20 fractional-STO estimate can be off by up to ~0.5 sample,
and at fs = BW that is half a bin of timing error on every symbol.
Sampling the corrected signal at a fractional start lets the decoder
probe those alignments.
"""
i0 = int(np.floor(start_real))
fr = start_real - i0
if i0 < 0:
return np.zeros(0, dtype=int)
lo = max(0, i0 - pad)
hi = min(len(sig), i0 + num_syms * self.N + pad)
seg = np.asarray(sig[lo:hi], dtype=complex)
if len(seg) < self.N:
return np.zeros(0, dtype=int)
seg = _frac_advance(seg, fr)
return self.demodulate(seg[i0 - lo:], num_syms)
def _chase_from_spectra(self, pay_spectra, header_part, payload_len,
has_crc, cr_val):
"""Run the Chase soft-decision decoder on payload spectra.
``pay_spectra`` are the per-symbol FFT spectra of the payload symbols
only; ``header_part`` (or None) are the already-demodulated header
symbols prepended to each candidate. Returns
``(win_syms, cinfo)`` on success and ``(None, cinfo)`` when no
candidate validated; ``cinfo['stage']`` is ``'none'`` then.
"""
def decode_fn(cand_payload):
full = cand_payload if header_part is None else np.concatenate(
[header_part, cand_payload])
try:
payload, crc_bytes, _info = decode(
full, self.sf, impl_header=self.implicit_header,
forced_payload_len=payload_len, forced_has_crc=has_crc,
forced_cr=cr_val)
except Exception:
return False
if has_crc and len(crc_bytes) == 2:
return bool(np.array_equal(calc_lora_crc16(payload),
crc_bytes))
return True
ok, cand_syms, cinfo = chase_decode(
pay_spectra, decode_fn, sf=self.sf,
**self.settings.chase_kwargs)
if not ok:
return None, cinfo
win = cand_syms if header_part is None else np.concatenate(
[header_part, cand_syms])
return win, cinfo
def _decode_xhonneux(self, iq, resample=True, carrier_offset_hz=0.0):
"""Demodulate and decode a single packet (Xhonneux paper sync).
The buffer may contain a packet anywhere inside it; synchronization
finds the preamble. Sample positions in the returned Packet are
relative to the start of ``iq``.
Parameters
----------
iq : ndarray
Complex baseband at ``fs``.
resample : bool
Resample ``fs -> bw`` before sync (Xhonneux sync runs at Nyquist).
carrier_offset_hz : float
Optional coarse carrier offset applied to ``iq`` before sync.
Returns
-------
Packet
``Packet.ok`` is True when the payload decoded without error
(``crc_valid`` may still be False); on a hard failure ``ok`` is
False and ``error`` holds the reason.
"""
impl_header = self.implicit_header
forced_payload_len = self.payload_len
forced_has_crc = self.crc_enabled
forced_cr = self.code_rate
decode_mode = self.settings.decode_mode
pkt = Packet(sf=self.sf, bw=self.bw, timestamp_sec=time.time())
iq = np.asarray(iq, dtype=np.complex128)
if carrier_offset_hz:
n = np.arange(len(iq))
iq = iq * np.exp(-1j * 2 * np.pi * carrier_offset_hz * n / self.fs)
fs = self.fs
if resample and abs(fs - self.bw) > 1.0:
iq = resample_to_bw(iq, fs, self.bw)
fs = self.bw
synced, payload_start, sync_params = self.sync(iq, resample=False)
synced_full = sync_params.pop('_corrected', None) if sync_params else None
if synced is None:
pkt.error = sync_params.get('error', 'Sync failed')
logger.debug('sync failed: %s', pkt.error)
return pkt
pkt.sync = sync_params
pkt.snr_est = sync_params.get('snr_est')
pkt.freq_offset_hz = sync_params.get('freq_shift_hz')
pkt.sample_start = sync_params.get('preamble_start', 0)
pkt.payload_start = payload_start
if impl_header:
if forced_payload_len is None or forced_payload_len < 1 or forced_payload_len > 255:
pkt.error = f'Invalid payload_len={forced_payload_len}'
return pkt
if forced_cr < 1 or forced_cr > 4:
pkt.error = f'Invalid code_rate={forced_cr}'
return pkt
payload_len = forced_payload_len
has_crc = forced_has_crc
cr_val = forced_cr
pkt.mode = 'implicit'
pkt.header_info = {
'payload_len': payload_len,
'has_crc': has_crc,
'cr': cr_val,
'mode': 'implicit',
}
else:
# The Nyquist-rate sync resolves the integer STO modulo N, so the
# payload can land one symbol early or late depending on the scan
# grid phase. Try the neighbouring symbol positions and keep the
# first whose header checksum validates.
hdr_syms = None
last_err = None
for shift in (0, self.N, -self.N):
base = payload_start + shift
if base < 0 or base >= len(synced_full):
continue
cand = self._demod_at(synced_full, base, 8)
if len(cand) < 8:
continue
try:
payload_len, has_crc, cr_val = \
decode_header(cand, self.sf)[:3]
except Exception as e:
last_err = e
continue
hdr_syms = cand
if shift:
payload_start = base
synced = synced_full[base:]
pkt.payload_start = payload_start
break
if hdr_syms is None:
pkt.error = f'Header decode failed: {last_err}'
logger.warning('header decode failed: %s', last_err)
return pkt
pkt.mode = 'explicit'
pkt.header_info = {
'payload_len': payload_len,
'has_crc': has_crc,
'cr': cr_val,
'mode': 'explicit',
}
total_syms = calc_payload_sym_num(
payload_len, has_crc, self.sf, cr_val, impl_header=impl_header
)
pkt.total_syms_needed = total_syms
pkt.sample_end = pkt.sample_start + int(
(self.preamble_syms + total_syms) * self.N
)
data_syms = self.demodulate(synced, total_syms)
pkt.data_symbols = data_syms
def run_decode(symbols):
"""Decode a candidate symbol vector into ``pkt``.
Returns True when the payload decoded cleanly (CRC-16 valid when
the packet carries one).
"""
pkt.error = None
try:
payload, crc_bytes, _info = decode(
symbols, self.sf,
impl_header=impl_header,
forced_payload_len=payload_len,
forced_has_crc=has_crc,
forced_cr=cr_val,
)
except Exception as e:
pkt.error = str(e)
return False
pkt.payload_bytes = payload
pkt.crc_bytes = crc_bytes
if has_crc and len(crc_bytes) == 2:
expected_crc = calc_lora_crc16(payload)
pkt.crc_valid = bool(np.array_equal(expected_crc, crc_bytes))
else:
pkt.crc_valid = None
text = bytes(payload).decode('utf-8', errors='replace').rstrip('\x00')
pkt.payload_text = text
return bool(pkt.crc_valid) if has_crc else True
if decode_mode == 'chase':
# Chase is the decoder from the start: no separate hard-decision
# pre-decode and no sub-sample timing probe. The payload spectra
# go straight into the soft-decision decoder, whose first
# candidate is the max-likelihood (argmax) symbol vector; single,
# pair and triple flips of the least-reliable positions follow if
# that fails. The header was already decoded above and is
# prepended to every candidate.
spectra = demodulate_spectra_from(
synced, 0, self.downchirp, self.N, total_syms
)
header_part = None if impl_header else hdr_syms
pay_spectra = spectra if impl_header else spectra[8:]
win_syms, cinfo = self._chase_from_spectra(
pay_spectra, header_part,
forced_payload_len, forced_has_crc, forced_cr)
if win_syms is not None:
logger.debug('chase recovered payload (stage=%s)',
cinfo.get('stage') if cinfo else '?')
pkt.data_symbols = win_syms
run_decode(win_syms)
else:
# No Chase candidate validated: report the best-effort decode
# of the argmax symbols so the packet still carries a payload
# and a crc_valid verdict.
logger.debug('chase failed (stage=%s), using hard symbols',
cinfo.get('stage') if cinfo else '?')
run_decode(data_syms)
else:
hard_ok = run_decode(data_syms)
# The sub-sample alignment probe is expensive and cannot succeed
# on a partially buffered packet: while the stream is still
# filling the buffer it would just repeat a failing CRC check on
# every chunk. Only probe once the whole packet fits in ``iq``.
full_buffered = (pkt.sample_end is None
or pkt.sample_end <= len(iq))
if not hard_ok and has_crc and synced_full is not None \
and full_buffered:
# Eq. 20's fractional STO can be off by up to ~0.5 sample and
# payload_start is rounded on top of that; at fs = BW that is
# a per-symbol bin error. Probe sub-sample alignments before
# giving up.
for samp in self.settings.timing_search:
base = payload_start + samp
if base < 0 or base >= len(synced_full):
continue
alt = self._demod_at(synced_full, base, total_syms)
if len(alt) < total_syms:
continue
if run_decode(alt):
logger.debug('timing probe +%.2f resolved payload',
samp)
payload_start = base
pkt.payload_start = payload_start
data_syms = alt
pkt.data_symbols = alt
hard_ok = True
break
if not hard_ok:
logger.warning(
'payload decode failed after full sync and '
'sub-sample timing probe (crc invalid)')
run_decode(data_syms)
pkt.ok = True
return pkt
# =====================================================================
# gr-lora_sdr synchronization (exact port of frame_sync_impl.cc)
# =====================================================================
def _decode_gr(self, iq, carrier_offset_hz=0.0, max_payload_bytes=255):
"""Dispatch wrapper: sweeps ``settings.sfo_ppm`` when it is 'auto'."""
ppm = getattr(self.settings, 'sfo_ppm', None)
if ppm != 'auto':
return self._decode_gr_once(iq, carrier_offset_hz,
max_payload_bytes, ppm)
# gr-lora_sdr infers the SFO from the residual CFO, which is wrong
# whenever the frequency error is not purely a shared-clock effect
# (offset LO, Doppler rate, or a pre-applied coarse correction). The
# required value is per-frame -- measured across three captures from
# one pass it ranges over roughly -30..+15 ppm with no common value --
# so sweep it and accept the first frame whose CRC validates.
best = None
for cand in self.settings.sfo_ppm_search:
pkt = self._decode_gr_once(iq, carrier_offset_hz,
max_payload_bytes, cand)
logger.debug('sfo_ppm sweep: %+d ppm -> ok=%s crc=%s',
cand, pkt.ok, pkt.crc_valid)
if pkt.crc_valid is True:
pkt.sync = dict(pkt.sync or {}, sfo_ppm=cand)
return pkt
if best is None or (pkt.ok and not best.ok):
best = pkt
return best
def _decode_gr_once(self, iq, carrier_offset_hz=0.0,
max_payload_bytes=255, sfo_ppm=None):
"""Decode one packet with gr-lora_sdr's ``frame_sync`` (exact port).
This is the faithful Python port of the EPFL gr-lora_sdr
synchronization block (``lib/frame_sync_impl.cc``): preamble
detection, Bernier fractional-CFO estimation, RCTSL fractional-STO
estimation, net-id validation, ``floor(down_val/2)`` integer-CFO
extraction and SFO/STO correction. The CFO is left in the output
signal and removed at demodulation time by building the reference
upchirp at ``cfo_int`` (the same trick gr-lora_sdr's ``fft_demod``
uses), so the sync works for any downchirp behaviour.
The input must be at the decoder's native sample rate ``fs`` (the
block requires oversampling ``fs/bw`` to select the STO decimation
phase). A coarse ``carrier_offset_hz`` (e.g. the nominal tuning
offset) may be applied first; the block then estimates the residual.
Parameters
----------
iq : ndarray
Complex baseband at ``fs`` (the whole recording or a window).
carrier_offset_hz : float
Optional coarse carrier offset applied to ``iq`` before sync.
max_payload_bytes : int
Symbols output when the header feedback has not been decoded yet.
Returns
-------
Packet
``Packet.ok`` is True when the payload decoded without error.
"""
from softlora.gr_frame_sync import GrFrameSync, build_upchirp
iq = np.asarray(iq, dtype=np.complex128)
if carrier_offset_hz:
n = np.arange(len(iq))
iq = iq * np.exp(-1j * 2 * np.pi * carrier_offset_hz * n / self.fs)
os_factor = int(round(self.fs / self.bw))
# GrFrameSync decimates by picking every os_factor-th sample with no
# anti-alias filter (as gr-lora_sdr does, which expects os<=8). At
# high oversampling that folds os_factor noise bands into the LoRa
# band; band-limit and decimate down to max_os first.
max_os = getattr(self.settings, 'max_os_factor', 4)
if max_os and os_factor > max_os and os_factor % max_os == 0:
from scipy.signal import resample_poly
iq = resample_poly(iq, 1, os_factor // max_os)
os_factor = max_os
fsync = GrFrameSync(sf=self.sf, bw=self.bw, center_freq=self.fc,
sync_word=self.sync_word,
preamble_len=self.preamble_len,
os_factor=os_factor, impl_head=self.implicit_header,
sfo_ppm=sfo_ppm)
synced, info = fsync.run(iq, max_payload_bytes=max_payload_bytes)
logger.debug('gr frame sync: synced=%d samples cfo_int=%s '
'cfo_frac=%.3f snr=%s',
len(synced), info.get('cfo_int'), info.get('cfo_frac', 0.0),
info.get('snr'))
pkt = Packet(sf=self.sf, bw=self.bw, timestamp_sec=time.time())
pkt.sync = info
pkt.snr_est = info.get('snr')
pkt.freq_offset_hz = (info.get('cfo_int', 0) + info.get('cfo_frac', 0.0)) \
* self.bw / self.N
if len(synced) < self.N:
pkt.error = 'No frame synchronized'
return pkt
# gr-lora_sdr fft_demod builds its reference at cfo_int and adjusts it
# by cfo_frac, so the residual CFO is removed at demodulation time.
N = self.N
up_ref = build_upchirp(info['cfo_int'] % N, self.sf)
down_ref = (np.conj(up_ref)
* np.exp(-1j * 2 * np.pi * info['cfo_frac']
/ N * np.arange(N)))
nsyms = len(synced) // N
syms = np.array([
int(np.argmax(np.abs(np.fft.fft(synced[k * N:(k + 1) * N]
* down_ref))))
for k in range(nsyms)
])
if self.implicit_header:
payload_len = self.payload_len
has_crc = self.crc_enabled
cr_val = self.code_rate
pkt.mode = 'implicit'
pkt.header_info = {
'payload_len': payload_len, 'has_crc': has_crc,
'cr': cr_val, 'mode': 'implicit',
}
else:
try:
payload_len, has_crc, cr_val = decode_header(syms[:8], self.sf)[:3]
except Exception as e:
pkt.error = f'Header decode failed: {e}'
return pkt
pkt.mode = 'explicit'
pkt.header_info = {
'payload_len': payload_len, 'has_crc': has_crc,
'cr': cr_val, 'mode': 'explicit',
}
total_syms = calc_payload_sym_num(
payload_len, has_crc, self.sf, cr_val,
impl_header=self.implicit_header)
pkt.total_syms_needed = total_syms
# Implicit-header frames carry no header symbols in the sync output.
pkt.data_symbols = (syms[:total_syms] if self.implicit_header
else syms[:8 + total_syms])
def apply_decode(symbols):
try:
payload, crc_bytes, _info = decode(
symbols, self.sf, impl_header=self.implicit_header,
forced_payload_len=payload_len,
forced_has_crc=has_crc, forced_cr=cr_val)
except Exception as e:
pkt.error = str(e)
return False
pkt.payload_bytes = payload
pkt.crc_bytes = crc_bytes
if has_crc and len(crc_bytes) == 2:
pkt.crc_valid = bool(np.array_equal(
calc_lora_crc16(payload), crc_bytes))
else:
pkt.crc_valid = None
pkt.payload_text = bytes(payload).decode(
'utf-8', errors='replace').rstrip('\x00')
return True
if self.settings.decode_mode == 'chase':
# Chase is the decoder from the start: no separate hard-decision
# payload decode. The explicit header (if any) was already
# decoded from the hard symbols above; the payload spectra go
# straight into the soft-decision decoder.
n_syms = len(syms)
full_spectra = demodulate_spectra_from(
synced, 0, down_ref, self.N, n_syms)
header_part = None if self.implicit_header else syms[:8]
pay_spectra = (full_spectra if self.implicit_header
else full_spectra[8:])
win_syms, _ = self._chase_from_spectra(
pay_spectra, header_part, payload_len, has_crc, cr_val)
if win_syms is not None:
pkt.data_symbols = win_syms
apply_decode(win_syms)
else:
# No Chase candidate validated: report the best-effort decode
# of the argmax symbols so the packet still carries a payload
# and a crc_valid verdict.
apply_decode(syms)
else:
if not apply_decode(syms):
return pkt
pkt.ok = True
return pkt
# =====================================================================
# Streaming decoding
# =====================================================================
def decode_stream(self, chunk):
"""Feed a chunk of IQ samples; returns packets found in this chunk.
Chunks may be any length and packets may straddle chunk boundaries:
the decoder buffers internally, detects preambles as they arrive, and
only decodes a packet once all of its symbols are buffered. Call
:meth:`flush` when the transmission ends to decode the remaining tail.
Both sync algorithms stream: ``xhonneux`` resamples ``fs -> bw`` and
runs the streaming scan; ``gr-lora-sdr`` runs the native-rate
run-detection scanner (no resampling).
Parameters
----------
chunk : array_like
Complex baseband samples at the decoder's ``fs``.
Returns
-------
list of Packet
Packets whose decode completed within this chunk.
"""
if self.settings.sync_algorithm == 'gr-lora-sdr':
if not self._stream_active:
self._stream_init()
return self._gr_stream.process(chunk)
if not self._stream_active:
self._stream_init()
chunk = np.asarray(chunk, dtype=np.complex64)
rs = self._resampler.process(chunk)
if len(rs):
self._buf = np.concatenate([self._buf, rs])
return self._scan(flush=False)
def flush(self):
"""End of stream: decode whatever remains buffered and reset state.
Returns
-------
list of Packet
"""
if not self._stream_active:
return []
logger.debug('flush: decoding remaining buffered data')
if self.settings.sync_algorithm == 'gr-lora-sdr':
found = self._gr_stream.flush()
else:
found = self._scan(flush=True)
self.reset()
return found
def reset(self):
"""Clear any streaming state (buffers, resampler, packet counter)."""
self._stream_active = False
self._buf = None
self._scan_pos = 0
self._pending = None
self._base = 0
self._pkt_count = 0
resampler = getattr(self, '_resampler', None)
if resampler is not None:
resampler.reset()
gr_stream = getattr(self, '_gr_stream', None)
if gr_stream is not None:
gr_stream.reset()
# ------------------------------------------------------------------
# Streaming internals
# ------------------------------------------------------------------
def _stream_init(self):
self._stream_active = True
logger.debug('streaming started (sync=%s)',
self.settings.sync_algorithm)
if self.settings.sync_algorithm == 'gr-lora-sdr':
self._gr_stream = _GrFrameStream(self)
return
self._resampler = _StatefulDecimator(self.fs, self.bw)
self._buf = np.zeros(0, dtype=np.complex64)
self._scan_pos = 0
self._pending = None
self._base = 0
self._pkt_count = 0
self._min_signal = int((self.preamble_syms + 1) * self.N)
self._max_packet_samples = int(self.settings.max_packet_syms * self.N)
def _scan(self, flush=False):
found = []
while True:
if self._pending is not None:
off = self._pending
p = self._decode_at(off)
if p.sync is None:
# Not actually a preamble after all.
logger.debug('pending candidate @%d is not a preamble', off)
self._pending = None
self._scan_pos = off + self.N
continue
if self._truncated(p, off, flush):
if flush:
self._pending = None
self._scan_pos = off + self.N
continue
if self._give_up(off):
logger.warning(
'giving up on candidate @%d '
'(buffered > %d packet symbols)',
off, self.settings.max_packet_syms)
self._pending = None
self._scan_pos = off + self.N
continue
logger.debug('packet @%d incomplete, waiting for more data',
off)
return found # wait for more data
if p.ok:
self._pending = None
self._emit(p, off, found)
continue
# Complete preamble but the payload still failed to decode.
logger.debug('preamble @%d decoded but payload failed', off)
self._pending = None
self._scan_pos = off + self.N
continue
limit = len(self._buf) if flush else len(self._buf) - self._min_signal
if self._scan_pos >= limit:
if flush and self._scan_pos < len(self._buf):
# Last, possibly short, candidate window.
p = self._decode_at(self._scan_pos)
if (p.sync is not None and not self._truncated(p, self._scan_pos, True)
and p.ok):
self._emit(p, self._scan_pos, found)
self._scan_pos = len(self._buf)
continue
break
# Drop the scanned prefix so the buffer stays bounded even when
# long stretches contain no packets (safe: nothing before the
# scan frontier is needed once there is no pending packet).
self._maybe_compact()
off = self._scan_pos
if not self._gate(off):
self._scan_pos = off + self.N
continue
p = self._decode_at(off)
if p.sync is None:
logger.debug('gate passed @%d but no preamble', off)
self._scan_pos = off + self.N
continue
# A truncated packet must never be emitted: its declared end can
# overrun the buffer and corrupt the scan bookkeeping. Wait for
# the rest of the packet to arrive (or give up in flush mode).
if self._truncated(p, off, flush):
if flush:
self._scan_pos = off + self.N
continue
logger.debug('packet @%d truncated, buffering until complete',
off)
self._pending = off
return found
if p.ok:
self._emit(p, off, found)
continue
# Complete preamble but the payload still failed to decode:
# spurious detection, keep scanning.
logger.debug('preamble @%d decoded but payload failed', off)
self._scan_pos = off + self.N
return found
def _truncated(self, p, off, flush):
"""True when the decoded packet's symbols don't all fit in the buffer.
``p.sample_end`` (preamble start + preamble symbols + payload symbols)
is the true packet span; a truncated packet must never be emitted
because its declared end can overrun the buffer and corrupt the scan
bookkeeping. During live streaming a one-symbol margin defers
emission until more data is clearly on the way; at flush the buffer is
final, so a packet that exactly reaches the end of the stream is fine.
"""
avail = len(self._buf) - off
margin = 0 if flush else self.N
if p.sample_end is not None:
return avail < p.sample_end + margin
return avail < p.sample_start + (self.preamble_syms + 8) * self.N + margin
def _decode_at(self, off):
p = self._decode_xhonneux(self._buf[off:], resample=False)
if p.sync is None:
return p
start = p.sample_start
if start <= 0:
return p
# The sync is sensitive to the slice start position: decoding from a
# few symbols before the preamble can land on a bad alignment. Re-run
# from the detected preamble boundary and prefer that (cleaner) result.
q = self._decode_xhonneux(self._buf[off + start:], resample=False)
if q.sync is None:
return p
if q.ok and (not p.ok or q.crc_valid is True and p.crc_valid is not True):
q.sample_start += start
q.payload_start += start
q.sample_end += start
return q
return p
def _gate(self, off):
"""Cheap preamble pre-filter before running the full 3-stage sync.
Looks for three consecutive dechirped windows peaking in nearly the
same FFT bin with a peak-to-median ratio above ``gate_ratio``. A
single window far above its median (``strong_ratio``) also passes,
which keeps the gate from missing preambles at very low SNR.
"""
nwin = 3
if off + nwin * self.N > len(self._buf):
return True # not enough data to judge; let decode decide
bins = []
ratios = []
for k in range(nwin):
y = np.abs(np.fft.fft(
self._buf[off + k * self.N:off + (k + 1) * self.N] * self.downchirp
))
b = int(np.argmax(y))
bins.append(b)
ratios.append(y[b] / (np.median(y) + 1e-12))
if any(r >= self.settings.strong_ratio for r in ratios):
return True
for b1, b2 in zip(bins, bins[1:]):
if abs(b1 - b2) > 2:
return False
return min(ratios) >= self.settings.gate_ratio
def _give_up(self, off):
return len(self._buf) - off >= self._max_packet_samples
def _emit(self, p, off, found):
# Convert buffer-relative positions to absolute stream positions.
abs_off = self._base + off
rel_start = p.sample_start
rel_end = p.sample_end
p.sample_start = abs_off + rel_start
p.payload_start = abs_off + p.payload_start
p.sample_end = abs_off + rel_end
p.time_start_sec = p.sample_start / self.bw
p.packet_index = self._pkt_count
self._pkt_count += 1
found.append(p)
self._scan_pos = off + rel_end
self._compact()
logger.info(
'emitted packet %d: ok=%s crc=%s snr=%s%s',
p.packet_index, p.ok, p.crc_valid,
f'{p.snr_est:.1f} dB' if p.snr_est is not None else 'n/a',
f' payload={p.payload_text!r}' if p.ok else '',
)
def _compact(self):
if self._scan_pos > 0:
# Never drop more than the buffer actually holds (a defensive
# clamp; _truncated normally prevents this).
drop = min(self._scan_pos, len(self._buf))
self._base += drop
self._buf = self._buf[drop:]
self._scan_pos -= drop
def _maybe_compact(self):
"""Periodically drop the already-scanned buffer prefix.
Without this the buffer would grow with every scanned sample between
packets, so long idle stretches (or a multi-GB file) would accumulate
the whole stream in memory. Only called when there is no pending
packet, so the pending offset can never be invalidated.
"""
if self._pending is None and self._scan_pos >= self._min_signal:
self._compact()
# =====================================================================
# Synchronization / demodulation primitives
# =====================================================================
def sync(self, iq, resample=True):
"""Run the 3-stage synchronization pipeline.
Parameters
----------
iq : ndarray
Input complex baseband signal.
resample : bool
Whether to resample to BW first.
Returns
-------
synced : ndarray or None
Frequency-corrected signal sliced from data start.
payload_start : int or None
Sample index in original signal of data start.
params : dict or None
Dict of all estimated sync parameters (incl. ``snr_est``).
"""
if resample and abs(self.fs - self.bw) > 1.0:
iq = resample_to_bw(iq, self.fs, self.bw)
downchirp = self.downchirp
N = self.N
N_detect = self.settings.N_detect
N_preamble_up = self.preamble_len
N_netid = self.settings.N_netid
N_sfd_down = self.settings.N_sfd_down
l, symbols, fft_results, lambda_cfo = stage1_detect_and_cfo(
iq, downchirp, N, N_detect
)
if l is None:
return None, None, {'error': 'No preamble detected'}
s_tilde_up, lambda_sto_prelim, Y_avg = stage2_preliminary_sto(
iq, downchirp, N, l, lambda_cfo, N_detect
)
L_CFO, L_STO, lambda_sto, s_hat_up, s_hat_down, M_hat = \
stage3_final_sync(
iq, downchirp, N, l, lambda_cfo,
Y_avg, N_preamble_up, N_netid, N_detect
)
# An integer STO of exactly N/2 (the wrap boundary of Eq. 18) is
# ambiguous: it means the single D1 window demodulated at the upchirp
# peak, which happens when the receiver grid is misaligned with the
# frame and the D1 window straddles the net-id/SFD boundary. Re-estimate
# from the second SFD downchirp (D2) to disambiguate.
if L_STO == N // 2:
L_CFO, L_STO, lambda_sto, s_hat_up, s_hat_down, M_hat = \
stage3_final_sync(
iq, downchirp, N, l, lambda_cfo,
Y_avg, N_preamble_up, N_netid, N_detect, down_shift=1
)
# Eq. 18 leaves L_STO in [0, N): the paper's model assumes an STO
# advance tau = (L_STO + lambda_STO)/B with 0 <= L_STO < N, so the
# correction is applied with the unsigned value.
# Eq. 18 returns L_STO in [0, N). The paper assumes the receiver
# window always starts before the frame, so tau in [0, Ts). A
# free-running scan has an arbitrary grid phase, and a slightly
# *negative* STO comes back as ~N, which would push payload_start a
# full symbol early. Unwrap into [-N/2, N/2).
total_sto = Gamma_N(L_STO, N) + lambda_sto
preamble_start = (l - (N_detect - 1)) * N
total_preamble_syms = N_preamble_up + N_netid + N_sfd_down
# L_STO is the STO advance in [0, N); ``preamble_start`` is the start
# of the first detected preamble window, and the payload begins exactly
# ``preamble_syms`` symbols later, minus the STO.
payload_start_real = preamble_start + total_preamble_syms * N - total_sto
payload_start = int(round(payload_start_real))
iq_corrected = apply_freq_correction(iq, L_CFO, lambda_cfo, N)
synced = iq_corrected[payload_start:]
snr_est = estimate_snr(
iq_corrected, downchirp, N,
start_sample=preamble_start,
num_syms=N_preamble_up,
)
params = {
'preamble_last_idx': l,
'lambda_cfo': lambda_cfo,
's_tilde_up': s_tilde_up,
'lambda_sto_prelim': lambda_sto_prelim,
'L_CFO': L_CFO,
'L_STO': L_STO,
'lambda_sto': lambda_sto,
'M_hat': M_hat,
's_hat_up': s_hat_up,
's_hat_down': s_hat_down,
'total_sto_samples': total_sto,
'preamble_start': preamble_start,
'payload_start': payload_start,
'freq_shift_hz': (self.bw / N) * (L_CFO + lambda_cfo),
'l_value': l,
'snr_est': snr_est,
'_corrected': iq_corrected,
}
logger.debug(
'sync: preamble_start=%d cfo=%.1f Hz (L_CFO=%d lam=%.2f) '
'sto=%.1f snr=%.1f dB',
preamble_start, (self.bw / N) * (L_CFO + lambda_cfo),
L_CFO, lambda_cfo, total_sto, snr_est)
return synced, payload_start, params
def demodulate(self, signal, num_symbols=None):
"""Extract symbol bin indices from a synced signal.
Parameters
----------
signal : ndarray
Synced complex baseband signal (1-D), assumed to be at
Nyquist rate (fs = BW) so that each symbol is N samples.
num_symbols : int, optional
Number of symbols to demodulate. If None, demodulates
as many full symbols as fit in the signal.
Returns
-------
ndarray
Array of symbol bin indices (0 to 2**sf - 1).
"""
if num_symbols is None:
num_symbols = len(signal) // self.N
return demodulate_symbols_from(
signal, 0, self.downchirp, self.N, num_symbols
)