Skip to content

Packet

softlora.packet

Packet result type and SNR estimation for the LoRa receiver.

Packet dataclass

A single decoded LoRa packet.

ok is True whenever the payload was demodulated and decoded without raising (crc_valid may still be False for a corrupted payload). On a hard failure ok is False and error holds the reason.

Sample positions (sample_start, payload_start, sample_end) are counts at the decoder bandwidth rate (bw samples/second). They are absolute within the recording for :meth:LoRaDecoder.decode_iq, :meth:LoRaDecoder.decode_file and :meth:LoRaDecoder.decode_stream, and relative to the input buffer for a direct :meth:LoRaDecoder.decode call.

Source code in softlora/packet.py 
 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
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
@dataclass
class Packet:
    """A single decoded LoRa packet.

    ``ok`` is True whenever the payload was demodulated and decoded without
    raising (``crc_valid`` may still be False for a corrupted payload).  On a
    hard failure ``ok`` is False and ``error`` holds the reason.

    Sample positions (``sample_start``, ``payload_start``, ``sample_end``) are
    counts at the decoder bandwidth rate (``bw`` samples/second).  They are
    absolute within the recording for :meth:`LoRaDecoder.decode_iq`,
    :meth:`LoRaDecoder.decode_file` and :meth:`LoRaDecoder.decode_stream`, and
    relative to the input buffer for a direct :meth:`LoRaDecoder.decode`
    call.
    """

    ok: bool = False
    error: str = None

    sf: int = None
    bw: float = None
    mode: str = None  # 'explicit' | 'implicit'

    payload_bytes: np.ndarray = field(default_factory=lambda: np.zeros(0, dtype=np.uint8))
    payload_text: str = ''
    crc_valid: bool = None
    crc_bytes: np.ndarray = field(default_factory=lambda: np.zeros(0, dtype=np.uint8))

    snr_est: float = None
    freq_offset_hz: float = None

    sample_start: int = None
    payload_start: int = None
    sample_end: int = None
    time_start_sec: float = None
    packet_index: int = None

    timestamp_sec: float = None

    # Raw diagnostics (power-user access to the sync/header internals).
    sync: dict = None
    params: dict = None
    header_info: dict = None
    data_symbols: np.ndarray = None
    total_syms_needed: int = None

    def __str__(self):
        """Human-readable detail table for a decoded packet.

        ``print(packet)`` renders the packet as an aligned ``key: value``
        table with the full payload hex (wrapped at 32 bytes per line) and
        the full payload text.  A failed decode prints a one-line error
        instead.  For a compact one-line form see :meth:`__repr__`.
        """
        if not self.ok:
            return f'Packet(error={self.error!r})'

        hdr = self.header_info or {}
        crc_on = hdr.get('has_crc')
        if crc_on is None:
            crc_on = self.crc_valid is not None
        if crc_on:
            valid = {True: 'valid', False: 'INVALID', None: 'not-checked'}[
                self.crc_valid]
            status = f'OK · crc {valid}'
        else:
            status = 'OK · crc off'

        idx = self.packet_index if self.packet_index is not None else '?'
        lines = [f'Packet {idx} — {status}']

        W = 14  # key column width
        pad = ' ' * W

        def wrap(s, width):
            return [s[i:i + width] for i in range(0, len(s), width)] or ['']

        def wrap_repr(s, width):
            # Tokenize \xNN / \0NN escapes so wrapping never splits them.
            toks = re.findall(r'\\x[0-9a-fA-F]{2}|\\0[0-7]{2}|.', s)
            lines, cur = [], ''
            for t in toks:
                if cur and len(cur) + len(t) > width:
                    lines.append(cur)
                    cur = t
                else:
                    cur += t
            if cur:
                lines.append(cur)
            return lines or ['']

        payload = np.asarray(self.payload_bytes, dtype=np.uint8)
        hex_lines = wrap(bytes(payload).hex(), 64)      # 64 hex = 32 bytes
        text_lines = wrap_repr(repr(self.payload_text or ''), 64)

        rows = []
        rows.append(('sf', str(self.sf) if self.sf is not None else '?'))
        rows.append(('bw (Hz)', f'{self.bw:g}' if self.bw is not None else '?'))
        rows.append(('mode', self.mode or '?'))
        rows.append(('payload (B)', str(len(payload))))
        rows.append(('hex', hex_lines))
        rows.append(('text', text_lines))
        if self.sample_start is not None:
            t = (f' (t={self.time_start_sec:.2f}s)'
                 if self.time_start_sec is not None else '')
            rows.append(('start', f'{self.sample_start}{t}'))
        if self.payload_start is not None:
            rows.append(('payload at', str(self.payload_start)))
        if self.sample_end is not None:
            rows.append(('end', str(self.sample_end)))
        if self.snr_est is not None:
            rows.append(('snr (dB)', f'{self.snr_est:.1f}'))
        if self.freq_offset_hz is not None:
            rows.append(('freq offset (Hz)', f'{self.freq_offset_hz:+.0f}'))

        for key, val in rows:
            if isinstance(val, list):
                for i, part in enumerate(val):
                    lines.append(f'{key:<{W}}{part}' if i == 0 else pad + part)
            else:
                lines.append(f'{key:<{W}}{val}')
        return '\n'.join(lines)

    def __repr__(self):
        """Unambiguous one-line summary, used when packets appear in a
        list/log; :meth:`__str__` provides the full detail table."""
        return self._summary()

    def _summary(self):
        if not self.ok:
            return f'Packet(error={self.error!r})'

        idx = self.packet_index if self.packet_index is not None else '?'
        sf = self.sf if self.sf is not None else '?'
        bw = f'{self.bw:g}' if self.bw is not None else '?'
        mode = self.mode if self.mode is not None else '?'

        n = int(len(self.payload_bytes)) if self.payload_bytes is not None else 0
        hex8 = bytes(np.asarray(self.payload_bytes, dtype=np.uint8)[:8]).hex() \
            if n else ''
        text = (self.payload_text or '')
        if len(text) > 40:
            text = text[:40] + '...'

        hdr = self.header_info or {}
        crc_on = hdr.get('has_crc')
        if crc_on is None:
            crc_on = self.crc_valid is not None
        if crc_on:
            valid = {True: 'valid', False: 'INVALID', None: 'not-checked'}[
                self.crc_valid]
            crc = f'on ({valid})'
        else:
            crc = 'off'

        snr = f'{self.snr_est:.1f}dB' if self.snr_est is not None else '?'
        freq = (f'{self.freq_offset_hz:+.0f}Hz'
                if self.freq_offset_hz is not None else '?')
        start = self.sample_start if self.sample_start is not None else '?'

        return (f'Packet[{idx}] sf={sf} bw={bw} mode={mode} len={n}B '
                f'crc={crc} snr={snr} start={start} freq={freq} '
                f'text={text!r} hex={hex8}...')

    def to_dict(self):
        """Plain JSON-serializable dict (numpy arrays converted to lists)."""
        out = {}
        for name, value in self.__dict__.items():
            if isinstance(value, np.ndarray):
                out[name] = value.tolist()
            elif isinstance(value, (np.integer,)):
                out[name] = int(value)
            elif isinstance(value, (np.floating,)):
                out[name] = float(value)
            else:
                out[name] = value
        return out
__repr__()

Unambiguous one-line summary, used when packets appear in a list/log; :meth:__str__ provides the full detail table.

Source code in softlora/packet.py 
133
134
135
136
def __repr__(self):
    """Unambiguous one-line summary, used when packets appear in a
    list/log; :meth:`__str__` provides the full detail table."""
    return self._summary()
__str__()

Human-readable detail table for a decoded packet.

print(packet) renders the packet as an aligned key: value table with the full payload hex (wrapped at 32 bytes per line) and the full payload text. A failed decode prints a one-line error instead. For a compact one-line form see :meth:__repr__.

Source code in softlora/packet.py 
 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
124
125
126
127
128
129
130
131
def __str__(self):
    """Human-readable detail table for a decoded packet.

    ``print(packet)`` renders the packet as an aligned ``key: value``
    table with the full payload hex (wrapped at 32 bytes per line) and
    the full payload text.  A failed decode prints a one-line error
    instead.  For a compact one-line form see :meth:`__repr__`.
    """
    if not self.ok:
        return f'Packet(error={self.error!r})'

    hdr = self.header_info or {}
    crc_on = hdr.get('has_crc')
    if crc_on is None:
        crc_on = self.crc_valid is not None
    if crc_on:
        valid = {True: 'valid', False: 'INVALID', None: 'not-checked'}[
            self.crc_valid]
        status = f'OK · crc {valid}'
    else:
        status = 'OK · crc off'

    idx = self.packet_index if self.packet_index is not None else '?'
    lines = [f'Packet {idx} — {status}']

    W = 14  # key column width
    pad = ' ' * W

    def wrap(s, width):
        return [s[i:i + width] for i in range(0, len(s), width)] or ['']

    def wrap_repr(s, width):
        # Tokenize \xNN / \0NN escapes so wrapping never splits them.
        toks = re.findall(r'\\x[0-9a-fA-F]{2}|\\0[0-7]{2}|.', s)
        lines, cur = [], ''
        for t in toks:
            if cur and len(cur) + len(t) > width:
                lines.append(cur)
                cur = t
            else:
                cur += t
        if cur:
            lines.append(cur)
        return lines or ['']

    payload = np.asarray(self.payload_bytes, dtype=np.uint8)
    hex_lines = wrap(bytes(payload).hex(), 64)      # 64 hex = 32 bytes
    text_lines = wrap_repr(repr(self.payload_text or ''), 64)

    rows = []
    rows.append(('sf', str(self.sf) if self.sf is not None else '?'))
    rows.append(('bw (Hz)', f'{self.bw:g}' if self.bw is not None else '?'))
    rows.append(('mode', self.mode or '?'))
    rows.append(('payload (B)', str(len(payload))))
    rows.append(('hex', hex_lines))
    rows.append(('text', text_lines))
    if self.sample_start is not None:
        t = (f' (t={self.time_start_sec:.2f}s)'
             if self.time_start_sec is not None else '')
        rows.append(('start', f'{self.sample_start}{t}'))
    if self.payload_start is not None:
        rows.append(('payload at', str(self.payload_start)))
    if self.sample_end is not None:
        rows.append(('end', str(self.sample_end)))
    if self.snr_est is not None:
        rows.append(('snr (dB)', f'{self.snr_est:.1f}'))
    if self.freq_offset_hz is not None:
        rows.append(('freq offset (Hz)', f'{self.freq_offset_hz:+.0f}'))

    for key, val in rows:
        if isinstance(val, list):
            for i, part in enumerate(val):
                lines.append(f'{key:<{W}}{part}' if i == 0 else pad + part)
        else:
            lines.append(f'{key:<{W}}{val}')
    return '\n'.join(lines)
to_dict()

Plain JSON-serializable dict (numpy arrays converted to lists).

Source code in softlora/packet.py 
174
175
176
177
178
179
180
181
182
183
184
185
186
def to_dict(self):
    """Plain JSON-serializable dict (numpy arrays converted to lists)."""
    out = {}
    for name, value in self.__dict__.items():
        if isinstance(value, np.ndarray):
            out[name] = value.tolist()
        elif isinstance(value, (np.integer,)):
            out[name] = int(value)
        elif isinstance(value, (np.floating,)):
            out[name] = float(value)
        else:
            out[name] = value
    return out

estimate_snr(signal, downchirp, N, start_sample=0, num_syms=8, guard=3)

Estimate per-sample SNR (dB) from demodulated preamble upchirps.

The signal is assumed frequency-corrected so that every preamble upchirp dechirps to the same FFT bin. With per-sample signal power Ps and noise power Pn, the dechirped FFT concentrates the signal into one bin of power Ps * N**2 while the noise spreads N * Pn per bin, so Ps / Pn = peak_power / (N * median_noise_power).

Parameters:

Name Type Description Default
signal ndarray

Frequency-corrected complex baseband signal at fs = BW.

 required
downchirp ndarray

N-sample reference downchirp.

 required
N int

Samples per symbol (2**sf).

 required
start_sample int

Sample index of the first preamble symbol.

 0
num_syms int

Number of preamble symbols to average (capped at 8).

 8
guard int

Bins around the peak excluded from the noise estimate.

 3

Returns:

Type Description
float or None

SNR estimate in dB, or None if no valid window was available.
Source code in softlora/packet.py 
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
def estimate_snr(signal, downchirp, N, start_sample=0, num_syms=8,
                 guard=3):
    """Estimate per-sample SNR (dB) from demodulated preamble upchirps.

    The signal is assumed frequency-corrected so that every preamble upchirp
    dechirps to the same FFT bin.  With per-sample signal power ``Ps`` and
    noise power ``Pn``, the dechirped FFT concentrates the signal into one bin
    of power ``Ps * N**2`` while the noise spreads ``N * Pn`` per bin, so
    ``Ps / Pn = peak_power / (N * median_noise_power)``.

    Parameters
    ----------
    signal : ndarray
        Frequency-corrected complex baseband signal at fs = BW.
    downchirp : ndarray
        N-sample reference downchirp.
    N : int
        Samples per symbol (2**sf).
    start_sample : int
        Sample index of the first preamble symbol.
    num_syms : int
        Number of preamble symbols to average (capped at 8).
    guard : int
        Bins around the peak excluded from the noise estimate.

    Returns
    -------
    float or None
        SNR estimate in dB, or None if no valid window was available.
    """
    num_syms = int(max(1, min(num_syms, 8)))
    avg = np.zeros(N, dtype=float)
    used = 0
    for i in range(num_syms):
        off = start_sample + i * N
        if off >= len(signal):
            break
        window = np.zeros(N, dtype=complex)
        avail = min(len(signal) - off, N)
        window[:avail] = signal[off:off + avail]
        Y = np.fft.fft(window * downchirp)
        avg += np.abs(Y) ** 2
        used += 1
    if used == 0:
        return None

    avg /= used
    peak = int(np.argmax(avg))
    mask = np.ones(N, dtype=bool)
    for d in range(-guard, guard + 1):
        mask[(peak + d) % N] = False
    noise = np.median(avg[mask])
    if noise <= 0:
        return None
    snr_lin = max(avg[peak] / (N * noise), 1e-12)
    return float(10.0 * np.log10(snr_lin))