Skip to content

Capturing All Receiver Telemetry

Every MpskReceiver probe captured from one ring

What you're seeing

Every panel is one telemetry probe an track.MpskReceiver exposes, and every trace came out of a single telemetry.MemoryCapture over a single telemetry.Telemetry ring. One set_telemetry attach registers the receiver's own carrier probes and forwards to its symbol-timing child loop, so a cold-start QPSK pull-in leaves a complete record of the whole receiver:

  • rx.lock / rx.tracking / rx.car.locked — the carrier lock EMA rises off its cold-start value, and the two verify-counted decisions flip 0→1 as the receiver hands over to decision-directed tracking.
  • rx.car.freq — the tracked NCO frequency pulls in to the injected 0.0015 cyc/sample offset.
  • rx.car.e / rx.sync.e / rx.sync.ctrl — the carrier discriminator and the timing TED / loop-filter control settle out of the acquisition transient.
  • rx.sync.rate / rx.sync.mu — the tracked samples/symbol settles on ~8.0 and the fractional interpolation phase sweeps its [0, 1) range.
  • rx.sync.lock / rx.sync.locked — the timing lock statistic climbs and its verify-counted decision declares.

Nothing is decimated (decim=1) and nothing is dropped — the summary cell reports the full capture: 11 probes, one 16-byte record per probe per recovered symbol. The x-axis is real time, because each record carries the sample index it was stamped with.

How it works

Two things the caller no longer does. The drain: a MemoryCapture sizes the ring from the probe count and the block, and set_now(i) drains at every boundary — so losslessness is arithmetic rather than a cadence you have to get right, and close() raises if a record was lost anyway. The split: read_dict(index=True) returns {name: (n, values)}, so nothing below filters by probe id, inverts an id-to-name map, or plots an event ordinal in place of a time axis.

import tempfile
from pathlib import Path

import numpy as np

from doppler.telemetry import MemoryCapture, Telemetry
from doppler.track import MpskReceiver
from doppler.wfm import SampleClock

FS = 1e6  # sample rate, and therefore the figure's time axis
BLOCK = 256  # the step of our own loop — and the capture's whole contract

# A QPSK signal at 8 samples/symbol with a residual carrier offset, cold-
# started (init_norm_freq=0) so the pull-in is real; 20 dB matched Es/N0.
rng = np.random.default_rng(1)
idx = rng.integers(0, 4, 4000)
tx = np.exp(1j * (2 * np.pi * idx / 4 + np.pi / 4)).astype(np.complex64)
tx = np.repeat(tx, 8).astype(np.complex64)
k = np.arange(tx.size)
sigma = np.sqrt(8 / (2 * 10 ** (20.0 / 10)))
iq = (
    tx * np.exp(2j * np.pi * 0.0015 * k)
    + rng.normal(0, sigma, tx.size)
    + 1j * rng.normal(0, sigma, tx.size)
).astype(np.complex64)

# ONE ring; attach the receiver at decim=1 = EVERY event on EVERY probe. The
# attach registers the receiver's own probes AND forwards to its child loops,
# so this is the full set of "all available telemetry". Probes must be
# attached BEFORE the capture opens: the ring is sized from the probe table.
tlm = Telemetry()
rx = MpskReceiver(
    m=4,
    sps=8,
    m_out=4,
    init_norm_freq=0.0,
    bn_carrier=0.02,
    bn_timing=0.01,
    acq_to_track=1,
    lock_thresh=0.65,
    warmup_syms=200,
)
rx.set_telemetry(tlm, "rx", 1)

# The capture owns the drain. `set_now(i)` marks the boundary and drains the
# block just finished; no ring size to guess, no read()/concatenate loop, and
# no post-hoc assert standing in for a guarantee. Leaving the block finalizes
# and RAISES if a record was lost — so reaching the next line is itself the
# losslessness proof — but it does not free, so the capture is still readable.
with MemoryCapture(tlm, BLOCK, SampleClock(FS)) as cap:
    for i in range(0, iq.size, BLOCK):
        tlm.set_now(i)
        rx.steps(iq[i : i + BLOCK])

series = cap.read_dict(index=True)  # {name: (sample_index, values)}
recs = cap.records()  # the same data, still 16-byte wire records

# The 16-byte record layout IS the capture format: .tofile() writes exactly
# the TLM16 payload tlm_sink frames onto the wire.
store = Path(tempfile.mkdtemp()) / "mpsk_tlm.tlm16"
recs.tofile(store)

assert rx.tracking == 1  # the receiver handed over to decision-directed track
assert set(series) == set(tlm.probe_names)  # every probe came back by name
assert sum(v.size for _, v in series.values()) == len(recs)  # nothing lost
assert store.stat().st_size == recs.nbytes == 16 * len(recs)
assert np.array_equal(np.fromfile(store, dtype=recs.dtype), recs)

The 16-byte record layout is the capture format. records() returns the exact C dp_tlm_rec_t as a structured array (n:u8, value:f4, probe:u2, flags:u2), so .tofile() writes it with no transformation and those bytes are byte-for-byte the TLM16 payload that dp_tlm_sink frames onto the NATS wire. File storage and streaming fan-out share one format — the capture above verifies the round-trip bit-exact.

The ring stays SPSC: producer (steps) and consumer (the capture's drain) run on one thread together, which is exactly why the boundary drain is both the fastest option and the provable one.

For a capture you never need back in-process, Capture(tlm, block, path, clock) writes those same bytes straight to disk plus a <path>-meta JSON sidecar carrying the probe registry and time base — self-describing, so a reader needs nothing from the process that wrote it. To take the same records across processes live, publish them as TLM16 frames instead; see Many Emitters, One Consumer.

Run it

python src/doppler/examples/mpsk_telemetry_capture_demo.py   # → mpsk_telemetry_capture_demo.png  (~2 s)

See the telemetry API for the probe tables and record layout, M-PSK Receiver for the receiver itself, and Lock Detection for the verify-counted decisions the *.locked / rx.tracking traces come from.