orion-sdr 0.0.27

DSP/SDR block library targeting HF-to-UHF, satellites, and Python bindings. Roadmap inside.
Documentation
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
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
187
188
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
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
"""
Type stubs for the orion-sdr native extension module.

All classes live in the flat ``orion_sdr`` namespace.  IQ arrays use
``numpy.complex64``; audio arrays use ``numpy.float32``; bit arrays use
``numpy.uint8`` (one bit per byte, value 0 or 1).  Every ``process()``
call returns a new 1-D array; arrays must be 1-D and C-contiguous, or a
``ValueError`` is raised.
"""

import numpy as np
from numpy.typing import NDArray

# ---------------------------------------------------------------------------
# Demodulators  (IQ complex64 → audio float32)
# ---------------------------------------------------------------------------

class CwEnvelopeDemod:
    """One-pole envelope detector for CW signals.

    Tracks the instantaneous magnitude of the IQ input with a low-pass
    time constant derived from *env_bw_hz*.  *tone_hz* is accepted for
    API symmetry but is not used internally (pre-tune the signal before
    passing it in).
    """

    def __init__(self, sample_rate: float, tone_hz: float, env_bw_hz: float) -> None: ...
    def set_gain(self, g: float) -> None: ...
    def process(self, iq: NDArray[np.complex64]) -> NDArray[np.float32]: ...

class AmEnvelopeDemod:
    """AM envelope demodulator with 4th-order IIR low-pass and DC blocker.

    Two envelope methods are available:

    * ``abs_approx=False`` (default) — ``sqrt(I² + Q²)`` after the LP filter
      (*PowerSqrt*); highest fidelity.
    * ``abs_approx=True`` — ``k1·|I| + k2·|Q|`` approximation (*AbsApprox*,
      k1=0.9482, k2=0.3920); slightly faster with a small amplitude error.
    """

    def __init__(
        self,
        fs: float,
        audio_bw_hz: float,
        abs_approx: bool = False,
    ) -> None: ...
    def process(self, iq: NDArray[np.complex64]) -> NDArray[np.float32]: ...

class SsbProductDemod:
    """SSB product detector with BFO rotator and 4th-order IIR audio filter.

    Set *bfo_hz* to 0 for a signal already tuned to baseband, or to a small
    offset to shift the recovered audio pitch.
    """

    def __init__(self, fs: float, bfo_hz: float, audio_bw_hz: float) -> None: ...
    def process(self, iq: NDArray[np.complex64]) -> NDArray[np.float32]: ...

class FmQuadratureDemod:
    """FM quadrature (phase-difference) discriminator.

    Output is scaled so that ±*dev_hz* of instantaneous frequency deviation
    maps to roughly ±1.0.  A 4th-order IIR low-pass at *audio_bw_hz* follows
    the discriminator.
    """

    def __init__(self, fs: float, dev_hz: float, audio_bw_hz: float) -> None: ...
    def process(self, iq: NDArray[np.complex64]) -> NDArray[np.float32]: ...

class PmQuadratureDemod:
    """PM quadrature demodulator (instantaneous phase difference).

    *k* scales the recovered phase difference to the output audio level.
    A 4th-order IIR low-pass at *audio_bw_hz* follows the discriminator.
    """

    def __init__(self, fs: float, k: float, audio_bw_hz: float) -> None: ...
    def process(self, iq: NDArray[np.complex64]) -> NDArray[np.float32]: ...

# ---------------------------------------------------------------------------
# Digital demodulators  (IQ complex64 → bits uint8)
# ---------------------------------------------------------------------------

class BpskDemod:
    """BPSK demodulator: hard-decision slicer.

    Input: complex64 IQ array, carrier-removed baseband, 1 sample per symbol.
    Output: uint8 bit array — one bit (0 or 1) per input symbol.
    Decision rule: Re(z) ≥ 0 → 0, Re(z) < 0 → 1.

    *gain* scales the soft metric before slicing (use 1.0 for normalized input).
    """

    def __init__(self, gain: float = 1.0) -> None: ...
    def set_gain(self, g: float) -> None: ...
    def process(self, iq: NDArray[np.complex64]) -> NDArray[np.uint8]: ...

class QpskDemod:
    """QPSK demodulator: hard-decision slicer.

    Input: complex64 IQ array, carrier-removed baseband, 1 sample per symbol.
    Output: uint8 bit array — two bits per input symbol, interleaved as
    ``[b0_I, b0_Q, b1_I, b1_Q, …]``.  Matches the Gray coding of ``QpskMod``.

    *gain* scales the soft metric before slicing (use 1.0 for normalized input).
    """

    def __init__(self, gain: float = 1.0) -> None: ...
    def set_gain(self, g: float) -> None: ...
    def process(self, iq: NDArray[np.complex64]) -> NDArray[np.uint8]: ...

class QamDemod:
    """QAM demodulator: hard-decision slicer for square QAM constellations.

    *order* must be 16, 64, or 256 (raises ``ValueError`` otherwise).
    Input: complex64 IQ array, carrier-removed baseband, 1 sample per symbol.
    Output: uint8 bit array — ``log2(order)`` bits per input symbol, laid out
    as ``log2(order)/2`` I-axis bits then ``log2(order)/2`` Q-axis bits
    (MSB-first within each axis Gray index).  Matches ``QamMod`` bit layout.

    *gain* scales the soft metric before slicing (use 1.0 for normalized input).
    """

    def __init__(self, order: int, gain: float = 1.0) -> None: ...
    def set_gain(self, g: float) -> None: ...
    def process(self, iq: NDArray[np.complex64]) -> NDArray[np.uint8]: ...

# ---------------------------------------------------------------------------
# Modulators  (audio float32 → IQ complex64)
# ---------------------------------------------------------------------------

class AmDsbMod:
    """AM double-sideband modulator with optional carrier and RF upconversion.

    * *carrier_level* — 1.0 produces full carrier (A3E); 0.0 gives DSB-SC.
    * *modulation_index* — values ≤ 1.0 are recommended to avoid
      over-modulation.
    * *rf_hz* — set to 0.0 for baseband IQ output.
    """

    def __init__(
        self,
        fs: float,
        rf_hz: float,
        carrier_level: float,
        modulation_index: float,
    ) -> None: ...
    def set_gain(self, g: float) -> None: ...
    def set_clamp(self, on: bool) -> None:
        """Clamp the modulated envelope to ±1 to prevent over-modulation."""
        ...
    def process(self, audio: NDArray[np.float32]) -> NDArray[np.complex64]: ...

class CwKeyedMod:
    """CW keyed carrier modulator with shaped rise/fall envelope.

    The input array is a **keying envelope** in the range 0..1 (not raw
    audio): 1.0 = key down, 0.0 = key up.  Rise and fall times are
    smoothed by one-pole filters with time constants *rise_ms* / *fall_ms*
    to suppress key clicks.
    """

    def __init__(
        self,
        sample_rate: float,
        tone_hz: float,
        rise_ms: float,
        fall_ms: float,
    ) -> None: ...
    def set_gain(self, g: float) -> None: ...
    def process(self, audio: NDArray[np.float32]) -> NDArray[np.complex64]: ...

class FmPhaseAccumMod:
    """FM modulator using a phasor-recurrence phase accumulator.

    Each sample multiplies a running phasor by ``exp(j·2π·kf·x/fs)``
    where ``kf = deviation_hz``.  The phasor is renormalized every 1024
    samples to prevent amplitude drift.  Set *rf_hz* to 0.0 for baseband
    output.
    """

    def __init__(
        self,
        sample_rate: float,
        deviation_hz: float,
        rf_hz: float,
    ) -> None: ...
    def set_deviation(self, hz: float) -> None: ...
    def set_gain(self, g: float) -> None: ...
    def process(self, audio: NDArray[np.float32]) -> NDArray[np.complex64]: ...

class PmDirectPhaseMod:
    """PM modulator: instantaneous phase φ = kp · x[n].

    *kp_rad_per_unit* maps ±1.0 audio to ±kp radians of carrier phase.
    Set *rf_hz* to 0.0 for baseband output.
    """

    def __init__(
        self,
        sample_rate: float,
        kp_rad_per_unit: float,
        rf_hz: float,
    ) -> None: ...
    def set_gain(self, g: float) -> None: ...
    def set_sensitivity(self, kp: float) -> None:
        """Update the phase sensitivity (rad per unit input amplitude)."""
        ...
    def process(self, audio: NDArray[np.float32]) -> NDArray[np.complex64]: ...

class SsbPhasingMod:
    """SSB phasing modulator (Weaver-style IIR variant).

    Audio is up-converted to *audio_if_hz* via a complex rotator, split
    into I and Q paths through matched 4th-order IIR low-pass filters,
    then combined to select the desired sideband.  Set *usb=True* for
    upper sideband, *usb=False* for lower sideband.  Set *rf_hz* to 0.0
    for baseband IQ output.
    """

    def __init__(
        self,
        fs: float,
        audio_bw_hz: float,
        audio_if_hz: float,
        rf_hz: float,
        usb: bool,
    ) -> None: ...
    def process(self, audio: NDArray[np.float32]) -> NDArray[np.complex64]: ...

# ---------------------------------------------------------------------------
# Digital modulators  (bits uint8 → IQ complex64)
# ---------------------------------------------------------------------------

class BpskMod:
    """BPSK modulator: Gray-coded constellation mapper + waveform stage.

    Input: uint8 bit array (LSB of each byte used), one bit per symbol.
    Output: complex64 IQ array of the same length.
    Constellation: bit 0 → (+1, 0), bit 1 → (−1, 0).

    Set *rf_hz* to 0.0 for baseband output; non-zero upconverts via an
    internal ``Rotator`` (phasor recurrence, no per-sample trig).
    """

    def __init__(self, fs: float, rf_hz: float, gain: float = 1.0) -> None: ...
    def set_gain(self, g: float) -> None: ...
    def process(self, bits: NDArray[np.uint8]) -> NDArray[np.complex64]: ...

class QpskMod:
    """QPSK modulator: Gray-coded constellation mapper + waveform stage.

    Input: uint8 bit array (LSB of each byte); consumed in pairs ``[b0, b1]``.
    Output: complex64 IQ array of length ``len(bits) // 2``.
    Constellation is normalized to unit energy (each axis ±1/√2).

    Set *rf_hz* to 0.0 for baseband output.
    """

    def __init__(self, fs: float, rf_hz: float, gain: float = 1.0) -> None: ...
    def set_gain(self, g: float) -> None: ...
    def process(self, bits: NDArray[np.uint8]) -> NDArray[np.complex64]: ...

class QamMod:
    """Square QAM modulator: Gray-coded constellation mapper + waveform stage.

    *order* must be 16, 64, or 256 (raises ``ValueError`` otherwise).
    Input: uint8 bit array (LSB of each byte); consumed ``log2(order)`` bytes
    per symbol.  Output: complex64 IQ array of length
    ``len(bits) // log2(order)``.

    Constellation is Gray-coded on each axis independently and normalized to
    unit average symbol energy.  Set *rf_hz* to 0.0 for baseband output.
    """

    def __init__(self, order: int, fs: float, rf_hz: float, gain: float = 1.0) -> None: ...
    def set_gain(self, g: float) -> None: ...
    def process(self, bits: NDArray[np.uint8]) -> NDArray[np.complex64]: ...

# ---------------------------------------------------------------------------
# FT8/FT4 waveform classes
# ---------------------------------------------------------------------------

class Ft8Mod:
    """FT8 frame modulator: 8-FSK CPFSK, 79 symbols, 151 680 samples at 12 kHz.

    Input: uint8 array of 58 tone indices (0–7).
    Output: complex64 IQ array of shape (151680,).
    """

    def __init__(self, fs: float, base_hz: float, rf_hz: float, gain: float = 1.0) -> None: ...
    def modulate(self, tones: NDArray[np.uint8]) -> NDArray[np.complex64]: ...

class Ft8Demod:
    """FT8 frame demodulator: Goertzel/dot-product tone detector.

    Input: complex64 IQ array of at least 151 680 samples.
    Output: uint8 array of 58 tone indices.
    Raises ``ValueError`` if input is too short or demodulation fails.
    """

    def __init__(self, fs: float, base_hz: float) -> None: ...
    def demodulate(self, iq: NDArray[np.complex64]) -> NDArray[np.uint8]: ...

class Ft8Codec:
    """FT8 channel codec: CRC-14 + LDPC(174,91) + Gray code.

    All methods are static; no per-instance state.
    """

    def __init__(self) -> None: ...
    @staticmethod
    def encode(payload: bytes) -> NDArray[np.uint8]:
        """Encode a 10-byte payload → uint8[58] Gray-coded tone indices."""
        ...
    @staticmethod
    def decode_hard(tones: NDArray[np.uint8]) -> bytes | None:
        """Hard-decision decode 58 tone indices → bytes[10], or None on failure."""
        ...
    @staticmethod
    def decode_soft(llr: NDArray[np.float32]) -> bytes | None:
        """Soft-decision decode float32[174] LLRs → bytes[10], or None on failure."""
        ...

class Ft4Mod:
    """FT4 frame modulator: 4-FSK CPFSK, 105 symbols, 60 480 samples at 12 kHz.

    Input: uint8 array of 87 tone indices (0–3).
    Output: complex64 IQ array of shape (60480,).
    """

    def __init__(self, fs: float, base_hz: float, rf_hz: float, gain: float = 1.0) -> None: ...
    def modulate(self, tones: NDArray[np.uint8]) -> NDArray[np.complex64]: ...

class Ft4Demod:
    """FT4 frame demodulator: Goertzel/dot-product tone detector.

    Input: complex64 IQ array of at least 60 480 samples.
    Output: uint8 array of 87 tone indices.
    Raises ``ValueError`` if input is too short or demodulation fails.
    """

    def __init__(self, fs: float, base_hz: float) -> None: ...
    def demodulate(self, iq: NDArray[np.complex64]) -> NDArray[np.uint8]: ...

class Ft4Codec:
    """FT4 channel codec: XOR scramble + CRC-14 + LDPC(174,91) + Gray code.

    All methods are static; no per-instance state.
    """

    def __init__(self) -> None: ...
    @staticmethod
    def encode(payload: bytes) -> NDArray[np.uint8]:
        """Encode a 10-byte payload → uint8[87] Gray-coded tone indices."""
        ...
    @staticmethod
    def decode_hard(tones: NDArray[np.uint8]) -> bytes | None:
        """Hard-decision decode 87 tone indices → bytes[10], or None on failure."""
        ...
    @staticmethod
    def decode_soft(llr: NDArray[np.float32]) -> bytes | None:
        """Soft-decision decode float32[174] LLRs → bytes[10], or None on failure."""
        ...

# ---------------------------------------------------------------------------
# FT8/FT4 sync functions
# ---------------------------------------------------------------------------

def ft8_sync(
    iq: NDArray[np.complex64],
    fs: float,
    base_hz: float,
    max_hz: float,
    t_min: int,
    t_max: int,
    max_cand: int,
) -> list[dict]:
    """Synchronise an FT8 IQ buffer and return up to *max_cand* frame candidates.

    Each candidate is a dict::

        {
            "time_sym": int,           # symbol offset of frame start
            "freq_bin": int,           # frequency bin of tone-0
            "score":    float,         # Costas match score
            "llr":      float32[174],  # soft LLRs for Ft8Codec.decode_soft
        }

    Pass each result's ``"llr"`` to ``Ft8Codec.decode_soft`` to recover the
    77-bit payload.
    """
    ...

def ft4_sync(
    iq: NDArray[np.complex64],
    fs: float,
    base_hz: float,
    max_hz: float,
    t_min: int,
    t_max: int,
    max_cand: int,
) -> list[dict]:
    """Synchronise an FT4 IQ buffer and return up to *max_cand* frame candidates.

    Same return shape as ``ft8_sync``.
    """
    ...

# ---------------------------------------------------------------------------
# FT8/FT4 message packing functions
# ---------------------------------------------------------------------------

def ft8_pack_standard(call_to: str, call_de: str, extra: str) -> bytes:
    """Pack a standard FT8/FT4 message → bytes[10].

    *extra* may be a Maidenhead grid (``"FN31"``), signal report (``"+07"``,
    ``"-12"``), R-prefixed report (``"R+05"``), or token
    (``"RRR"``, ``"RR73"``, ``"73"``).  Pass ``""`` for no extra field.

    Raises ``ValueError`` if the callsigns cannot be encoded.
    """
    ...

def ft8_pack_free_text(text: str) -> bytes:
    """Pack a free-text FT8/FT4 message (up to 13 chars) → bytes[10].

    Raises ``ValueError`` if the text is too long or contains invalid characters.
    """
    ...

def ft8_pack_telemetry(data: bytes) -> bytes:
    """Pack a telemetry FT8/FT4 message (exactly 9 bytes) → bytes[10].

    Raises ``ValueError`` if *data* is not exactly 9 bytes.
    """
    ...

def ft8_unpack(payload: bytes) -> dict:
    """Unpack a 10-byte FT8/FT4 payload → dict.

    The ``"type"`` key indicates the message type:

    * ``"standard"``  — ``{"type", "call_to", "call_de", "extra"}``
    * ``"free_text"`` — ``{"type", "text"}``
    * ``"telemetry"`` — ``{"type", "data"}`` (bytes[9])
    * ``"nonstd"``    — ``{"type", "call_to", "call_de", "extra"}``
    * ``"unknown"``   — ``{"type", "payload"}`` (bytes[10])

    Raises ``ValueError`` if *payload* is not exactly 10 bytes.
    """
    ...