Capture Replay

rfgen.capture_replay replays an existing, licensed SigMF capture. It is not a synthetic emitter and does not infer an RF source label. Its output remains capture-derived truth after an allowed deterministic transform, with enough metadata to identify the exact source bytes and the derived output interval.

Maturity: implemented, with contract tests for transform order, source-immutability checks, capture-boundary rejection, right-edge behavior, and streamed replay equivalence.

Scope and boundary

The replay input is one local SigMF *.sigmf-data file and its paired *.sigmf-meta file. The constructor also accepts the *.sigmf-meta path or a basename and resolves the pair. Remote object stores, HDF5, random excerpt selection, synthetic waveform generation, label assignment, and dataset writing are outside this module.

The implementation accepts only cf32_le and cf32_be complex-float SigMF data. It requires a finite positive core:sample_rate and a non-empty core:license. When the optional rfgen[sigmf] extra is installed, the SigMF library validates the metadata too; the replay-specific checks remain mandatory in every installation.

SigMFReplay loads the capture into memory and records SHA-256 digests of both source artifacts. Every replay operation recomputes both digests. A changed data or metadata file raises ProvenanceError, rather than returning samples with stale provenance.

Public contract

API	Contract
ReplayQuery(start_sample, num_samples, channel=0)	Immutable source selection. start_sample is zero-based and inclusive; num_samples is positive; channel is zero-based.
ReplayTransform(output_rate_hz, freq_shift_hz=0.0, gain_db=0.0, length_policy=...)	Immutable deterministic transform. All numeric transform values must be finite; output_rate_hz must be positive.
SigMFReplay(uri)	Loads and validates one paired local SigMF artifact, records source digests, and exposes replay() and open_stream().
replay_sigmf(uri, query, transform, seed=None)	One-call convenience form of SigMFReplay(uri).replay(...).
ReplayStream	Contiguous-selection assembler returned by SigMFReplay.open_stream(). It applies one final polyphase resampling operation.
ReplayResult	Frozen pair of complex64 iq and CapturedReplayMetadata.

Right-edge policy

ReplayLengthPolicy.ERROR is the default. A query that extends past the capture then raises ProvenanceError.

ReplayLengthPolicy.CROP_RIGHT returns only available source samples. ReplayLengthPolicy.PAD_RIGHT_ZERO appends source-domain zero samples to meet the requested source length before resampling. In both cases, metadata.original_range names only source samples that existed in the capture, never the derived padding.

Transform and frequency contract

For one replay query, the transform order is fixed and recorded as:

1. Crop the selected SigMF channel.

2. Resample with scipy.signal.resample_poly at rational ratio p / q.

3. Apply linear amplitude gain derived from gain_db.

4. Apply the complex frequency shift at output_rate_hz.

The rational rate ratio is derived from output_rate_hz / core:sample_rate with a denominator limit of 1,000,000. Output length is round(input_length * p / q) using Python’s half-even rounding; the result is cropped or right-zero-padded to that exact length after resampling. Returned IQ is numpy.complex64.

The effective capture frequency comes from captures[*].core:frequency when present, otherwise global.core:frequency. A query may not cross a SigMF capture-metadata boundary, because one derived output cannot honestly carry two source centre frequencies. The reported output centre frequency is the effective source centre frequency plus freq_shift_hz; both are None when the capture has no declared centre frequency.

Provenance returned with every result

CapturedReplayMetadata.as_dict() returns JSON-serializable values with these fields:

Field	Meaning
truth_origin	Always captured; transforms do not relabel captured truth as synthetic.
source_uri	Resolved local file: URI for the *.sigmf-data artifact.
source_data_sha256, source_meta_sha256	SHA-256 digests recorded when the replay object was opened.
license	Required core:license value from the SigMF global metadata.
original_sample_rate_hz, output_sample_rate_hz	Input and requested output rates in hertz.
original_center_freq_hz, output_center_freq_hz	Effective source and derived centre frequencies in hertz, or None.
original_range, output_range	Half-open source and output sample intervals. Output always starts at zero.
rational_p, rational_q	Exact integer rate-conversion ratio used by the operation.
ordered_transforms	crop, resample_poly, gain, then frequency_shift.
seed	Caller-supplied value retained for run provenance. It does not introduce random replay behavior.
channel	Selected SigMF channel.

One-shot replay

from pathlib import Path

from rfgen.capture_replay import ReplayQuery, ReplayTransform, SigMFReplay

capture = SigMFReplay(Path("captures/example.sigmf-data"))
result = capture.replay(
    ReplayQuery(start_sample=0, num_samples=48_000, channel=0),
    ReplayTransform(
        output_rate_hz=1_000_000.0,
        gain_db=-3.0,
        freq_shift_hz=25_000.0,
    ),
    seed=20260714,
)

assert result.metadata.truth_origin == "captured"
assert result.iq.dtype.name == "complex64"

Streamed replay

Use a stream only for adjacent source selections with one channel and one transform. Calling resample_poly independently for each chunk resets filter history, so independently transformed chunks are not equivalent to a one-shot replay. ReplayStream collects contiguous selections and applies the fixed transform sequence once in finish().

append() rejects a channel change, a gap, a call after finish(), or a call after a right-edge crop or pad. It also rejects a capture-metadata boundary. finish() requires at least one append and may be called once. The resulting IQ is byte-equivalent to replaying the same uninterrupted source interval in one call, subject to the same numeric environment.

from rfgen.capture_replay import ReplayQuery, ReplayTransform, SigMFReplay

capture = SigMFReplay("captures/example.sigmf-meta")
stream = capture.open_stream(ReplayTransform(output_rate_hz=1_000_000.0))
stream.append(ReplayQuery(start_sample=0, num_samples=24_000))
stream.append(ReplayQuery(start_sample=24_000, num_samples=24_000))
result = stream.finish()

Failure behavior and limitations

The module fails closed with ProvenanceError for missing paired files, invalid or unsupported SigMF metadata, a missing license, malformed channel count, ambiguous data shape, an unavailable channel, a capture boundary, non-contiguous stream selections, a prohibited right edge, or source mutation.

Replay transforms derive a new sample representation, not a new licensed source. Consumers must preserve the attached license and source digests when writing or distributing the result, and must establish that their intended capture use is authorized. This module records provenance; it does not perform license compatibility analysis or grant distribution rights.

See Also

• Emitter algorithms: synthetic-emitter overview and cross-family invariants.

• Signal catalog: browsable signal-family reference.

• SigMF specification: standard defining the paired metadata and data artifacts.

References

1. SigMF Contributors. Signal Metadata Format (SigMF) Specification v1.2.0. https://sigmf.org/sigmf-spec.pdf. (Metadata fields, capture segments, and raw IQ artifacts.)

2. SciPy Developers. scipy.signal.resample_poly. https://docs.scipy.org/doc/scipy/reference/generated/scipy.signal.resample_poly.html. (Polyphase rational resampling primitive.)