Python Waveform Generator API — Synth / PN¶

Everything in the doppler.wfm package imports from one place — from doppler.wfm import …. The two low-level generators are:

Class	Output	Use when
`Synth`	CF32 — the eight-type waveform engine	Generate tone / noise / PN / BPSK / QPSK / chirp / bits / symbols, with optional LO offset and AWGN
`PN`	uint8 — raw LFSR chips (0/1)	Spreading / ranging codes, scrambling, test vectors

Synth is also the unit of composition — pass synths into Segment.sum to mix them (see compose below).

Source: src/doppler/wfm/__init__.py

These same C cores back the one command-line tool, wfmgen — see the Waveform Generator guide.

Reading or writing a capture file is on its own page — Python: Capture I/O covers Reader, Writer and write_blue_header, with the Capture I/O guide as the narrative version.

`Synth` — the eight-type waveform engine¶

One declarative engine produces every waveform type, selected by the string type (tone, noise, pn, bpsk, qpsk, chirp, bits, symbols). Construction takes keyword arguments mirroring the generator flags; sensible defaults mean a bare Synth() is a clean, unit-power baseband tone.

from doppler.wfm import Synth
import numpy as np

# Bare construct → clean baseband tone, unit power, no noise
x = Synth().steps(4096)            # complex64

# Tone at Fs/10 with 20 dB SNR
tone = Synth(type="tone", fs=1e6, freq=100_000, snr=20).steps(4096)

# Complex AWGN (unit power)
noise = Synth(type="noise", seed=7).steps(8192)

# PN / BPSK / QPSK — sps samples per chip/symbol
pn   = Synth(type="pn",   pn_length=7, sps=1).steps(127)
bpsk = Synth(type="bpsk", sps=8, snr=10).steps(8192)
qpsk = Synth(type="qpsk", sps=8, snr=10).steps(8192)

# Scalar (one sample at a time)
s = Synth(type="tone", freq=1000, fs=1e6).step()

Bits (user-defined pattern)¶

A bits waveform plays back a specific bit sequence — preambles, sync words, test vectors, exact packet structures. The pattern is a binary string ("10110101"), a hex string ("0xAA55", MSB first), or any array-like of 0/1; modulation maps the bits to symbols ("none" → 0/1 amplitude, "bpsk" → ±1, "qpsk" → two bits per symbol, Gray-coded). Each bit is held sps samples and the pattern cycles to fill the requested length, so one pass is Synth.n_samples.

from doppler.wfm import Synth, bits

# 8-bit preamble, BPSK, 4 samples/bit → 32 samples for one pass
s = bits(pattern="10110101", sps=4, modulation="bpsk")
preamble = s.steps(32)  # 8 bits * 4 sps

# Hex sync word, unmodulated 0/1; direct construction is equivalent
sync = Synth(type="bits", pattern="0xAA55", modulation="none", sps=8)

# From a numpy array
import numpy as np
payload = bits(
    pattern=np.array([1, 0, 1, 1, 0, 1, 0, 1], np.uint8), modulation="qpsk"
)

Symbols (arbitrary constellation)¶

Where bits maps data through a fixed modulation, type="symbols" skips the map entirely: you supply a complex64 constellation stream and each element is an output point directly, oversampled by sps, cycled to fill the request, and RRC-shaped through the same matched-FIR path (pulse="rrc"). That expresses any modulation an enum doesn't — pi/4-QPSK, QAM, APSK — by computing the points yourself and passing them. On the composer Synth the stream is the symbols= keyword; on the low-level _SynthEngine it is attached with set_symbols() after construction. A complex128 array is accepted (force-cast to complex64).

import numpy as np
from doppler.wfm import Synth

# pi/4-QPSK: rotate every other QPSK symbol by pi/4, then pass the stream
qpsk = np.array([1 + 1j, -1 + 1j, -1 - 1j, 1 - 1j], np.complex64) / np.sqrt(2)
stream = np.array(
    [qpsk[i % 4] * (np.exp(1j * np.pi / 4) if i % 2 else 1) for i in range(64)],
    np.complex64,
)
iq = Synth(type="symbols", symbols=stream, sps=8, pulse="rrc").steps(64 * 8)

See the Symbols gallery walkthrough for worked pi/4-QPSK and 16-QAM constellations, rect vs RRC pulses, and the envelope floor behind pi/4-QPSK's lower PAPR.

Chirp (LFM sweep)¶

A chirp is a linear-FM sweep: its instantaneous frequency ramps from freq (the start, also spellable f_start=) to f_end over the generated length, then holds at f_end. The phase is continuous, so multi-segment chirps join seamlessly — pulse-compression, SAR, sonar, and frequency-response test signals all fall out of this one type. f_end < freq is a down-chirp; snr adds AWGN exactly as for a tone.

from doppler.wfm import Synth, chirp

# Up-chirp 100 kHz → 300 kHz over 10000 samples at 1 MS/s
up = chirp(f_start=100e3, f_end=300e3, fs=1e6).steps(10000)

# Down-chirp (equivalent direct construction; freq IS the start frequency)
down = Synth(type="chirp", freq=1e6, f_end=500e3, fs=2e6).steps(50000)

The sweep span is the length you ask for: steps(N) sweeps over exactly N samples standalone, and in a Segment the sweep fills the segment's num_samples — so f_end is reached at the last sample either way.

Clean vs noisy, baseband vs offset¶

snr is in dB. snr >= 100 (the default) is clean — no AWGN is generated at all, so a clean waveform pays no noise cost. Lower it to add noise. freq = 0 (the default) is baseband — the LO is skipped entirely.

clean   = Synth(type="qpsk", sps=8, snr=100).steps(8192)   # no AWGN
noisy   = Synth(type="qpsk", sps=8, snr=12).steps(8192)     # Es/No 12 dB
offset  = Synth(type="pn", pn_length=9, sps=1, freq=2.5e5, fs=1e6).steps(511)

snr_mode ("auto", "fs", "ebno", "esno") sets how snr is interpreted; "auto" uses over-fs for tone/noise/PN and Es/No for BPSK/QPSK.

RRC pulse shaping (band-limited carriers)¶

By default the modulated types (pn / bpsk / qpsk) emit rectangular sample-and-hold chips — a wide sinc² spectrum. Set pulse="rrc" for root-raised-cosine pulse shaping: the symbol stream is filtered to a band-limited channel, so a realistic carrier (e.g. WCDMA QPSK, RRC roll-off 0.22) comes straight from the generator. rrc_beta is the roll-off and rrc_span the filter support in symbols. The taps are unit-transmit-power scaled, so the output stays at unit average power.

from doppler.wfm import qpsk

shaped = qpsk(sps=8, pulse="rrc", rrc_beta=0.22, rrc_span=8).steps(1 << 16)
# band-limited: its occupied bandwidth is ~(1+beta)/sps, far below the rect sinc²

PN modulation: length, polynomial, realization¶

# Auto-pick the maximum-length polynomial for the register length (2..64)
Synth(type="pn", pn_length=23, sps=1).steps(8192)

# Explicit 64-bit polynomial
Synth(type="pn", pn_length=40, pn_poly=0x800000001C, sps=1).steps(8192)

# Fibonacci realization (same polynomial/period, different chip ordering)
Synth(type="pn", pn_length=9, sps=1, lfsr="fibonacci").steps(511)

Determinism¶

s = Synth(type="qpsk", sps=4, seed=11)
a = s.steps(512)
s.reset()
assert np.array_equal(a, s.steps(512))   # same seed → identical stream

Synth ¶

Synth.

Parameters:

Name	Type	Description	Default
`type`	`str`	Waveform type. One of `"tone"`, `"noise"`, `"pn"`, `"bpsk"`, `"qpsk"`, `"chirp"`, `"bits"`, `"symbols"`, `"dsss"`.	``"tone"``
`freq`	`float \| tuple[float, float]`	Carrier/offset frequency in Hz (normalised cycles/sample when fs=1); for chirp it is the start frequency.	`0.0`
`snr`	`float \| tuple[float, float]`	Signal-to-noise ratio in dB, interpreted per snr_mode; >=100 is treated as clean (no AWGN).	`100.0`
`snr_mode`	`str`	How snr is interpreted: auto picks fs for tone/pn/chirp/bits and Es/No for bpsk/qpsk. One of `"auto"`, `"fs"`, `"ebno"`, `"esno"`.	``"auto"``
`seed`	`int`	PRNG/LFSR seed for the noise and PN streams.	`0`
`sps`	`int`	Samples per symbol (PSK) or per chip (PN); the oversampling factor.	`1`
`pn_length`	`int`	PN LFSR register length; the sequence period is 2^pn_length - 1.	`15`
`pn_poly`	`int`	PN generator polynomial; 0 auto-selects a maximal-length (MLS) polynomial for pn_length.	`0`
`lfsr`	`str`	PN LFSR realization (galois or fibonacci); same period, different chip order. One of `"galois"`, `"fibonacci"`.	``"galois"``
`level`	`float \| tuple[float, float]`	Source power in dBFS (<=0; 0 = unit power). Applies only when summed in a Segment/Composer (gain 10^(level/20)); ignored by standalone Synth.steps().	`0.0`
`background`	`int`	Mark this source as part of the static background field (0/1). Plan.prepare() folds a contiguous leading run of background sources into ONE pre-summed cache entry instead of caching each separately, so a scene of many fixed emitters costs one buffer rather than hundreds. The composite is overridable as a unit: it takes a single slot in gains/phases/enable and counts as one in n_sources(), so scaling it trims the whole field while its members keep their relative levels. Background sources must come first in the segment (a non-prefix ordering is rejected by prepare, since the fold would no longer reproduce compose bit-for-bit). Ignored by compose() and by standalone Synth.steps().	`0`
`f_end`	`float \| tuple[float, float]`	Chirp end frequency in Hz; ignored by non-chirp types.	`0.0`
`bits`	`bytes \| None`	For type=bits: the 0/1 pattern, oversampled by sps and cycled to fill the request. For type=dsss (as `payload`): the payload bits of the burst frame.	`None`
`modulation`	`str`	For type=bits: symbol mapping of the pattern (none=0/1 amplitude, bpsk, qpsk). One of `"none"`, `"bpsk"`, `"qpsk"`.	``"bpsk"``
`pulse`	`str`	Pulse shape for the symbol stream (pn/bpsk/qpsk/bits): rect sample-and-hold or rrc matched filter. One of `"rect"`, `"rrc"`.	``"rect"``
`rrc_beta`	`float`	RRC roll-off factor in (0, 1] when pulse=rrc.	`0.35`
`rrc_span`	`int`	RRC filter span in symbols when pulse=rrc (taps = 2spansps + 1).	`8`
`symbols`	`NDArray[complex64] \| None`	For type=symbols: a complex64 constellation stream — each element is the output point itself, oversampled by sps, cycled, and RRC-shaped with pulse=rrc. Generalises any modulation (pi/4-QPSK, QAM, ...).	`None`
`acq_code`	`bytes \| None`	For type=dsss: the acquisition/preamble code (0/1 chips), repeated acq_reps times unmodulated at the head of the burst — the coherent pull-in target BurstDespreader.set_acq/BurstDemod.set_preamble lock to.	`None`
`acq_reps`	`int`	For type=dsss: preamble repetitions (periods of acq_code before the frame).	`1`
`data_code`	`bytes \| None`	For type=dsss: the payload spreading code (0/1 chips) — a second code, distinct from acq_code; every frame bit (sync \| payload \| crc) is XOR-spread across its full length, so len(data_code) is the spreading factor.	`None`
`sync`	`bytes \| None`	For type=dsss: the frame-sync word bits (e.g. Barker-13) between the preamble and the payload — what BurstDemod.set_sync correlates to resolve frame position and BPSK polarity. Optional.	`None`
`crc`	`str`	For type=dsss: the frame trailer — crc16 appends a CRC-16-CCITT over the payload bits (what BurstDemod validates as frame_valid); none omits it. One of `"none"`, `"crc16"`.	``"crc16"``
`symbol_rate`	`float`	For type=dsss: > 0 selects CONTINUOUS asynchronous mode — the spreading code repeats endlessly and data rides on it at this symbol rate (Hz), independent of the code-epoch rate (chips/symbol = fs/sps/symbol_rate, non-integer). No preamble/sync/CRC frame; data comes from the payload when supplied, else a seeded PN a receiver regenerates. Absent/0 = burst.	`0.0`
`dsss_code_only`	`int`	Continuous dsss data source: 1 = code-only (the pure spreading code, no data modulation); 0 = data-modulated (the payload when supplied, else the seeded PN). Ignored for burst dsss and non-dsss types.	`0`
`fs`	`float`	Sample rate in Hz — one per segment (all sources share it).	`1.0`

steps ¶

steps(n: int) -> NDArray[np.complex64]

Generate n complex samples.

step ¶

step() -> complex

Generate one complex sample.

reset ¶

reset() -> None

Reset to initial state.

`PN` — raw LFSR m-sequence¶

A right-shift LFSR producing one bit (0/1) per call. With a primitive polynomial it is a maximum-length sequence: period 2**n - 1 with 2**(n-1) ones per period. Registers up to 64 bits are supported, in either the Galois (internal-XOR, default) or Fibonacci (external-XOR) realization — both realize the same polynomial and period.

from doppler.wfm import PN
import numpy as np

# Length-7 MLS (primitive polynomial 0x41), one full period
chips = np.asarray(PN(0x41, 1, 7).generate(127))   # uint8, 64 ones / 63 zeros

# Fibonacci realization of the same polynomial
fib = np.asarray(PN(0x41, 1, 7, lfsr="fibonacci").generate(127))

# 64-bit register
big = np.asarray(PN(0x800000001C, 1, 40).generate(50_000))

# Deterministic replay
p = PN(0x41, 1, 7)
a = np.asarray(p.generate(127)).copy()
p.reset()
assert np.array_equal(a, np.asarray(p.generate(127)))

The constructor is PN(poly, seed, length, lfsr="galois"). seed must be non-zero (the all-zero register is a fixed point). To map chips to ±1 BPSK symbols, use Synth(type="pn", ...) instead, which also handles oversampling, the LO, and AWGN.

PN ¶

Allocate and initialise a maximal-length-sequence LFSR. The register is seeded from seed and will produce a pseudo-random binary sequence with period 2^length - 1 for any primitive poly. Both Galois and Fibonacci realizations share the same primitive polynomial and therefore the same period; they differ only in chip ordering/phase.

Parameters:

Name	Type	Description	Default
`poly`	`int`	Galois feedback tap polynomial (right-shift convention). The LSB is the tap at position 0 (always 1 for a primitive poly); bit k=1 means tap at position k. Default 96 (0x60) is primitive for length=7, giving period 127. The Fibonacci taps are derived automatically so you only supply one value.	`0`
`seed`	`int`	Initial LFSR register state; must be non-zero (the all-zero state is a fixed point). Default 1.	`0`
`length`	`int`	Register width in bits, 1..64. The sequence period is 2^length - 1 for a primitive polynomial. Default 7.	`0`
`lfsr`	`Literal['galois', 'fibonacci']`	Realization: PN_GALOIS (0, default) or PN_FIBONACCI (1).	`"galois"`

Examples:

>>> from doppler.wfm import PN
>>> import numpy as np
>>> p = PN(poly=96, seed=1, length=7)
>>> chips = p.generate(127)
>>> chips.dtype
dtype('uint8')
>>> int(chips.sum())   # 64 ones per MLS period (2^(n-1))
64

reset ¶

reset() -> None

Reset PN to its post-create state. Reloads the LFSR register from the original seed so the sequence restarts from chip 0. Useful for reproducible captures without re-allocating.

Examples:

>>> from doppler.wfm import PN
>>> import numpy as np
>>> p = PN(poly=96, seed=1, length=7)
>>> a = p.generate(8).copy()
>>> p.reset()
>>> np.array_equal(a, p.generate(8))
True

generate ¶

generate(
    count: int = 1, out: NDArray[uint8] | None = None
) -> NDArray[np.uint8]

Generate n chips into out and advance the LFSR by n positions. Each element of out is 0 or 1. Requesting more than one MLS period is valid — the sequence simply wraps around. The Python binding returns a zero-copy NumPy uint8 view over a pre-allocated buffer; copy the result before calling generate again if you need a snapshot.

Returns:

Type	Description
`NDArray[uint8]`	min(n, max_out) chips.

Examples:

>>> from doppler.wfm import PN
>>> import numpy as np
>>> p = PN(poly=96, seed=1, length=7)
>>> chips = p.generate(127)
>>> chips[:8].tolist()
[1, 0, 0, 0, 0, 0, 1, 1]
>>> int(chips.sum())   # 64 ones per MLS period
64

generate_max_out ¶

generate_max_out() -> int

Largest number of samples generate() can return in the current state.

Size an out= buffer with this before calling generate(), or use it to allocate one up front. The bound is this object's own: what it depends on is a property of the algorithm, so a header block on generate_max_out() replaces this text.

Returns:

Type	Description
`int`	Upper bound on the output length; the actual call may return fewer.

state_bytes ¶

state_bytes() -> int

Size in bytes of this object's serialized state.

The exact length get_state returns and set_state requires. It depends on how the object was constructed (state arrays are sized at construction), so read it from the instance rather than assuming a constant.

Raises RuntimeError if the PN has already been destroyed.

Returns:

Type	Description
`int`	Byte length of one serialized state blob.

get_state ¶

get_state() -> bytes

Serialize this object's mutable state to bytes.

Captures exactly the state that evolves as the object runs, so a blob taken now and restored later resumes from this point. Construction parameters are not included: restore into an object built the same way.

The blob is opaque and always state_bytes() long. Its layout is an implementation detail of the C core and is not a stable format across builds.

Raises RuntimeError if the PN has already been destroyed.

Returns:

Type	Description
`bytes`	Opaque snapshot, `state_bytes()` bytes long.

set_state ¶

set_state(blob: bytes) -> None

Restore mutable state from a get_state() blob.

Overwrites the live state in place; the object keeps the parameters it was constructed with. Length is validated against state_bytes() before the blob is handed to the C core, and the core may reject it as well.

Raises TypeError if blob is not bytes, ValueError if its length differs from state_bytes() or the core rejects it, and RuntimeError if the PN has already been destroyed.

Parameters:

Name	Type	Description	Default
`blob`	`bytes`	A `get_state()` blob from this type, exactly `state_bytes()` long.	required

destroy ¶

destroy() -> None

Release the underlying C resources immediately.

Ordinarily unnecessary: the resources are freed when the object is garbage-collected. Call this to release them at a definite point instead, or use the object as a context manager, which calls it on exit.

Idempotent: calling it again on an already-released object does nothing. Every other method raises RuntimeError once it has run.

enter ¶

__enter__() -> PN

Enter a context manager, returning this object.

Lets a PN be used in a with statement so its C resources are released deterministically on exit rather than at collection time.

Returns:

Type	Description
`PN`	This same object, not a copy.

exit ¶

__exit__(
    exc_type: object | None = ...,
    exc: object | None = ...,
    tb: object | None = ...,
) -> None

Exit a context manager, releasing the PN.

Equivalent to calling destroy(). Returns None, so an exception raised inside the with body propagates normally; this never suppresses one.

Parameters:

Name	Type	Description	Default
`exc_type`	`object \| None`	Exception class, or None. Ignored.	`...`
`exc`	`object \| None`	Exception instance, or None. Ignored.	`...`
`tb`	`object \| None`	Traceback object, or None. Ignored.	`...`

`compose` — multi-segment composition, writers, and a NATS sink¶

The composition layer is the Python face of the C wfmgen composer subsystem — the same engine behind the wfmgen CLI, output byte-identical for the same parameters. There are two composition verbs:

Segment.sum(*synths, num_samples=…) mixes synths at the same time over one resolved noise floor (a multi-source scene);
Segment.add(*segments) sequences segments in time (a timeline).

The ladder is Synth → (.sum) → Segment → (.add) → Timeline → Composer → samples: .sum stacks synths in the same time window (one column), .add lays segments out along time (one row).

flowchart LR
    subgraph SEG["Segment — .sum() mixes at the SAME time, one noise floor"]
        direction TB
        y1["Synth qpsk · level −10 dBFS"]
        y2["Synth tone · level −3 dBFS"]
        y3["Synth noise · the floor"]
    end
    subgraph TL["Timeline — .add() sequences in TIME ▶"]
        direction LR
        sA["Segment A"] --> sB["Segment B<br/>(+ trailing gap)"] --> sC["…"]
    end
    SEG -- ".add(B, …)" --> sA
    TL --> COMP["Composer(…).compose()"] --> IQ[("complex64 I/Q")]

    classDef syn fill:#ede7f6,stroke:#5e35b1,color:#000;
    classDef seg fill:#e3f2fd,stroke:#1565c0,color:#000;
    class y1,y2,y3 syn;
    class sA,sB,sC seg;

A Composer turns a Segment / Timeline / segment-list into samples, optionally looping (repeat) or running forever (continuous); Writer serialises to the four file types (raw / CSV / BLUE type-1000 / SigMF), and StreamSink publishes over NATS (requires a nats-server reachable at the endpoint). The resolved spec round-trips through JSON, so a capture is fully reproducible.

import numpy as np
from doppler.wfm import Composer, Reader, Segment, Writer, mls_poly, qpsk, tone

# Mix: a QPSK signal of interest under a CW interferer, one noise floor.
scene = Segment.sum(
    qpsk(snr=15, sps=8, level=-10),       # builders return Synth
    tone(freq=2e5, level=-3),
    num_samples=65536,
)

# Sequence: a PN preamble, then the scene, back-to-back in time.
timeline = Segment("pn", num_samples=127, pn_length=7).add(scene)
iq = Composer(timeline).compose()         # one complex64 array

# Or stream block-by-block (an empty block marks the end):
c = Composer(timeline)
with Writer("frame.cf32", fs=1e6, sample_type="cf32") as w:
    while len(blk := c.execute(4096)):
        w.write(blk)
with Reader("frame.cf32", sample_type="cf32") as r:  # C reader → complex64
    back = r.read(r.num_samples)

# Reproducible: the resolved spec serialises to JSON and back.
j = Composer(timeline).to_json()
assert np.array_equal(Composer.from_json(j).compose(), iq)

# Utilities
mls_poly(7)                               # 0x41 — the length-7 MLS polynomial

The builders tone() / bpsk() / qpsk() / pn() / noise() / chirp(f_start=…, f_end=…) / bits(pattern=…, modulation=…) each return a Synth (a noise(level=…) is a bare AWGN floor at that level in dBFS; a chirp is an LFM sweep; a bits(...) plays a user pattern); or construct Synth(...) directly. In a Segment.sum the per-synth snr resolves into one shared noise floor, and each synth's level (dBFS) sets its share.

Getting samples into and out of a file is its own topic — Writer, its dual Reader, content-based file-type detection, and what metadata each file type survives are all on Python: Capture I/O. For SigMF, pair a Writer(..., file_type="sigmf") data file with Composer(...).to_sigmf(...), which is the piece that knows the scene and can therefore annotate it.

The StreamSink is POSIX-only. DSP helpers rrc_taps(beta, sps, span) and dsss_spread(syms, code, sf) expose the pulse-shaping and spreading primitives.

SampleClock (POSIX) paces and timestamps a stream against an ideal fs-Hz clock — the same C core behind the wfmgen --realtime CLI flag. Use it to throttle a producer to real time and to tag blocks with their ideal timestamp:

from doppler.wfm import Composer, SampleClock, StreamSink

# Stream at the true 1 MS/s instead of as fast as possible. Requires a
# nats-server reachable at the endpoint.
comp = Composer(type="qpsk", sps=8, continuous=True)
clk = SampleClock(fs=1e6)
with StreamSink("nats://127.0.0.1:4222/iq") as sink:
    while True:
        blk = comp.execute(4096)
        ts = clk.stamp()              # ideal ns timestamp of this block
        sink.send(blk, fs=1e6, fc=0.0)
        clk.pace(len(blk))            # sleep to epoch + n/fs (GIL released)

The schedule is drift-free (deadlines come from the cumulative sample count, not summed sleeps); underruns are counted in clk.underruns / clk.max_lateness, and SampleClock(fs, resync=True) re-anchors to "now" on each underrun.

Classes¶

Composer ¶

Composer.

Parameters:

Name	Type	Description	Default
`segments`	`Segment \| Timeline \| list[Segment] \| None`	Initial segment list.	`None`
`repeat`	`bool`	Loop the sequence after the last segment.	`False`
`continuous`	`bool`	Never finish; execute always returns the requested count.	`False`

execute ¶

execute(n: int) -> NDArray[np.complex64]

Execute for n samples.

compose ¶

compose(block: int = ...) -> NDArray[np.complex64]

Compose the full sequence into one array.

stream ¶

stream(
    block: int = ..., realtime: float = ...
) -> Iterator[NDArray[np.complex64]]

Iterate the sequence in blocks.

to_dict ¶

to_dict() -> dict

Serialise the composer state to a dict.

to_sigmf ¶

to_sigmf(
    sample_type: str = ...,
    endian: str = ...,
    fs: float = ...,
    fc: float = ...,
    t0: float = ...,
) -> str

Serialise as to_sigmf.

close ¶

close() -> None

Release native resources.

Segment ¶

Segment.

Parameters:

Name	Type	Description	Default
`type`	`str`	Waveform type. One of `"tone"`, `"noise"`, `"pn"`, `"bpsk"`, `"qpsk"`, `"chirp"`, `"bits"`, `"symbols"`, `"dsss"`.	``"tone"``
`freq`	`float \| tuple[float, float]`	Carrier/offset frequency in Hz (normalised cycles/sample when fs=1); for chirp it is the start frequency.	`0.0`
`snr`	`float \| tuple[float, float]`	Signal-to-noise ratio in dB, interpreted per snr_mode; >=100 is treated as clean (no AWGN).	`100.0`
`snr_mode`	`str`	How snr is interpreted: auto picks fs for tone/pn/chirp/bits and Es/No for bpsk/qpsk. One of `"auto"`, `"fs"`, `"ebno"`, `"esno"`.	``"auto"``
`seed`	`int`	PRNG/LFSR seed for the noise and PN streams.	`0`
`sps`	`int`	Samples per symbol (PSK) or per chip (PN); the oversampling factor.	`1`
`pn_length`	`int`	PN LFSR register length; the sequence period is 2^pn_length - 1.	`15`
`pn_poly`	`int`	PN generator polynomial; 0 auto-selects a maximal-length (MLS) polynomial for pn_length.	`0`
`lfsr`	`str`	PN LFSR realization (galois or fibonacci); same period, different chip order. One of `"galois"`, `"fibonacci"`.	``"galois"``
`level`	`float \| tuple[float, float]`	Source power in dBFS (<=0; 0 = unit power). Applies only when summed in a Segment/Composer (gain 10^(level/20)); ignored by standalone Synth.steps().	`0.0`
`background`	`int`	Mark this source as part of the static background field (0/1). Plan.prepare() folds a contiguous leading run of background sources into ONE pre-summed cache entry instead of caching each separately, so a scene of many fixed emitters costs one buffer rather than hundreds. The composite is overridable as a unit: it takes a single slot in gains/phases/enable and counts as one in n_sources(), so scaling it trims the whole field while its members keep their relative levels. Background sources must come first in the segment (a non-prefix ordering is rejected by prepare, since the fold would no longer reproduce compose bit-for-bit). Ignored by compose() and by standalone Synth.steps().	`0`
`f_end`	`float \| tuple[float, float]`	Chirp end frequency in Hz; ignored by non-chirp types.	`0.0`
`bits`	`bytes \| None`	For type=bits: the 0/1 pattern, oversampled by sps and cycled to fill the request. For type=dsss (as `payload`): the payload bits of the burst frame.	`None`
`modulation`	`str`	For type=bits: symbol mapping of the pattern (none=0/1 amplitude, bpsk, qpsk). One of `"none"`, `"bpsk"`, `"qpsk"`.	``"bpsk"``
`pulse`	`str`	Pulse shape for the symbol stream (pn/bpsk/qpsk/bits): rect sample-and-hold or rrc matched filter. One of `"rect"`, `"rrc"`.	``"rect"``
`rrc_beta`	`float`	RRC roll-off factor in (0, 1] when pulse=rrc.	`0.35`
`rrc_span`	`int`	RRC filter span in symbols when pulse=rrc (taps = 2spansps + 1).	`8`
`symbols`	`NDArray[complex64] \| None`	For type=symbols: a complex64 constellation stream — each element is the output point itself, oversampled by sps, cycled, and RRC-shaped with pulse=rrc. Generalises any modulation (pi/4-QPSK, QAM, ...).	`None`
`acq_code`	`bytes \| None`	For type=dsss: the acquisition/preamble code (0/1 chips), repeated acq_reps times unmodulated at the head of the burst — the coherent pull-in target BurstDespreader.set_acq/BurstDemod.set_preamble lock to.	`None`
`acq_reps`	`int`	For type=dsss: preamble repetitions (periods of acq_code before the frame).	`1`
`data_code`	`bytes \| None`	For type=dsss: the payload spreading code (0/1 chips) — a second code, distinct from acq_code; every frame bit (sync \| payload \| crc) is XOR-spread across its full length, so len(data_code) is the spreading factor.	`None`
`sync`	`bytes \| None`	For type=dsss: the frame-sync word bits (e.g. Barker-13) between the preamble and the payload — what BurstDemod.set_sync correlates to resolve frame position and BPSK polarity. Optional.	`None`
`crc`	`str`	For type=dsss: the frame trailer — crc16 appends a CRC-16-CCITT over the payload bits (what BurstDemod validates as frame_valid); none omits it. One of `"none"`, `"crc16"`.	``"crc16"``
`symbol_rate`	`float`	For type=dsss: > 0 selects CONTINUOUS asynchronous mode — the spreading code repeats endlessly and data rides on it at this symbol rate (Hz), independent of the code-epoch rate (chips/symbol = fs/sps/symbol_rate, non-integer). No preamble/sync/CRC frame; data comes from the payload when supplied, else a seeded PN a receiver regenerates. Absent/0 = burst.	`0.0`
`dsss_code_only`	`int`	Continuous dsss data source: 1 = code-only (the pure spreading code, no data modulation); 0 = data-modulated (the payload when supplied, else the seeded PN). Ignored for burst dsss and non-dsss types.	`0`
`fs`	`float`	Sample rate in Hz — one per segment (all sources share it).	`1.0`
`num_samples`	`int \| tuple[int, int]`	Segment on-time in samples (the active span).	`1024`
`off_samples`	`int \| tuple[int, int]`	Trailing off-time gap in samples (zeros) appended after the segment.	`0`
`repeats`	`int`	Play the segment this many times back-to-back (each instance = delay + on-time + trailing gap) before advancing. Ranged fields re-draw and the AWGN is fresh per instance; the signal (codes, payload, PN phase) stays fixed.	`1`
`delay_samples`	`int \| tuple[int, int]`	Leading gap before the on-time (samples) — the burst arrives after this delay. Ranged like off_samples and re-drawn per repeats instance, so a (lo, hi) delay is per-burst arrival jitter. Use off_samples for inter-burst spacing, delay_samples for arrival jitter.	`0`
`gap_noise`	`str`	Gap policy for this segment's delay and trailing gap. auto (default): gaps carry the segment's noise floor — the sources' AWGN keeps running while the signal stops (clean scenes still get exact-zero gaps). off: gaps are hard zeros. One of `"auto"`, `"off"`.	``"auto"``

sum `classmethod` ¶

sum(
    *sources: Synth,
    fs: float = ...,
    num_samples: int | tuple[int, int] = ...,
    off_samples: int | tuple[int, int] = ...,
    repeats: int = ...,
    delay_samples: int | tuple[int, int] = ...,
    gap_noise: str = ...,
) -> Segment

Combine sources into a single Segment.

add ¶

add(*others: Segment) -> Timeline

Append segments; return a Timeline.

Timeline ¶

Timeline.

add ¶

add(*segments: Segment) -> Timeline

Append and return self.

StreamSink ¶

Open a stream sink (PUB) bound to a NATS subject.

Parameters:

Name	Type	Description	Default
`endpoint`	`int`	Endpoint, e.g. "nats://127.0.0.1:4222/iq".	required
`sample_type`	`str`	Wire type (wavegen order): 0 cf32, 1 cf64, 2 ci32, 3 ci16, 4 ci8. Integer types use full-scale ±1.0. One of `"cf32"`, `"cf64"`, `"ci32"`, `"ci16"`, `"ci8"`.	``"cf32"``

clip_fraction `property` ¶

clip_fraction: float

clip_fraction (float).

peak_dbfs `property` ¶

peak_dbfs: float

peak_dbfs (float).

clipped `property` ¶

clipped: bool

clipped (bool).

send ¶

send(x: NDArray[Any], fs: float, fc: float) -> int

Convert a cf32 block to the wire type and publish it.

Parameters:

Name	Type	Description	Default
`iq`	`NDArray[Any]`	Complex-float samples; @param n complex sample count.	required
`fs`	`float`	sample rate (Hz); @param fc center frequency (Hz) — wire header.	required
`fc`	`float`	the sink handle.	required

Returns:

Type	Description
`int`	0 on success, non-zero on a send/allocation error.

track_clipping ¶

track_clipping(on: int = ...) -> None

Enable the per-component clip counter (off by default; peak always on).

Parameters:

Name	Type	Description	Default
`on`	`int`	Input.	`...`

close ¶

close() -> None

Release the handle and free resources.

enter ¶

__enter__() -> StreamSink

Enter context; return self.

exit ¶

__exit__(*exc: Any) -> None

Exit context and close the handle.

SampleClock ¶

SampleClock handle.

Parameters:

Name	Type	Description	Default
`fs`	`float`		required
`resync`	`int`		`0`

samples `property` ¶

samples: int

samples (int).

underruns `property` ¶

underruns: int

underruns (int).

max_lateness `property` ¶

max_lateness: float

max_lateness (float).

pace ¶

pace(count: int) -> float

Advance by count samples and sleep until that block's deadline (epoch + n/fs). Returns the slack in seconds measured before sleeping: >= 0 means early (and it slept that long); < 0 means it arrived late — an underrun, which is counted (and the epoch re-anchored when resync is set), with no sleep.

Parameters:

Name	Type	Description	Default
`count`	`int`	Input.	required

Returns:

Type	Description
`float`	Output.

stamp ¶

stamp() -> int

Ideal wall-clock timestamp (ns since the UNIX epoch) of the next sample to be produced — sample index n. Call it before pace() to tag the block you are about to emit, or after to tag the following block. Equivalent to dp_sample_clock_stamp_at(c, c->n).

Returns:

Type	Description
`int`	Output.

stamp_at ¶

stamp_at(n: int) -> int

Ideal wall-clock timestamp (ns since the UNIX epoch) of an ARBITRARY sample index n — past, present, or future, not just the clock's own live position. The receive-side counterpart of dp_sample_clock_stamp(): a block emitting several per-record outputs from one buffered input (e.g. several detections spanning different epochs from one streamed message) stamps each at its own historical sample offset instead of reusing the whole buffer's single arrival time.

Parameters:

Name	Type	Description	Default
`n`	`int`	Input.	required

Returns:

Type	Description
`int`	Output.

track ¶

track(
    observed_timestamp_ns: int,
    n_at_observation: int,
    tolerance_ns: int,
) -> int

Reconcile c's epoch_real_ns against one OBSERVED (timestamp, sample index) pair read off an incoming stream header — the receive-side dual of pace()'s resync: instead of sleeping toward a deadline, this adopts or corrects the epoch from ground truth the sender already stamped.

The FIRST call always adopts observed_timestamp_ns as the epoch (has_anchor starts false — a fresh clock has no real observation yet, so there is nothing to compare against). Every later call only re-anchors if the discrepancy between the observation and what the clock's current model predicts exceeds tolerance_ns (same step-correction semantics as pace()'s own resync, applied to tracking instead of sleeping) — this corrects accumulated epoch OFFSET only, it does not model sample-rate SKEW, exactly like pace()'s resync.

Rejects (no-op, returns 0) any observation with n_at_observation less than the clock's current n outright: a stale, out-of-order, or redelivered header must never walk the epoch backward. Never treat two reconciled observations as literal replay-safe state — always resync from an ARRIVING message, not a cached one.

Parameters:

Name	Type	Description	Default
`observed_timestamp_ns`	`int`	Input.	required
`n_at_observation`	`int`	Input.	required
`tolerance_ns`	`int`	Input.	required

Returns:

Type	Description
`int`	Nonzero if this call adopted or re-anchored the epoch; 0 if it was accepted as already consistent, or rejected as stale.

reset ¶

reset() -> None

Re-capture both epochs and zero the counters — a fresh clock at n=0.

resync ¶

resync() -> None

Re-anchor the pacing epoch to "now" without clearing n or counters, dropping any accumulated lateness so future blocks pace forward from the present. (pace() does this automatically when resync is set.)

close ¶

close() -> None

Release the handle and free resources.

Module-level helpers¶

The SigMF sidecar is now Composer(...).to_sigmf(...) — see the Composer class above. write_blue_header (detached BLUE headers) is on Python: Capture I/O.

rrc_taps ¶

rrc_taps(
    beta: float, sps: int, span: int
) -> NDArray[np.float32]

Root-raised-cosine pulse-shaping taps (2spansps+1 unit-energy cf32 taps).

dsss_spread ¶

dsss_spread(
    syms: NDArray[complex64], code: NDArray[uint8], sf: int
) -> NDArray[np.complex64]

Direct-sequence spread syms by the ±1 chip code; yields len(syms)*sf chips.

crc16 ¶

crc16(bits: NDArray[uint8]) -> int

CRC-16-CCITT (poly 0x1021, init 0xFFFF) over an unpacked 0/1 bit array, MSB-first — the DSSS burst frame trailer wfmgen appends and BurstDemod validates.

Parameters:

Name	Type	Description	Default
`bits`	`NDArray[uint8]`	Array of 0/1 bit values (one per byte).	required

Returns:

Type	Description
`int`	The 16-bit CRC.

Examples:

>>> import numpy as np
>>> from doppler.wfm import crc16
>>> ascii_bits = np.unpackbits(np.frombuffer(b"123456789", np.uint8))
>>> hex(crc16(ascii_bits))   # the standard CCITT check vector
'0x29b1'

mls_poly ¶

mls_poly(n: int) -> int

Maximal-length-sequence primitive polynomial for an LFSR of length n.

Parameters:

Name	Type	Description	Default
`n`	`int`	LFSR length in stages (2..64).	required

Returns:

Type	Description
`int`	Primitive-polynomial tap mask, or 0 if n is out of range.

Examples:

>>> from doppler.wfm import mls_poly
>>> hex(mls_poly(7))
'0x41'

Modulation & SNR helpers¶

Low-level primitives behind the modulated types and the SNR model, exposed for callers who build their own symbol streams or noise budgets: bpsk_map / qpsk_map map bits (and Gray-coded symbol indices) to unit-energy constellation points, wfm_ebno_to_snr_db converts an Eb/No target to the over-fs SNR the generator actually places, and wfm_awgn_amplitude returns the noise amplitude for a given SNR.

bpsk_map ¶

bpsk_map(bits: NDArray[uint8]) -> NDArray[np.complex64]

Map bits {0,1} to BPSK symbols {+1,-1} (cf32).

Parameters:

Name	Type	Description	Default
`bits`	`NDArray[uint8]`	Array of uint8 values; only the LSB of each byte is used.	required

Returns:

Type	Description
`NDArray[complex64]`	Output.

Examples:

>>> from doppler.wfm import bpsk_map
>>> import numpy as np
>>> bits = np.array([0, 1, 0, 1], dtype=np.uint8)
>>> bpsk_map(bits).tolist()
[(1+0j), (-1+0j), (1+0j), (-1+0j)]

qpsk_map ¶

qpsk_map(syms: NDArray[uint8]) -> NDArray[np.complex64]

Map QPSK symbol indices {0,1,2,3} to Gray-coded symbols (cf32).

Parameters:

Name	Type	Description	Default
`syms`	`NDArray[uint8]`	Array of uint8 symbol indices; values must be in {0,1,2,3}. Bits above position 1 are ignored.	required

Returns:

Type	Description
`NDArray[complex64]`	Output.

Examples:

>>> from doppler.wfm import qpsk_map
>>> import numpy as np
>>> idx = np.array([0, 1, 2, 3], dtype=np.uint8)
>>> out = qpsk_map(idx)
>>> [round(float(v.real), 4) for v in out]
[0.7071, -0.7071, 0.7071, -0.7071]
>>> [round(float(v.imag), 4) for v in out]
[0.7071, 0.7071, -0.7071, -0.7071]

wfm_ebno_to_snr_db ¶

wfm_ebno_to_snr_db(
    ebno_db: float,
    bits_per_symbol: int,
    samples_per_symbol: float,
) -> float

Convert Eb/No (dB) to SNR (dB over fs).

Parameters:

Name	Type	Description	Default
`ebno_db`	`float`	Eb/No in dB (energy per bit over noise spectral density).	required
`bits_per_symbol`	`int`	Bits carried per modulation symbol: 1 for BPSK, 2 for QPSK.	required
`samples_per_symbol`	`float`	Oversampling ratio (sps), e.g. 8.0.	required

Returns:

Type	Description
`float`	SNR in dB measured over the full sample-rate bandwidth.

Examples:

>>> from doppler.wfm import wfm_ebno_to_snr_db
>>> round(float(wfm_ebno_to_snr_db(10.0, 2, 8.0)), 4)
3.9794
>>> round(float(wfm_ebno_to_snr_db(10.0, 1, 8.0)), 4)
0.9691

wfm_awgn_amplitude ¶

wfm_awgn_amplitude(
    snr_db: float, signal_power: float
) -> float

AWGN amplitude for a target SNR (dB, over fs) given signal power.

Parameters:

Name	Type	Description	Default
`snr_db`	`float`	Target SNR in dB, referenced to the full sample rate.	required
`signal_power`	`float`	RMS power of the signal (e.g. 1.0 for unit-power complex tones or unit-energy BPSK/QPSK symbols).	required

Returns:

Type	Description
`float`	Per-component AWGN amplitude (sigma for one I or Q channel).

Examples:

>>> from doppler.wfm import wfm_awgn_amplitude
>>> round(float(wfm_awgn_amplitude(10.0, 1.0)), 6)
0.223607
>>> round(float(wfm_awgn_amplitude(0.0, 1.0)), 6)
0.707107

`Plan` — prepare-once stimulus engine¶

A composed scene is a linear form Σ gainₖ·signalₖ + noise, and the expensive DSP lives entirely in the signal terms — invariant across a parameter sweep. prepare(scene) renders and caches each source once; Plan.render / Plan.at then re-materialize any variation (per-source gains / phases / enable, global snr, Monte-Carlo seed) as a cheap re-weighted sum, bit-for-bit identical to a full compose. The stimulus for a detection/BER sweep or a Monte-Carlo campaign that re-runs one scene at many operating points — see the gallery walkthrough.

Plan ¶

A prepared scene that re-materializes parameter variations cheaply.

A composed scene is a sequence of segments, each a linear form Σ gainₖ·signalₖ + noise; the expensive DSP (spreading, pulse shaping, the LO) lives entirely in the signal terms, which do not change when you sweep a level, a phase, the SNR, or the noise seed — nor across a segment's repeats instances (only the AWGN, and any ranged gap length, vary per instance). :class:Plan renders each segment's signal once (bit-identically to a full compose), caches it, and then serves every variation as a cheap re-weighted sum plus a regenerated noise synth per instance — so a BER/Pd curve or a Monte-Carlo campaign that re-runs one scene hundreds of times pays the DSP cost once.

Construct it from anything :func:_spec_json accepts — most often a :class:Composer (or call :func:prepare). The baseline render() (no overrides) is bit-for-bit identical to Composer(scene).compose().

Parameters:

Name	Type	Description	Default
`scene`	`Composer or str or bytes`	The scene to prepare — a :class:`Composer` (or any object with a `to_json()`) or a spec JSON string.	required

Notes

Any number of finite segments (no continuous/repeat scene — that has no fixed capacity); each segment may declare repeats and ranged off_samples/delay_samples (redrawn per instance, deterministic from the Plan's seed). A lone bundled noisy source (one source carrying its own SNR) is supported: its AWGN is reconstructed via a per-instance noise synth rather than an external multiply. Still out of scope, and raising ValueError at construction: a ranged on-time (num_samples — it would invalidate the fixed-length signal cache) or any ranged per-source field (freq/snr/level/f_end — redrawing one would invalidate its cached render). The overridable axes are per-source gains (dBFS levels), phases (radians), enable (drop a source), the global snr (the noise floor), and the Monte-Carlo seed (also redraws any ranged gap length) — applied uniformly across every segment/instance that carries noise. Frequency (Doppler) and multipath delay are planned follow-ups.

Examples:

>>> import numpy as np
>>> from doppler.wfm.compose import Composer, Segment, qpsk, tone, prepare
>>> scene = Composer(
...     Segment.sum(
...         qpsk(snr=12.0, seed=7, sps=8, pn_length=7),
...         tone(freq=1e5, seed=3, sps=8),
...         fs=1e6,
...         num_samples=4096,
...     )
... )
>>> plan = prepare(scene)
>>> len(plan), plan.n_sources
(4096, 2)
>>> # baseline reproduces a full compose exactly
>>> np.array_equal(plan.render(), scene.compose())
True
>>> # sweep SNR with no re-synthesis; each point is a cheap re-weight
>>> curve = {s: plan.at(s) for s in (0.0, 3.0, 6.0, 9.0)}
>>> {s: v.shape for s, v in curve.items()}
{0.0: (4096,), 3.0: (4096,), 6.0: (4096,), 9.0: (4096,)}

n_sources `property` ¶

n_sources: int

Number of cached signal sources (excludes the noise floor).

anchor_seed `property` ¶

anchor_seed: int

The noise seed that reproduces a full compose at the base SNR.

render ¶

render(
    *,
    gains: Sequence[float] | None = None,
    phases: Sequence[float] | None = None,
    enable: Sequence[bool] | None = None,
    snr: float | None = None,
    seed: int | None = None,
) -> NDArray[np.complex64]

Materialize the scene with per-axis overrides applied.

Every argument is optional; omit them all for the baseline (identical to Composer(scene).compose()). gains/phases/enable are per signal source (length :attr:n_sources, in scene order).

Parameters:

Name	Type	Description	Default
`gains`	`sequence of float`	Absolute source levels in dBFS (`0` = unit power). Replaces each source's resolved level; does not move the noise floor.	`None`
`phases`	`sequence of float`	Per-source phase rotations in radians (`0` = identity).	`None`
`enable`	`sequence of bool`	`False` drops a source (an exact `gain=0` term).	`None`
`snr`	`float`	Global SNR in dB — moves only the noise floor (its convention is the anchor source's `snr_mode`).	`None`
`seed`	`int`	Noise seed for this realization (defaults to the scene's, i.e. the value that reproduces a full compose).	`None`

Returns:

Type	Description
`ndarray`	`complex64` samples, length :func:`len`.

at ¶

at(
    snr: float, seed: int | None = None
) -> NDArray[np.complex64]

Scalar fast path: render at one (snr, seed) (no JSON parse).

The hot loop of an SNR sweep or Monte-Carlo run. seed defaults to :attr:anchor_seed (which reproduces a full compose at the scene's base SNR).

sweep ¶

sweep(
    snrs: Sequence[float], *, seed: int | None = None
) -> Iterator[tuple[float, NDArray[np.complex64]]]

Yield (snr, samples) across an SNR list at a fixed noise seed.

A held seed isolates the SNR axis (same noise realization, only the floor moves) — the natural stimulus for a Pd/BER-vs-SNR curve.

monte_carlo ¶

monte_carlo(
    snr: float, n: int, *, seed0: int = 0
) -> Iterator[NDArray[np.complex64]]

Yield n independent noise realizations at a fixed SNR.

Seeds run seed0 … seed0 + n − 1; the signal is identical across draws, only the noise differs.

save ¶

save() -> bytes

Serialize this prepared Plan to a bytes blob.

The blob embeds the scene spec, a build-time fingerprint of the DSP source, and the cached per-source signal buffers, so :func:PlanFromBlob reconstructs an equivalent Plan without re-running prepare()'s DSP. A fingerprint mismatch (the library was rebuilt with different DSP) rebuilds transparently — never silently wrong. All of it happens in C.

dump ¶

dump(path: str | PathLike[str]) -> None

Save this prepared Plan to path (the file form of :meth:save).

Raises :class:OSError if the file cannot be written. The write is done in C, not via Python I/O.

prepare ¶

prepare(scene: Composer | str | bytes) -> Plan

Prepare a scene into a reusable :class:Plan (Plan(scene)).

Examples:

>>> from doppler.wfm.compose import Composer, Segment, qpsk, tone, prepare
>>> plan = prepare(
...     Composer(
...         Segment.sum(
...             qpsk(snr=10.0, seed=1),
...             tone(freq=2e5, seed=2),
...             fs=1e6,
...             num_samples=1024,
...         )
...     )
... )
>>> len(plan)
1024

A prepared Plan can be saved and restored so the one-time DSP is paid once across processes or machines: plan.save() returns the cache as bytes and plan.dump(path) writes it to a file; PlanFromBlob(blob) and PlanFromFile(path) reconstruct a Plan without re-running prepare() (the blob carries a DSP-source fingerprint, so a stale cache transparently rebuilds rather than returning wrong samples).

PlanFromBlob ¶

PlanFromBlob(blob: bytes) -> Plan

Reconstruct a :class:Plan from a blob produced by :meth:Plan.save.

Skips prepare()'s DSP by copying the cached signal buffers out of the blob (rebuilding transparently on a DSP-fingerprint mismatch). The blob's reconstruction is entirely in C.

Examples:

>>> from doppler.wfm import prepare, Composer, Segment, qpsk, PlanFromBlob
>>> scene = Composer(Segment.sum(qpsk(seed=1), fs=1e6, num_samples=1024))
>>> plan = prepare(scene)
>>> restored = PlanFromBlob(plan.save())
>>> import numpy as np
>>> np.array_equal(restored.render(), plan.render())
True

PlanFromFile ¶

PlanFromFile(path: str | PathLike[str]) -> Plan

Reconstruct a :class:Plan from a file produced by :meth:Plan.dump.

The file form of :func:PlanFromBlob; the read is done in C.

Gallery — Async DSSS Receiver: the SPEC waveform through coupled Doppler, CarrierAcquisition: RRC Pulse Shaping, A Crowded Band — Many Signals, One Parallel prepare, DSSS Acquisition — Pd / Pfa vs Es/N0, A 5-Burst DSSS Link — wfmgen's Three Faces, the Full Receiver Chain, Gallery, One Cache Slot for a Whole Background Field, Prepare Once, Sweep Many — the Plan stimulus engine, type="symbols" — Bring Your Own Constellation, Composing a Scene — .sum(), .add(), and Headroom, Waveform I/O — One Capture, Four File Types, Waveform Write — Compose, Write, Read Back, wfmgen — One Engine, Every Waveform Guides — Real-Time Pacing & Timestamping, Reading captures, Writing captures — output & file types, Concepts — the object model, DSSS bursts — a burst train in one declaration, Waveform Generator — wfmgen, Prepare Once, Sweep Many — the Plan engine, Python API, Scenes — multi-segment specs, Streaming — real-time pacing, Waveforms Design — API taxonomy: the DSP building-block hierarchy and its naming axis, DsssReceiver Specifications, Design, MPSK Receiver, Telemetry — zero-cost scalar taps for running pipelines, Waveform amplitude & composition

Python Waveform Generator API — Synth / PN¶

Synth — the eight-type waveform engine¶

Bits (user-defined pattern)¶

Symbols (arbitrary constellation)¶

Chirp (LFM sweep)¶

Clean vs noisy, baseband vs offset¶

RRC pulse shaping (band-limited carriers)¶

PN modulation: length, polynomial, realization¶

Determinism¶

Synth ¶

steps ¶

step ¶

reset ¶

PN — raw LFSR m-sequence¶

PN ¶

reset ¶

generate ¶

generate_max_out ¶

state_bytes ¶

get_state ¶

set_state ¶

destroy ¶

__enter__ ¶

__exit__ ¶

compose — multi-segment composition, writers, and a NATS sink¶

Classes¶

Composer ¶

execute ¶

compose ¶

stream ¶

to_dict ¶

to_sigmf ¶

close ¶

Segment ¶

sum classmethod ¶

add ¶

Timeline ¶

add ¶

StreamSink ¶

clip_fraction property ¶

peak_dbfs property ¶

clipped property ¶

send ¶

track_clipping ¶

close ¶

__enter__ ¶

__exit__ ¶

SampleClock ¶

samples property ¶

underruns property ¶

max_lateness property ¶

pace ¶

stamp ¶

stamp_at ¶

track ¶

reset ¶

resync ¶

close ¶

Module-level helpers¶

rrc_taps ¶

dsss_spread ¶

crc16 ¶

mls_poly ¶

Modulation & SNR helpers¶

bpsk_map ¶

qpsk_map ¶

wfm_ebno_to_snr_db ¶

wfm_awgn_amplitude ¶

Plan — prepare-once stimulus engine¶

Plan ¶

n_sources property ¶

anchor_seed property ¶

render ¶

at ¶

sweep ¶

monte_carlo ¶

save ¶

dump ¶

prepare ¶

PlanFromBlob ¶

`Synth` — the eight-type waveform engine¶

`PN` — raw LFSR m-sequence¶

enter ¶

exit ¶

`compose` — multi-segment composition, writers, and a NATS sink¶

sum `classmethod` ¶

clip_fraction `property` ¶

peak_dbfs `property` ¶

clipped `property` ¶

enter ¶

exit ¶

samples `property` ¶

underruns `property` ¶

max_lateness `property` ¶

`Plan` — prepare-once stimulus engine¶

n_sources `property` ¶

anchor_seed `property` ¶