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 seven-type waveform engine	Generate tone / noise / PN / BPSK / QPSK / chirp / bits, 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.

`Synth` — the six-type waveform engine¶

One declarative engine produces every waveform type, selected by the string type. 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("10110101", sps=4, modulation="bpsk")
preamble = s.steps(s.n_samples)

# 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(np.array([1, 0, 1, 1, 0, 1, 0, 1], np.uint8), modulation="qpsk")

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 `dataclass` ¶

One waveform — the single object that both generates and composes.

A Synth is the configuration of a waveform (tone / noise / PN / BPSK / QPSK). Use it two ways with the same object:

Generate now: :meth:steps / :meth:step lazily spin the C engine and return samples. :meth:reset rewinds it.
Compose: pass one or more into :meth:Segment.sum to mix them over a shared noise floor. Composition reads the config only — it spins no engine, so summing N synths allocates nothing extra.

Build with the :func:tone / :func:qpsk / :func:bpsk / :func:pn / :func:noise helpers, or construct directly.

snr is per-synth because it is self-referential — snr_mode's Eb/No needs this waveform's bits/symbol and samples/symbol. In a multi-source :meth:Segment.sum the composer resolves the per-synth SNRs into one shared floor; a lone synth keeps its bundled AWGN.

Parameters:

Name	Type	Description	Default
`type`	`('tone', 'noise', 'pn', 'bpsk', 'qpsk', 'chirp')`	Waveform kind (`"noise"` is a bare AWGN floor at `level` dBFS; `"chirp"` is a linear-FM sweep — see `f_end`).	`"tone"`
`fs`	`float`	Sample rate (Hz) for standalone generation. Ignored when the synth is used in a :class:`Segment` — the segment owns the rate.	`1000000.0`
`freq`	`float`	Carrier/offset frequency (Hz). For a `chirp` this is the start frequency f_start (the instantaneous frequency at t=0); `f_start=` is accepted as an alias.	`0.0`
`f_end`	`float`	Chirp end frequency (Hz); used only when `type="chirp"`. The instantaneous frequency sweeps linearly `freq → f_end` over the generated length (`steps(n)` standalone, or the segment's `num_samples` in composition), then holds at `f_end`. `f_end < freq` is a down-chirp.	`0.0`
`f_start`	`float`	Readable alias for `freq` (chirp start frequency); folded into `freq` at construction when given.	`None`
`snr`	`float`	SNR in dB under `snr_mode`; `100` (the default) is numerically clean.	`100.0`
`snr_mode`	`('auto', 'fs', 'ebno', 'esno')`	How `snr` is interpreted; `auto` resolves per `type`.	`"auto"`
`seed`	`int`	PRNG / LFSR seed.	`1`
`sps`	`int`	PN/PSK data-source parameters (see :class:`Segment`).	`8`
`pn_length`	`int`	PN/PSK data-source parameters (see :class:`Segment`).	`8`
`pn_poly`	`int`	PN/PSK data-source parameters (see :class:`Segment`).	`8`
`lfsr`	`int`	PN/PSK data-source parameters (see :class:`Segment`).	`8`
`level`	`float`	Level in dBFS (`<= 0`) applied in composition — the synth's output is scaled by `10 ** (level / 20)`; for a `noise` synth this is the floor. Standalone :meth:`steps` generates at unit power (`level` is a composition concern), so it is byte-identical to the bare engine.	`0.0`

Examples:

>>> from doppler.wfm import Synth
>>> Synth(type="tone", fs=1.0, freq=0.0).steps(4).tolist()
[(1+0j), (1+0j), (1+0j), (1+0j)]
>>> # an up-chirp sweeping 100 kHz → 300 kHz over 4096 samples
>>> Synth(type="chirp", fs=1e6, f_start=1e5, f_end=3e5).steps(4096).shape
(4096,)

n_samples `property` ¶

n_samples: int

One full pass of a bits pattern, in samples (n_bits*sps, halved for qpsk). 0 for the streaming types (no natural length).

steps ¶

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

Generate n cf32 samples (spins the engine on first use).

step ¶

step()

Generate one cf32 sample (spins the engine on first use).

reset ¶

reset() -> None

Rewind the engine so generation repeats from sample 0 (no-op if it has not generated yet).

`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`	poly constructor parameter.	`96`
`seed`	`int`	seed constructor parameter.	`1`
`length`	`int`	length constructor parameter.	`7`
`lfsr`	`Literal['galois', 'fibonacci']`	lfsr constructor parameter.	`"galois"`

Examples:

Create with defaults:

>>> from doppler.wfm import PN
>>> obj = PN(poly=96, seed=1, length=7, lfsr="galois")

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() -> 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]`	`n` (the number of chips written; always equal to the request).

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

destroy ¶

destroy() -> None

Release C resources immediately.

`compose` — multi-segment composition, writers, and a ZMQ 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 containers (raw / CSV / BLUE type-1000 / SigMF), and ZmqSink publishes over ZeroMQ. The resolved spec round-trips through JSON, so a capture is fully reproducible.

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

# 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", sample_type="cf32") as w:
    while len(blk := c.execute(4096)):
        w.write(blk)
back = read_iq("frame.cf32", "cf32")      # zero-copy complex64 view

# 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.

Reader is the dual of Writer — it reads a capture back to complex64, auto-detecting the container (BLUE magic / .sigmf-meta sidecar / .csv / raw) and recovering fs/fc/sample type from BLUE and SigMF metadata. All parsing and conversion is in C:

from doppler.wfm import Composer, Writer, Reader

Composer(type="qpsk", sps=8, num_samples=4096).compose()  # ... write it ...
with Reader("capture.blue") as r:          # container auto-detected
    print(r.file_type, r.fs, r.num_samples)
    x = r.read_all()                        # or block-wise: r.read(4096)

For a quick raw-only read with no object, read_iq still works; Reader is the full container-aware dual. Writer pairs with read_iq or Reader; for SigMF, pair a Writer(..., file_type="sigmf") data file with sigmf_meta(...), and for detached BLUE use write_blue_header(...). The ZmqSink 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, ZmqSink

# Stream at the true 1 MS/s instead of as fast as possible.
comp = Composer(type="qpsk", sps=8, continuous=True)
clk = SampleClock(fs=1e6)
with ZmqSink("tcp://0.0.0.0:5555") 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 ¶

A multi-segment waveform generator over a list of :class:Segment.

Construct from an explicit segment list, from a single segment's keyword arguments, or from a JSON spec (:meth:from_json / :meth:from_file). Pull samples a block at a time with :meth:execute, or drain a finite spec to one array with :meth:compose. Usable as a context manager.

Parameters:

Name	Type	Description	Default
`segments`	`Sequence[Segment] or None`	The segments to compose. If `None`, a single segment is built from the remaining keyword arguments (see :class:`Segment`).	`None`
`repeat`	`bool`	Loop the whole sequence after the last segment.	`False`
`continuous`	`bool`	Never finish (implies `repeat`); :meth:`compose` then raises.	`False`
`**segment_kwargs`	`object`	Used to build a single :class:`Segment` when `segments` is `None`.	`{}`

Examples:

>>> from doppler.wfm.compose import Composer
>>> with Composer(type="noise", snr=10.0, num_samples=4096) as c:
...     blk = c.execute(1024)
>>> blk.dtype, len(blk)
(dtype('complex64'), 1024)

segments `property` ¶

segments: list[Segment]

The resolved segment list (defaults filled in).

from_json `classmethod` ¶

from_json(json: str) -> Composer

Build a composer from a JSON spec string (the --from-file schema).

from_file `classmethod` ¶

from_file(path: str | PathLike) -> Composer

Build a composer from a JSON spec file.

execute ¶

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

Generate up to n samples; a shorter (or empty) array marks the end.

compose ¶

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

Drain a finite spec into a single array (raises if continuous).

stream ¶

stream(block: int = 4096, *, realtime: bool | float = False) -> Iterator[NDArray[np.complex64]]

Yield successive blocks — a generator over :meth:execute.

Turns the while len(b := c.execute(n)): boilerplate into for b in c.stream(n):. A finite spec ends when execute drains; a continuous spec streams forever.

This is the Python equivalent of the wfmgen --realtime flag: pass realtime to pace each block to real time (the same timing_core clock the CLI uses). realtime=True paces at the first segment's fs; pass a float to override the rate. Each block is emitted, then the generator sleeps until the next block's deadline before producing it — so the first block is immediate and the long-run rate is exactly fs (POSIX only; raises NotImplementedError off-platform, like :class:SampleClock).

Parameters:

Name	Type	Description	Default
`block`	`int`	Samples per yielded array (the last finite block may be shorter).	`4096`
`realtime`	`bool or float`	Pace to real time. `True` uses `segments[0].fs`; a float sets the sample-clock rate explicitly. Default `False` (as fast as possible).	`False`

Yields:

Type	Description
`NDArray[complex64]`	One block per iteration.

Examples:

>>> from doppler.wfm.compose import Composer
>>> c = Composer(type="tone", freq=1e5, num_samples=1000)
>>> total = sum(len(b) for b in c.stream(256))
>>> total
1000

to_json ¶

to_json() -> str

Serialise the resolved spec to JSON (round-trips via :meth:from_json).

close ¶

close() -> None

Release the underlying C state (idempotent).

Segment `dataclass` ¶

One waveform segment of a composed stream.

Fields mirror the C wfm_segment_t and the wfmgen CLI flags; the string fields (type, snr_mode, lfsr) are mapped to the C enums. Defaults match the CLI: a clean 1024-sample baseband tone at 1 MS/s.

Parameters:

Name	Type	Description	Default
`type`	`('tone', 'noise', 'pn', 'bpsk', 'qpsk', 'chirp')`	Waveform kind.	`"tone"`
`fs`	`float`	Sample rate (Hz).	`1000000.0`
`freq`	`float`	Carrier/offset frequency (Hz); normalised by `fs` internally. For a `chirp` this is the start frequency (`f_start=` is an alias).	`0.0`
`f_end`	`float`	Chirp end frequency (Hz); used only when `type="chirp"`. The sweep `freq → f_end` spans the segment's `num_samples`.	`0.0`
`snr`	`float`	SNR in dB under `snr_mode`; the default (100 dB) is numerically clean.	`100.0`
`snr_mode`	`('auto', 'fs', 'ebno', 'esno')`	How `snr` is interpreted; `auto` resolves per `type`.	`"auto"`
`seed`	`int`	PRNG / LFSR seed (reproducible by default).	`1`
`sps`	`int`	Samples per symbol (PSK) or per chip (PN).	`8`
`pn_length`	`int`	LFSR register length (PN/PSK data source).	`7`
`pn_poly`	`int`	LFSR polynomial; `0` selects the maximal-length poly for `pn_length`.	`0`
`lfsr`	`('galois', 'fibonacci')`	LFSR topology.	`"galois"`
`num_samples`	`int`	On-time length of the segment (samples).	`1024`
`off_samples`	`int`	Trailing zero gap after the segment (samples).	`0`
`level`	`float`	Segment level in dBFS (`<= 0`); the segment's output is scaled by `10 ** (level / 20)`. Default `0` (unit power) is a bit-exact no-op, so the segment's internal SNR is preserved.	`0.0`

sum `classmethod` ¶

sum(*sources: Synth, num_samples: int, off: int = 0, fs: float = 1000000.0) -> 'Segment'

Mix several :class:Synth together over the same num_samples.

The synths mix at the same time (one receiver), so they share one sample rate fs and one noise floor: per-synth snr resolves to a single AWGN source in C (see :class:Synth). off is the trailing gap. This is the multi-source counterpart of the single-source :class:Segment constructor.

Examples:

>>> from doppler.wfm import Segment, qpsk, tone
>>> seg = Segment.sum(
...     qpsk(snr=15), tone(freq=2e5, level=-12), num_samples=4096
... )
>>> len(seg.sources)
2

add ¶

add(*others: 'Segment') -> 'Timeline'

Sequence this segment then others in time → a :class:Timeline.

The time-sequence counterpart of :meth:sum (which mixes sources at the same time): a.add(b) plays a then b back-to-back. Chain with :meth:Timeline.add, then hand the result to :class:Composer.

Examples:

>>> from doppler.wfm import Composer, Segment, qpsk, tone
>>> tl = Segment("tone", freq=1e5, num_samples=1000).add(
...     Segment.sum(qpsk(snr=15), tone(level=-12), num_samples=4096)
... )
>>> len(Composer(tl).compose())
5096

Timeline `dataclass` ¶

An ordered run of :class:Segment played back-to-back in time.

The composer already sequences a segment list; :class:Timeline is the fluent face of that list, built by :meth:Segment.add / :meth:add. It is a plain sequence — iterate it, index it, or pass it straight to :class:Composer.

add ¶

add(*segments: Segment) -> 'Timeline'

Append more segments to the end of the timeline (chainable).

Writer ¶

Stream complex64 samples to a container file.

Wraps the C writer for the four containers the CLI supports. For "sigmf" the samples land in path (use <name>.sigmf-data) and the companion metadata is produced by :func:sigmf_meta; for detached BLUE use :func:write_blue_header.

Parameters:

Name	Type	Description	Default
`path`	`str or PathLike`	Output file (opened in binary mode; truncated).	required
`file_type`	`('raw', 'csv', 'blue', 'sigmf')`	Output container.	`"raw"`
`sample_type`	`('cf32', 'cf64', 'ci32', 'ci16', 'ci8')`	On-disk element type; integer types are quantised from unit-scale.	`"cf32"`
`endian`	`('le', 'be')`	Byte order for multi-byte integer types (ignored for CSV).	`"le"`
`fs`	`float`	Sample rate and centre frequency (Hz), recorded in BLUE/SigMF metadata.	`1000000.0`
`fc`	`float`	Sample rate and centre frequency (Hz), recorded in BLUE/SigMF metadata.	`1000000.0`
`total`	`int`	Expected sample count (lets seekable BLUE headers be patched on close).	`0`

Examples:

>>> import tempfile, os
>>> from doppler.wfm.compose import Composer, Writer
>>> from doppler.wfm.readback import read_iq
>>> x = Composer(type="tone", freq=1e5, num_samples=512).compose()
>>> p = os.path.join(tempfile.mkdtemp(), "cap.cf32")
>>> with Writer(p, sample_type="cf32") as w:
...     _ = w.write(x)
>>> import numpy as np
>>> bool(np.allclose(read_iq(p, "cf32"), x))
True

peak_dbfs `property` ¶

peak_dbfs: float

Largest sample magnitude written, in dBFS (0 dB = full scale). Positive means an integer wire type clipped; the value is the headroom it would need. Readable while open or after :meth:close.

clip_fraction `property` ¶

clip_fraction: float

Fraction (0..1) of I/Q components that saturated. Always 0 unless :meth:track_clipping was enabled; only meaningful for integer types.

clipped `property` ¶

clipped: bool

True if an integer capture ran past full scale (peak > 1).

write ¶

write(iq: NDArray[complex64]) -> int

Write a block of samples; returns the number written.

track_clipping ¶

track_clipping(on: bool = True) -> None

Enable the per-component clip counter (off by default; the peak is always tracked). Call before writing if you want :attr:clip_fraction.

close ¶

close() -> None

Flush, patch any header, and close the file (idempotent).

Reader ¶

Read a capture back to complex64 — the dual of :class:Writer.

The container is auto-detected from the file (BLUE "BLUE" magic, a .sigmf-meta sidecar, the .csv extension, else raw), and self-describing containers (BLUE, SigMF) recover the sample type, byte order, sample rate and centre frequency from their metadata. Headerless raw / CSV take the sample_type / endian hints. All detection, header parsing and wire→unit conversion happen in C; this class is thin glue.

Parameters:

Name	Type	Description	Default
`path`	`str or PathLike`	Capture to read. A BLUE `.det` or SigMF `.sigmf-data` data file resolves its `.hdr` / `.sigmf-meta` sidecar automatically.	required
`sample_type`	`('cf32', 'cf64', 'ci32', 'ci16', 'ci8')`	Wire type for headerless raw / CSV (ignored once BLUE/SigMF metadata is parsed).	`"cf32"`
`endian`	`('le', 'be')`	Byte order for headerless raw.	`"le"`

Examples:

>>> import tempfile, os, numpy as np
>>> from doppler.wfm.compose import Composer, Writer, Reader
>>> x = Composer(type="tone", freq=1e5, num_samples=512).compose()
>>> p = os.path.join(tempfile.mkdtemp(), "cap.blue")
>>> with Writer(p, file_type="blue", fs=1e6) as w:
...     _ = w.write(x)
>>> with Reader(p) as r:            # BLUE self-describes — no hints needed
...     y = r.read_all()
...     print(r.file_type, int(r.fs), bool(np.allclose(y, x)))
blue 1000000 True

file_type `property` ¶

file_type: str

Detected container: "raw" / "csv" / "blue" / "sigmf".

sample_type `property` ¶

sample_type: str

Resolved wire sample type.

endian `property` ¶

endian: str

Resolved byte order.

fs `property` ¶

fs: float

Sample rate (Hz); 0.0 if the container doesn't carry it.

fc `property` ¶

fc: float

Centre frequency (Hz); 0.0 if not recorded.

num_samples `property` ¶

num_samples: int

Total complex samples available; 0 if unknown (a stream).

read ¶

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

Read up to n samples; a shorter (or empty) array marks EOF.

read_all ¶

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

Drain the whole capture into one complex64 array.

close ¶

close() -> None

Close the file (idempotent).

ZmqSink ¶

Publish complex64 samples over a ZeroMQ PUB socket (POSIX only).

Each :meth:send frames the block with its sample rate and centre frequency and the chosen sample_type wire format — the same framing the C CLI's --output zmq://… uses.

Parameters:

Name	Type	Description	Default
`endpoint`	`str`	ZeroMQ endpoint, e.g. `"tcp://0.0.0.0:5555"`.	required
`sample_type`	`('cf32', 'cf64', 'ci32', 'ci16', 'ci8')`	Wire element type.	`"cf32"`

send ¶

send(iq: NDArray[complex64], fs: float, fc: float = 0.0) -> None

Publish a block tagged with its sample rate and centre frequency.

close ¶

close() -> None

Close the publisher (idempotent).

SampleClock ¶

Pace and timestamp a stream against an ideal fs-Hz clock (POSIX).

A :class:SampleClock mimics a hardware sample clock in software. Off one drift-free timeline anchored at construction it does two things:

:meth:pace — sleep so each block leaves at its real-time deadline epoch + n/fs, throttling a producer (e.g. a :class:Composer feeding a :class:ZmqSink) to real time.
:meth:stamp — return the ideal UNIX-epoch-ns time of the next sample, for reproducible capture metadata (SigMF core:datetime, records).

The schedule is anchored, not incremental: every deadline is recomputed from the cumulative sample count against a fixed epoch, so an over- or under-sleep on one block is corrected on the next — the long-run rate is exactly fs, with only bounded per-block jitter. Software on a non-realtime OS gives drift-free average rate, never true sample-clock fidelity; keep blocks large enough (period N/fs ≫ scheduler jitter, say ≥ 1 ms) and let the consumer's buffer absorb the jitter.

When the producer can't keep up — a block takes longer than its N/fs period — that's an underrun: :meth:pace doesn't sleep (it's already behind), counts it (:attr:underruns, :attr:max_lateness), and, if resync=True, re-anchors the timeline to now instead of keeping the (now unreachable) absolute schedule.

Parameters:

Name	Type	Description	Default
`fs`	`float`	Sample rate (Hz). Must be > 0.	required
`resync`	`bool`	Re-anchor the timeline to "now" on each underrun (default: keep the absolute schedule and let the average rate self-heal if it catches up).	`False`

Examples:

>>> from doppler.wfm.compose import SampleClock
>>> clk = SampleClock(fs=1e6)
>>> slack = clk.pace(1000)        # advance 1000 samples (~1 ms) and wait
>>> clk.samples
1000
>>> isinstance(clk.stamp(), int)  # ideal ns timestamp of the next sample
True

samples `property` ¶

samples: int

Cumulative samples advanced through :meth:pace.

underruns `property` ¶

underruns: int

Number of :meth:pace calls that arrived past their deadline.

max_lateness `property` ¶

max_lateness: float

Worst lateness observed (seconds); 0.0 if never behind.

pace ¶

pace(count: int) -> float

Advance count samples and sleep to that block's deadline.

Returns the slack in seconds measured before sleeping: >= 0 means the block was early (and it slept that long); < 0 means it arrived late — an underrun (no sleep, counted).

stamp ¶

stamp() -> int

Ideal UNIX-epoch-ns timestamp of the next sample (index n).

Call before :meth:pace to tag the block you're about to emit, or after to tag the following one.

reset ¶

reset() -> None

Re-anchor to now and zero the counters — a fresh clock at n=0.

resync ¶

resync() -> None

Drop accumulated lateness; pace forward from now (keeps n).

Module-level helpers¶

sigmf_meta ¶

sigmf_meta(*, sample_type: str = 'cf32', endian: str = 'le', fs: float = 1000000.0, fc: float = 0.0, segments: Sequence[Segment]) -> str

Build the SigMF .sigmf-meta JSON for a composed capture.

The capture's segments become per-segment SigMF annotations; pair this with a Writer(..., file_type="sigmf") writing the .sigmf-data companion.

write_blue_header ¶

write_blue_header(path: str | PathLike, *, sample_type: str = 'cf32', endian: str = 'le', fs: float = 1000000.0, fc: float = 0.0, total: int, data_start: float = 0.0, detached: bool = True) -> None

Write a standalone BLUE type-1000 HCB header (the detached .hdr).

The 512-byte header carries the "BLUE" magic, byte order, data_size (total × bytes-per-sample), the type-1000 tag and xdelta = 1/fs; pair it with a detached .det body of raw interleaved I/Q.

Examples:

>>> import os, tempfile
>>> from doppler.wfm.compose import write_blue_header
>>> p = os.path.join(tempfile.mkdtemp(), "cap.hdr")
>>> write_blue_header(p, sample_type="cf32", fs=1e6, total=512)
>>> with open(p, "rb") as f:
...     head = f.read()
>>> head[:4], len(head)
(b'BLUE', 512)

rrc_taps ¶

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

Root-raised-cosine pulse-shaping taps.

Returns 2*span*sps + 1 float32 taps for roll-off beta, sps samples per symbol, and a ±span-symbol support.

Examples:

>>> from doppler.wfm.compose import rrc_taps
>>> t = rrc_taps(0.35, 4, 6)
>>> t.dtype, len(t)
(dtype('float32'), 49)

dsss_spread ¶

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

Direct-sequence spread syms by the ±1 chip code (length ≥ sf).

Each symbol is repeated against the first sf chips, yielding len(syms) * sf output chips.

Examples:

>>> import numpy as np
>>> from doppler.wfm.compose import dsss_spread
>>> syms = np.array([1+0j, -1+0j], dtype=np.complex64)
>>> code = np.array([0, 1, 0, 1], dtype=np.uint8)
>>> dsss_spread(syms, code, 4).shape
(8,)

mls_poly ¶

mls_poly(n: int) -> int

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

Mirrors the table the synth/PN engine uses for pn_poly=0; valid for n in 2..64 (returns 0 otherwise).

Examples:

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

read_iq ¶

read_iq(path: str, sample_type: SampleType = 'cf32', endian: Literal['le', 'be'] = 'le', *, raw: bool = False) -> NDArray

Read an interleaved-I/Q capture into a NumPy array.

Parameters:

Name	Type	Description	Default
`path`	`str`	File written by the `wfmgen` CLI with `--file_type raw` (or a BLUE `.det` data file).	required
`sample_type`	`('cf32', 'cf64', 'ci32', 'ci16', 'ci8')`	The `--sample_type` the file was written with.	`"cf32"`
`endian`	`('le', 'be')`	The `--endian` the file was written with.	`"le"`
`raw`	`(bool, keyword - only)`	If true, return the zero-copy on-disk samples as an `(N, 2)` array (`[:, 0]` = I, `[:, 1]` = Q) in the file's own dtype, with no rescale. Otherwise return a complex array (see Returns).	`False`

Returns:

Type	Description
`NDArray`	With `raw=False` (default): `complex128` for `cf64`, else `complex64` — a zero-copy view for the float types, a rescaled copy for the integer types. With `raw=True`: an `(N, 2)` array in the on-disk dtype.

Examples:

>>> import numpy as np
>>> from doppler.wfm.readback import read_iq
>>> # round-trip a ci16 capture back to unit-scale complex64
>>> iq = read_iq("capture.iq", "ci16")
>>> iq.dtype
dtype('complex64')

Python Waveform Generator API — Synth / PN¶

Synth — the six-type waveform engine¶

Bits (user-defined pattern)¶

Chirp (LFM sweep)¶

Clean vs noisy, baseband vs offset¶

RRC pulse shaping (band-limited carriers)¶

PN modulation: length, polynomial, realization¶

Determinism¶

Synth dataclass ¶

n_samples property ¶

steps ¶

step ¶

reset ¶

PN — raw LFSR m-sequence¶

PN ¶

reset ¶

generate ¶

destroy ¶

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

Classes¶

Composer ¶

segments property ¶

from_json classmethod ¶

from_file classmethod ¶

execute ¶

compose ¶

stream ¶

to_json ¶

close ¶

Segment dataclass ¶

sum classmethod ¶

add ¶

Timeline dataclass ¶

add ¶

Writer ¶

peak_dbfs property ¶

clip_fraction property ¶

clipped property ¶

write ¶

track_clipping ¶

close ¶

Reader ¶

file_type property ¶

sample_type property ¶

endian property ¶

fs property ¶

fc property ¶

num_samples property ¶

read ¶

read_all ¶

close ¶

ZmqSink ¶

send ¶

close ¶

SampleClock ¶

samples property ¶

underruns property ¶

max_lateness property ¶

pace ¶

stamp ¶

reset ¶

resync ¶

Module-level helpers¶

sigmf_meta ¶

write_blue_header ¶

rrc_taps ¶

dsss_spread ¶

mls_poly ¶

read_iq ¶

`Synth` — the six-type waveform engine¶

Synth `dataclass` ¶

n_samples `property` ¶

`PN` — raw LFSR m-sequence¶

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

segments `property` ¶

from_json `classmethod` ¶

from_file `classmethod` ¶

Segment `dataclass` ¶

sum `classmethod` ¶

Timeline `dataclass` ¶

peak_dbfs `property` ¶

clip_fraction `property` ¶

clipped `property` ¶

file_type `property` ¶

sample_type `property` ¶

endian `property` ¶

fs `property` ¶

fc `property` ¶

num_samples `property` ¶

samples `property` ¶

underruns `property` ¶

max_lateness `property` ¶