|
| 1 | +"""Run every file-mode and synthetic example's main() end to end. |
| 2 | +
|
| 3 | +test_examples_smoke.py proves each example PARSES and IMPORTS. About half of |
| 4 | +them had nothing beyond that: argparse surfaces, file loading, the pipeline |
| 5 | +glue, and the output writers of am_receiver, ook_decoder, inspect_capture, |
| 6 | +iq_basics, waterfall and friends never executed under test. This runs each one |
| 7 | +as a subprocess the way a user would, with the smallest arguments that still |
| 8 | +exercise the real path: |
| 9 | +
|
| 10 | +* file-based examples get a short synthetic SigMF capture written by this test |
| 11 | + (an FM station, an OOK burst train, an AM carrier), so no sample_data or |
| 12 | + hardware is needed; |
| 13 | +* plotting examples run with MPLBACKEND=Agg, where plt.show() is a no-op, so |
| 14 | + figures are built but never block; |
| 15 | +* anything that needs a radio is excluded by name -- the stub-radio adapter |
| 16 | + tests cover that seam. |
| 17 | +
|
| 18 | +A non-zero exit, a traceback, or a 120 s timeout fails the example. These are |
| 19 | +smoke tests: they assert the run completed and produced its declared output, |
| 20 | +not the DSP (the core tests and test_examples_demods do that). |
| 21 | +""" |
| 22 | + |
| 23 | +import os |
| 24 | +import subprocess |
| 25 | +import sys |
| 26 | +from pathlib import Path |
| 27 | + |
| 28 | +import numpy as np |
| 29 | +import pytest |
| 30 | + |
| 31 | +from sdr_dsp.io import save_iq |
| 32 | +from sdr_dsp.core import ook_modulate |
| 33 | + |
| 34 | +EXAMPLES = Path(__file__).resolve().parent.parent / "examples" |
| 35 | +TIMEOUT = 120 |
| 36 | + |
| 37 | +# Examples that need a radio, a sound device, or an interactive window loop |
| 38 | +# with no non-interactive mode. Each is covered elsewhere or by name here so |
| 39 | +# the list is explicit, not a silent omission. |
| 40 | +NEEDS_HARDWARE = { |
| 41 | + "hackrf_capture.py", # adapter: tests/test_hackrf_adapter_stub.py |
| 42 | + "hackrf_sink.py", # TX adapter: tests/test_tx_sink.py |
| 43 | + "live_433_capture.py", "live_spectrum.py", "live_fm_listen.py", |
| 44 | + "collect_sample_data.py", # tests/test_collect_sample_data.py |
| 45 | + "fm_band_explorer.py", # interactive UI; tests/test_fm_band_explorer.py |
| 46 | + "sdr_adapter_template.py", # a template, exercised by tests/test_modularity.py |
| 47 | +} |
| 48 | + |
| 49 | + |
| 50 | +# --------------------------------------------------------------------------- |
| 51 | +# synthetic captures (cf32 SigMF, which load_iq reads like any capture) |
| 52 | +# --------------------------------------------------------------------------- |
| 53 | + |
| 54 | +@pytest.fixture(scope="module") |
| 55 | +def captures(tmp_path_factory): |
| 56 | + d = tmp_path_factory.mktemp("captures") |
| 57 | + rng = np.random.default_rng(0) |
| 58 | + |
| 59 | + # FM: 1 kHz tone, 75 kHz deviation, 2 Msps, 0.25 s, a touch of noise |
| 60 | + fs = 2_000_000 |
| 61 | + n = fs // 4 |
| 62 | + t = np.arange(n) / fs |
| 63 | + msg = np.sin(2 * np.pi * 1000 * t) |
| 64 | + fm = 0.5 * np.exp(1j * 2 * np.pi * 75e3 * np.cumsum(msg) / fs) |
| 65 | + fm += 0.005 * (rng.standard_normal(n) + 1j * rng.standard_normal(n)) |
| 66 | + fm_path = d / "fm.sigmf-data" |
| 67 | + save_iq(fm_path, fm.astype(np.complex64), fs, center_freq=100e6) |
| 68 | + |
| 69 | + # OOK: a few bursts of a 1 kbaud pattern at 250 ksps |
| 70 | + fs_ook = 250_000 |
| 71 | + sps = 250 |
| 72 | + bits = np.array([1, 0, 1, 1, 0, 0, 1, 0] * 4) |
| 73 | + burst = ook_modulate(bits, sps) |
| 74 | + gap = np.zeros(20 * sps, dtype=np.complex64) |
| 75 | + ook = np.concatenate([gap, burst, gap, burst, gap]).astype(np.complex64) |
| 76 | + ook += 0.01 * (rng.standard_normal(len(ook)) |
| 77 | + + 1j * rng.standard_normal(len(ook))).astype(np.complex64) |
| 78 | + ook_path = d / "ook.sigmf-data" |
| 79 | + save_iq(ook_path, ook, fs_ook, center_freq=433.92e6) |
| 80 | + |
| 81 | + # AM: carrier at +50 kHz offset, 60% modulation by a 700 Hz tone |
| 82 | + fs_am = 500_000 |
| 83 | + n = fs_am // 2 |
| 84 | + t = np.arange(n) / fs_am |
| 85 | + am = (1 + 0.6 * np.sin(2 * np.pi * 700 * t)) * np.exp(1j * 2 * np.pi * 50e3 * t) |
| 86 | + am = 0.4 * am + 0.005 * (rng.standard_normal(n) + 1j * rng.standard_normal(n)) |
| 87 | + am_path = d / "am.sigmf-data" |
| 88 | + save_iq(am_path, am.astype(np.complex64), fs_am, center_freq=1e6) |
| 89 | + |
| 90 | + return {"fm": str(fm_path), "ook": str(ook_path), "am": str(am_path), |
| 91 | + "dir": d} |
| 92 | + |
| 93 | + |
| 94 | +# --------------------------------------------------------------------------- |
| 95 | +# the table: example -> argv (callables receive the captures dict) |
| 96 | +# --------------------------------------------------------------------------- |
| 97 | + |
| 98 | +CASES = { |
| 99 | + # file-based receivers / inspectors |
| 100 | + "am_receiver.py": lambda c: [c["am"], "--tune", "50000", "--out", |
| 101 | + str(c["dir"] / "am.wav")], |
| 102 | + "fm_receiver.py": lambda c: [c["fm"], "--out", str(c["dir"] / "fm.wav")], |
| 103 | + "ook_decoder.py": lambda c: [c["ook"]], |
| 104 | + "inspect_capture.py": lambda c: [c["fm"], "--max-samples", "200000"], |
| 105 | + "signal_survey.py": lambda c: [c["fm"], c["ook"]], |
| 106 | + "spectrum_analyzer.py": lambda c: [c["fm"], "--no-plot"], |
| 107 | + "waterfall.py": lambda c: [c["fm"], "--max-samples", "200000"], |
| 108 | + "burst_detector.py": lambda c: [c["ook"]], |
| 109 | + "annotate_bursts.py": lambda c: [c["ook"], "--out", |
| 110 | + str(c["dir"] / "annotated.sigmf-data")], |
| 111 | + "channelizer.py": lambda c: [c["fm"]], |
| 112 | + "channel_sweep.py": lambda c: ["--scheme", "fsk", "--msg", "HI"], |
| 113 | + "power_calibration.py": lambda c: ["--save", str(c["dir"] / "cal.json")], |
| 114 | + "live_pipeline_fm.py": lambda c: [c["fm"], "--block", "50000"], |
| 115 | + "cfo_demo.py": lambda c: [c["fm"], "--correct"], |
| 116 | + "dc_offset_demo.py": lambda c: [c["fm"]], |
| 117 | + "filter_explorer.py": lambda c: [c["fm"]], |
| 118 | + "iq_basics.py": lambda c: [c["fm"], "--samples", "500"], |
| 119 | + # synthetic / teaching demos (no file) |
| 120 | + "constellation_recovery.py": lambda c: ["--nsym", "300"], |
| 121 | + "decimation_stages.py": lambda c: ["--n", "100000"], |
| 122 | + "impairment_extraction.py": lambda c: [], |
| 123 | + "matched_filter_demo.py": lambda c: ["--total", "2000"], |
| 124 | + "nyquist_aliasing_demo.py": lambda c: [], |
| 125 | + "resampler_benchmark.py": lambda c: ["--n", "50000"], |
| 126 | + "spectrogram_chirp.py": lambda c: ["--dur", "0.02"], |
| 127 | + "windowing_demo.py": lambda c: [], |
| 128 | + "agc_demo.py": lambda c: [], |
| 129 | + "modulate_demo.py": lambda c: [], |
| 130 | + "differential_psk_demo.py": lambda c: [], |
| 131 | + "dsss_demo.py": lambda c: [], |
| 132 | + "fhss_visualizer.py": lambda c: [], |
| 133 | + "cw_decoder.py": lambda c: [], |
| 134 | + "fsk_decoder.py": lambda c: [], |
| 135 | + "nask_decoder.py": lambda c: [], |
| 136 | + "ssb_receiver.py": lambda c: [], |
| 137 | + "dsb_sc_demo.py": lambda c: [], |
| 138 | + "packet_loopback.py": lambda c: [], |
| 139 | + "two_station_link.py": lambda c: [], |
| 140 | + "transmit_examples.py": lambda c: [], |
| 141 | +} |
| 142 | + |
| 143 | + |
| 144 | +def test_every_example_is_either_run_or_explicitly_excluded(): |
| 145 | + """No example may fall through the cracks: each is in CASES or in |
| 146 | + NEEDS_HARDWARE with a reason. Adding an example without deciding which is |
| 147 | + a failure, on purpose.""" |
| 148 | + names = {p.name for p in EXAMPLES.glob("*.py")} |
| 149 | + covered = set(CASES) | NEEDS_HARDWARE |
| 150 | + missing = sorted(names - covered) |
| 151 | + assert not missing, f"examples with no run test and no exclusion: {missing}" |
| 152 | + stale = sorted(covered - names) |
| 153 | + assert not stale, f"CASES/NEEDS_HARDWARE name files that do not exist: {stale}" |
| 154 | + |
| 155 | + |
| 156 | +def _run(path, argv, cwd): |
| 157 | + env = dict(os.environ, MPLBACKEND="Agg", PYTHONUNBUFFERED="1") |
| 158 | + env.pop("DISPLAY", None) |
| 159 | + return subprocess.run([sys.executable, str(path), *argv], cwd=cwd, |
| 160 | + capture_output=True, text=True, timeout=TIMEOUT, |
| 161 | + env=env) |
| 162 | + |
| 163 | + |
| 164 | +@pytest.mark.parametrize("name", sorted(CASES), ids=lambda n: n) |
| 165 | +def test_example_runs_to_completion(name, captures, tmp_path): |
| 166 | + path = EXAMPLES / name |
| 167 | + argv = CASES[name](captures) |
| 168 | + if "matplotlib" in path.read_text() and not pytest.importorskip("matplotlib"): |
| 169 | + pytest.skip("matplotlib not installed") |
| 170 | + try: |
| 171 | + r = _run(path, argv, cwd=tmp_path) |
| 172 | + except subprocess.TimeoutExpired: |
| 173 | + pytest.fail(f"{name} did not finish in {TIMEOUT}s with args {argv}") |
| 174 | + if r.returncode != 0: |
| 175 | + tail = "\n".join((r.stderr or r.stdout).strip().splitlines()[-12:]) |
| 176 | + pytest.fail(f"{name} exited {r.returncode} with args {argv}:\n{tail}") |
| 177 | + assert "Traceback" not in r.stderr, r.stderr[-800:] |
| 178 | + |
| 179 | + |
| 180 | +def test_am_receiver_wrote_audio(captures): |
| 181 | + out = captures["dir"] / "am.wav" |
| 182 | + if not out.exists(): # the run test for am_receiver.py creates it |
| 183 | + _run(EXAMPLES / "am_receiver.py", CASES["am_receiver.py"](captures), |
| 184 | + cwd=captures["dir"]) |
| 185 | + assert out.exists() and out.stat().st_size > 1000 |
0 commit comments