From 167dad4f4415853082a5fc45bad628d9e6413895 Mon Sep 17 00:00:00 2001 From: hunterhubble Date: Mon, 31 Aug 2026 21:19:57 +0000 Subject: [PATCH 1/2] Added FCC Check API calls - No new tests were designed here - Add API Call to see if current tone is FCC compliant with spurious emissions and the occupied bandwidth. - Previously, this was only visual in the spectrum analyzer view. - Returns dict with pass/fail, carrier wave stats (frequence, power, snr, offset), and test stats of occupied bandwidth and spurious emissions tests Signed-off-by: hunterhubble --- src/stream_web/app.py | 98 ++++++++++++++++- src/stream_web/spectrogram.py | 193 +++++++++++++++++++++++----------- 2 files changed, 230 insertions(+), 61 deletions(-) diff --git a/src/stream_web/app.py b/src/stream_web/app.py index 68ba4bd..9e3ff13 100644 --- a/src/stream_web/app.py +++ b/src/stream_web/app.py @@ -25,10 +25,11 @@ import numpy as np from flask import Flask, Response, jsonify, render_template, send_file from flask import request as flask_request -from hubble_satnet_decoder import reset_chipset_stats +from hubble_satnet_decoder import compute_spec_chunk, reset_chipset_stats from . import analysis, config from .processor import processor_main +from .spectrogram import evaluate_fcc_compliance, spectrum_traces from .spectrum_renderer import spectrum_renderer_main from .td_renderer import td_renderer_main @@ -668,6 +669,101 @@ def api_record_analyze(): return resp +_FCC_CHECK_DEFAULT_SECONDS = int(config.SPECTRUM_AVG_CHUNKS * config.SPEC_CHUNK_S) + + +@app.route("/api/fcc_check", methods=["GET"]) +def api_fcc_check(): + """Record N seconds of IQ (default: the same window the spectrum-analyzer + view averages over) and evaluate FCC 15.247 compliance on the strongest + tone in it -- a one-shot, machine-readable version of the pass/fail box + drawn on the spectrum-analyzer overlay. Shares evaluate_fcc_compliance() + with that overlay, so the verdict here and the picture can't disagree. + + Intended flow: `POST /api/tx/start {"mode":"tone"}`, then this call, which + captures the *next* N seconds fresh -- not the rolling display buffer -- + so the result reflects only what was transmitted after the call was made. + """ + seconds, err = _parse_seconds(default=_FCC_CHECK_DEFAULT_SECONDS) + if err: + return err + + if not _capture_lock.acquire(blocking=False): + return jsonify(error="Another capture is already in progress"), 409 + try: + segment = _capture_iq(seconds * config.SAMPLE_RATE) + except CaptureError as ce: + return jsonify(error=str(ce)), ce.status + finally: + _capture_lock.release() + + chunk_n = config.SPEC_CHUNK_SAMPLES + n_chunks = len(segment) // chunk_n + if n_chunks < 1: + return jsonify(error=f"'seconds' must be >= {config.SPEC_CHUNK_S} " + "to fill one FFT window"), 400 + chunks = [compute_spec_chunk(segment[i * chunk_n:(i + 1) * chunk_n]) + for i in range(n_chunks)] + + n_bins = chunks[0].shape[0] + fs = config.SAMPLE_RATE + center_freq_hz = state.lo_freq_hz + freqs_hz = np.linspace(-fs / 2.0, fs / 2.0, n_bins) + center_freq_hz + bin_hz = fs / n_bins + + avg_dB, _peak_dB = spectrum_traces(chunks) + fcc = evaluate_fcc_compliance(freqs_hz, avg_dB, bin_hz) + + analysis_info = { + "seconds": seconds, + "n_chunks": n_chunks, + "nfft": n_bins, + "bin_hz": bin_hz, + "span_hz": fs, + "center_freq_hz": int(center_freq_hz), + "rx_gain_db": state.rx_gain_dB, + "min_snr_db": config.FCC_OVERLAY_MIN_SNR_DB, + "dc_notch_hz": config.SPEC_DC_NOTCH_BINS * bin_hz, + } + + if fcc is None: + return jsonify(state="no_signal", **{"pass": None}, analysis=analysis_info) + + return jsonify( + state="pass" if fcc["overall_ok"] else "fail", + **{"pass": fcc["overall_ok"]}, + carrier={ + "freq_hz": fcc["carrier_freq_hz"], + "offset_from_lo_hz": fcc["carrier_freq_hz"] - center_freq_hz, + "power_db": fcc["carrier_power_db"], + "snr_db": fcc["snr_db"], + }, + checks={ + "occupied_bandwidth": { + "pass": fcc["spacing_ok"], + "rule": "FCC 15.247(a)(1)", + "bw_20db_hz": fcc["bw_20db_hz"], + "limit_hz": fcc["bw_limit_hz"], + "channel_spacing_hz": fcc["channel_spacing_hz"], + }, + "spurious_emissions": { + "pass": fcc["spurs_ok"], + "rule": "FCC 15.247(d)", + "n_spurs": len(fcc["spurs"]), + "n_over_limit": fcc["n_spur_fail"], + "worst_dbc": fcc["worst_dbc"], + "limit_dbc": -fcc["spur_limit_dbc"], + }, + }, + spurs=[ + {"freq_hz": s["freq_hz"], "power_db": s["power_db"], + "dbc": s["dbc"], "pass": s["pass"]} + for s in fcc["spurs"] + ], + analysis=analysis_info, + ) + + # =========================================================================== # TX API routes # =========================================================================== diff --git a/src/stream_web/spectrogram.py b/src/stream_web/spectrogram.py index e0a8261..c9a87d3 100644 --- a/src/stream_web/spectrogram.py +++ b/src/stream_web/spectrogram.py @@ -124,23 +124,54 @@ def render_spec_image(chunks: list[np.ndarray], detections: list[dict] | None = # Spectrum analyzer (power vs frequency) rendering # =========================================================================== -def _draw_fcc_tone_overlay(ax, freqs_mhz: np.ndarray, avg_dB: np.ndarray, - bin_hz: float) -> None: - """Overlay an FCC 15.247 FHSS check on the strongest tone in ``avg_dB``. - - Only draws when a tone stands at least ``FCC_OVERLAY_MIN_SNR_DB`` above the - noise-floor median. Aligned to the detected peak, it shows the allocated - channel band (``config.CHANNEL_SPACING``) and the 20 dB bandwidth, and - checks ยง15.247(a)(1): carrier separation must be >= 2/3 of the 20 dB - bandwidth (equivalently, 20 dB BW <= 1.5 x channel spacing). +def spectrum_traces(chunks: list[np.ndarray]) -> tuple[np.ndarray, np.ndarray]: + """Collapse Sxx_dB chunks over the time axis into (average, peak-hold) traces. + + The average is a proper linear-power mean (a mean of dB understates real + power); the peak hold is the per-bin max over the window. + + Both get the DC notch interpolated away: compute_spec_chunk zeroes the DC + bin (a -120 dB notch at the exact centre frequency) and LO leakage smears + into a few neighbours. For the display that is cosmetic; for the compliance + check it also stops residual LO leakage from being scored as the carrier. + + Callers window the input themselves (``chunks[-N:]``); this does not slice. + """ + Sxx_dB = np.concatenate(chunks, axis=1) + lin = np.power(10.0, Sxx_dB / 10.0) + avg_dB = 10.0 * np.log10(np.mean(lin, axis=1) + 1e-12) + peak_dB = np.max(Sxx_dB, axis=1) + _interpolate_dc(avg_dB, config.SPEC_DC_NOTCH_BINS) + _interpolate_dc(peak_dB, config.SPEC_DC_NOTCH_BINS) + return avg_dB, peak_dB + + +def evaluate_fcc_compliance(freqs_hz: np.ndarray, avg_dB: np.ndarray, + bin_hz: float) -> dict | None: + """Evaluate FCC 15.247 compliance for the strongest tone in ``avg_dB``. + + Pure measurement -- no plotting, no I/O. Returns ``None`` when no tone + stands at least ``config.FCC_OVERLAY_MIN_SNR_DB`` above the noise-floor + median; the caller decides whether that means "draw nothing" (the spectrum + overlay) or "no_signal" (``/api/fcc_check``). + + Two checks, both keyed to the detected carrier: + + * **15.247(a)(1)** -- carrier separation must be >= 2/3 of the 20 dB + bandwidth, equivalently 20 dB BW <= 1.5 x ``config.CHANNEL_SPACING``. + * **15.247(d)** -- peaks outside the carrier's own skirt/channel must sit + at least ``config.SPUR_LIMIT_DBC`` dB below the carrier. + + ``freqs_hz`` gives the absolute frequency of each bin (ascending, centred + on the LO) and ``bin_hz`` the bin width. Bin indices are returned next to + the physical values so the renderer can draw from exactly the numbers the + API reports -- this function is the single source of truth for both. """ noise_dB = float(np.median(avg_dB)) peak_idx = int(np.argmax(avg_dB)) peak_dB = float(avg_dB[peak_idx]) if peak_dB - noise_dB < config.FCC_OVERLAY_MIN_SNR_DB: - return # no strong tone -> no overlay - - peak_mhz = float(freqs_mhz[peak_idx]) + return None # no strong tone # 20 dB-down bandwidth: walk out from the peak until the trace drops 20 dB. thr_dB = peak_dB - 20.0 @@ -155,6 +186,80 @@ def _draw_fcc_tone_overlay(ax, freqs_mhz: np.ndarray, avg_dB: np.ndarray, spacing_hz = config.CHANNEL_SPACING spacing_ok = (2.0 / 3.0) * bw_hz <= spacing_hz + # -- Spurious emissions: peaks elsewhere in the band vs the -N dBc limit -- + limit_dbc = config.SPUR_LIMIT_DBC + sep_bins = max(1, int(config.SPUR_MIN_SEP_KHZ * 1e3 / bin_hz)) + cand, _ = find_peaks( + avg_dB, + height=noise_dB + config.SPUR_MIN_SNR_DB, + distance=sep_bins, + prominence=config.SPUR_MIN_PROMINENCE_DB, + ) + # Exclude the carrier and its own skirt/channel from the spur list. + guard = max(peak_idx - li, ri - peak_idx, int(spacing_hz / 2 / bin_hz)) + 3 + + spurs: list[dict] = [] + n_spur_fail = 0 + worst_dbc = None + for p in (int(c) for c in cand): + if abs(p - peak_idx) <= guard: + continue + dbc = float(avg_dB[p]) - peak_dB # negative: dB below carrier + if worst_dbc is None or dbc > worst_dbc: + worst_dbc = dbc + spur_ok = dbc <= -limit_dbc + if not spur_ok: + n_spur_fail += 1 + spurs.append({ + "idx": p, + "freq_hz": float(freqs_hz[p]), + "power_db": float(avg_dB[p]), + "dbc": dbc, + "pass": spur_ok, + }) + + return { + "carrier_idx": peak_idx, + "carrier_freq_hz": float(freqs_hz[peak_idx]), + "carrier_power_db": peak_dB, + "noise_floor_db": noise_dB, + "snr_db": peak_dB - noise_dB, + "bw_20db_hz": bw_hz, + "bw_lo_idx": li, + "bw_hi_idx": ri, + "bw_threshold_db": thr_dB, + "bw_limit_hz": 1.5 * spacing_hz, + "channel_spacing_hz": spacing_hz, + "spacing_ok": spacing_ok, + "spur_limit_dbc": limit_dbc, + "spur_limit_db": peak_dB - limit_dbc, + "spurs": spurs, + "n_spur_fail": n_spur_fail, + "worst_dbc": worst_dbc, + "spurs_ok": n_spur_fail == 0, + "overall_ok": spacing_ok and n_spur_fail == 0, + } + + +def _draw_fcc_tone_overlay(ax, freqs_mhz: np.ndarray, fcc: dict) -> None: + """Draw the FCC 15.247 overlay from an :func:`evaluate_fcc_compliance` result. + + Pure presentation: every number rendered here comes out of ``fcc``, so the + on-screen verdict and ``/api/fcc_check`` cannot drift apart. + """ + peak_idx = fcc["carrier_idx"] + peak_mhz = float(freqs_mhz[peak_idx]) + peak_dB = fcc["carrier_power_db"] + li, ri = fcc["bw_lo_idx"], fcc["bw_hi_idx"] + thr_dB = fcc["bw_threshold_db"] + bw_hz = fcc["bw_20db_hz"] + spacing_hz = fcc["channel_spacing_hz"] + spacing_ok = fcc["spacing_ok"] + limit_dbc = fcc["spur_limit_dbc"] + spurs = fcc["spurs"] + n_spur_fail = fcc["n_spur_fail"] + worst_dbc = fcc["worst_dbc"] + pass_col, fail_col = "#4ade80", "#f87171" # Distinct magenta accent for the carrier geometry so the marker doesn't # blend with the cyan average / orange peak-hold traces. @@ -174,44 +279,22 @@ def _draw_fcc_tone_overlay(ax, freqs_mhz: np.ndarray, avg_dB: np.ndarray, ax.plot([xf], [thr_dB], marker="|", markersize=10, markeredgewidth=1.6, color=accent, zorder=6) - # -- Spurious emissions: peaks elsewhere in the band vs the -N dBc limit --- - limit_dbc = config.SPUR_LIMIT_DBC - limit_dB = peak_dB - limit_dbc + limit_dB = fcc["spur_limit_db"] ax.axhline(limit_dB, color="#cbd5e1", linewidth=1.0, linestyle="--", alpha=0.6, zorder=4) ax.text(freqs_mhz[-1], limit_dB, f"-{limit_dbc:.0f} dBc ", color="#cbd5e1", fontsize=7, va="bottom", ha="right", fontfamily="monospace", alpha=0.85, zorder=4) - sep_bins = max(1, int(config.SPUR_MIN_SEP_KHZ * 1e3 / bin_hz)) - cand, _ = find_peaks( - avg_dB, - height=noise_dB + config.SPUR_MIN_SNR_DB, - distance=sep_bins, - prominence=config.SPUR_MIN_PROMINENCE_DB, - ) - # Exclude the carrier and its own skirt/channel from the spur list. - guard = max(peak_idx - li, ri - peak_idx, int(spacing_hz / 2 / bin_hz)) + 3 - spurs = [int(p) for p in cand if abs(int(p) - peak_idx) > guard] - - n_spur_fail = 0 - worst_dbc = None - for p in spurs: - dbc = float(avg_dB[p]) - peak_dB # negative: dB below carrier - if worst_dbc is None or dbc > worst_dbc: - worst_dbc = dbc - spur_ok = dbc <= -limit_dbc - if not spur_ok: - n_spur_fail += 1 - scol = pass_col if spur_ok else fail_col - ax.plot([freqs_mhz[p]], [avg_dB[p]], marker="v", markersize=7, - color=scol, zorder=6) - ax.text(freqs_mhz[p], avg_dB[p] + 1.5, f"{dbc:.0f}", color=scol, - fontsize=7, va="bottom", ha="center", fontfamily="monospace", - zorder=6) + for s in spurs: + scol = pass_col if s["pass"] else fail_col + ax.plot([freqs_mhz[s["idx"]]], [s["power_db"]], marker="v", + markersize=7, color=scol, zorder=6) + ax.text(freqs_mhz[s["idx"]], s["power_db"] + 1.5, f"{s['dbc']:.0f}", + color=scol, fontsize=7, va="bottom", ha="center", + fontfamily="monospace", zorder=6) - overall_ok = spacing_ok and n_spur_fail == 0 - box_col = pass_col if overall_ok else fail_col + box_col = pass_col if fcc["overall_ok"] else fail_col def _row(label: str, value: str) -> str: return f"{label:<9}{value}" @@ -253,25 +336,13 @@ def render_spectrum_image(chunks: list[np.ndarray], lo_freq_hz: float) -> bytes: # Average over a shorter window than the spectrogram so the trace reacts # faster to changes. chunks = chunks[-config.SPECTRUM_AVG_CHUNKS:] - Sxx_dB = np.concatenate(chunks, axis=1) - n_bins = Sxx_dB.shape[0] - - # Average in the linear-power domain (mean of dB understates real power), - # and hold the max over the visible window as a peak trace. - lin = np.power(10.0, Sxx_dB / 10.0) - avg_dB = 10.0 * np.log10(np.mean(lin, axis=1) + 1e-12) - peak_dB = np.max(Sxx_dB, axis=1) - - # compute_spec_chunk zeroes the DC bin (a -120 dB notch at the exact centre - # frequency) and LO leakage smears into a few neighbours; interpolate across - # +-SPEC_DC_NOTCH_BINS so the trace isn't a spike/notch at our frequency of - # interest. Cosmetic only -- these traces never feed the decoder. - _interpolate_dc(avg_dB, config.SPEC_DC_NOTCH_BINS) - _interpolate_dc(peak_dB, config.SPEC_DC_NOTCH_BINS) + n_bins = chunks[0].shape[0] + avg_dB, peak_dB = spectrum_traces(chunks) # Frequency axis: bins run -fs/2 .. +fs/2 (ascending), centred on the LO. fs = config.SAMPLE_RATE - freqs_mhz = (np.linspace(-fs / 2.0, fs / 2.0, n_bins) + lo_freq_hz) / 1e6 + freqs_hz = np.linspace(-fs / 2.0, fs / 2.0, n_bins) + lo_freq_hz + freqs_mhz = freqs_hz / 1e6 center_mhz = lo_freq_hz / 1e6 dpi = 100 @@ -298,7 +369,9 @@ def render_spectrum_image(chunks: list[np.ndarray], lo_freq_hz: float) -> bytes: ax.set_xlim(freqs_mhz[0], freqs_mhz[-1]) # FCC compliance overlay locked onto the strongest tone (skipped if none). - _draw_fcc_tone_overlay(ax, freqs_mhz, avg_dB, fs / n_bins) + fcc = evaluate_fcc_compliance(freqs_hz, avg_dB, fs / n_bins) + if fcc is not None: + _draw_fcc_tone_overlay(ax, freqs_mhz, fcc) ax.text(center_mhz, y_hi, f" {center_mhz:.5f} MHz", color="#3399ff", fontsize=8, fontweight="bold", va="top", ha="left", From af11e1cba7529f6ee5e472202da3ead5025a1184 Mon Sep 17 00:00:00 2001 From: hunterhubble Date: Tue, 1 Sep 2026 22:40:32 +0000 Subject: [PATCH 2/2] Added tests to the spectrum_renderer.py - Added tests to generate artificial signals and verify if they pass/fai - Testing clean tones, noise only, wide band tone, nearby spurs, and dict formatting Signed-off-by: hunterhubble --- tests/test_spectrum.py | 209 +++++++++++++++++++++++++++++++++++++++++ 1 file changed, 209 insertions(+) create mode 100644 tests/test_spectrum.py diff --git a/tests/test_spectrum.py b/tests/test_spectrum.py new file mode 100644 index 0000000..89408db --- /dev/null +++ b/tests/test_spectrum.py @@ -0,0 +1,209 @@ +"""No-hardware tests for the FCC 15.247 compliance check. + +Two layers, both without an SDR: + 1. TestEvaluateFccCompliance -- the pure math (evaluate_fcc_compliance / + spectrum_traces) against synthetic spectra built the same way real data + flows: raw IQ -> compute_spec_chunk -> spectrum_traces. No Flask, no + capture, no lock. + 2. TestFccCheckEndpoint -- the /api/fcc_check route's wiring (status codes, + JSON shape, the capture-lock and bad-input error paths), with + _capture_iq monkeypatched to hand back synthetic IQ instead of touching + hardware. + +Real-signal, real-hardware validation lives in hitl-endpoint's +tests/test_fcc_emissions.py -- that's the only layer that can catch an actual +RF/gain problem; this file exists so a bug in the pass/fail math itself +doesn't have to wait for that rig to be caught. +""" + +import numpy as np +from hubble_satnet_decoder import compute_spec_chunk + +from stream_web import app as app_module +from stream_web import config +from stream_web.app import _capture_lock, app +from stream_web.spectrogram import evaluate_fcc_compliance, spectrum_traces + +SR = config.SAMPLE_RATE +# Arbitrary test LO -- evaluate_fcc_compliance only cares about frequencies +# relative to it, so this doesn't need to match any real config default. +_TEST_LO_HZ = 903_000_000.0 + + +# --------------------------------------------------------------------------- +# Helpers -- build synthetic signals the same way the real pipeline does +# --------------------------------------------------------------------------- + +def _make_chunks(tones, seconds=5.0, noise_std=0.01, seed=0): + """Sxx_dB chunks for a sum of CW tones plus noise, via the real compute_spec_chunk. + + tones: list of (offset_hz, amplitude) pairs, offset from _TEST_LO_HZ. + """ + rng = np.random.default_rng(seed) + n_chunks = int(seconds / config.SPEC_CHUNK_S) + n = config.SPEC_CHUNK_SAMPLES + chunks = [] + for c in range(n_chunks): + t = (np.arange(n) + c * n) / SR + iq = rng.normal(0, noise_std, n) + 1j * rng.normal(0, noise_std, n) + for offset_hz, amp in tones: + iq = iq + amp * np.exp(2j * np.pi * offset_hz * t) + chunks.append(compute_spec_chunk(iq.astype(np.complex64))) + return chunks + + +def _make_chirp_chunks(f0_hz, f1_hz, seconds=5.0, amp=0.6, noise_std=0.01, seed=0): + """Sxx_dB chunks for a tone swept f0_hz -> f1_hz within every chunk. + + A single CW tone always measures a narrow 20 dB bandwidth (a few hundred + Hz) -- there's no way to make one fail the occupied-bandwidth check. + Sweeping the frequency within each chunk's window spreads real, contiguous + energy across a wide band, the way an actual over-modulated or drifting + carrier would. + """ + rng = np.random.default_rng(seed) + n_chunks = int(seconds / config.SPEC_CHUNK_S) + n = config.SPEC_CHUNK_SAMPLES + dur = n / SR + k = (f1_hz - f0_hz) / dur # Hz/sec sweep rate, resets every chunk + chunks = [] + for _c in range(n_chunks): + t = np.arange(n) / SR + phase = 2 * np.pi * (f0_hz * t + 0.5 * k * t**2) + iq = rng.normal(0, noise_std, n) + 1j * rng.normal(0, noise_std, n) + iq = iq + amp * np.exp(1j * phase) + chunks.append(compute_spec_chunk(iq.astype(np.complex64))) + return chunks + + +def _evaluate(chunks): + n_bins = chunks[0].shape[0] + freqs_hz = np.linspace(-SR / 2.0, SR / 2.0, n_bins) + _TEST_LO_HZ + bin_hz = SR / n_bins + avg_dB, _peak_dB = spectrum_traces(chunks) + return evaluate_fcc_compliance(freqs_hz, avg_dB, bin_hz) + + +def _make_tone_iq(n_samples, offset_hz=100_000.0, amp=1.0, noise_std=0.01, seed=0): + """Raw complex64 IQ (not pre-chunked) for the Flask-layer tests below -- + this is what _capture_iq hands back in the real code path.""" + rng = np.random.default_rng(seed) + t = np.arange(n_samples) / SR + iq = rng.normal(0, noise_std, n_samples) + 1j * rng.normal(0, noise_std, n_samples) + iq = iq + amp * np.exp(2j * np.pi * offset_hz * t) + return iq.astype(np.complex64) + + +# --------------------------------------------------------------------------- +# 1. Pure math -- evaluate_fcc_compliance / spectrum_traces +# --------------------------------------------------------------------------- + + +class TestEvaluateFccCompliance: + def test_noise_only_returns_none(self): + assert _evaluate(_make_chunks([])) is None + + def test_clean_tone_passes(self): + result = _evaluate(_make_chunks([(100_000, 1.0)])) + assert result is not None + assert result["overall_ok"] is True + assert result["spacing_ok"] is True + assert result["spurs_ok"] is True + assert result["n_spur_fail"] == 0 + assert result["snr_db"] > config.FCC_OVERLAY_MIN_SNR_DB + + def test_wide_tone_fails_occupied_bandwidth(self): + # 60 kHz sweep centred on +100 kHz -- well past the 38.625 kHz limit + # (1.5x the 25.75 kHz channel spacing). + result = _evaluate(_make_chirp_chunks(70_000, 130_000)) + assert result is not None + assert result["spacing_ok"] is False + assert result["bw_20db_hz"] > result["bw_limit_hz"] + assert result["overall_ok"] is False + + def test_strong_nearby_spur_fails_spurious_emissions(self): + result = _evaluate(_make_chunks([(100_000, 1.0), (150_000, 0.2)])) + assert result is not None + assert result["spurs_ok"] is False + assert result["n_spur_fail"] == 1 + assert result["overall_ok"] is False + spur = next(s for s in result["spurs"] if not s["pass"]) + assert spur["dbc"] > -result["spur_limit_dbc"] # less than 20 dB down + + def test_weak_distant_spur_still_detected_but_passes(self): + result = _evaluate(_make_chunks([(100_000, 1.0), (250_000, 0.03)])) + assert result is not None + assert result["spurs_ok"] is True + assert result["overall_ok"] is True + # It should still show up in the list -- "passes" isn't "invisible". + assert len(result["spurs"]) == 1 + assert result["spurs"][0]["pass"] is True + + def test_result_schema(self): + """Contract test: every key the API/overlay code reads must be present + and correctly typed, so a refactor can't silently drop one.""" + result = _evaluate(_make_chunks([(100_000, 1.0), (250_000, 0.03)])) + assert isinstance(result["carrier_freq_hz"], float) + assert isinstance(result["carrier_power_db"], float) + assert isinstance(result["snr_db"], float) + assert isinstance(result["bw_20db_hz"], float) + assert isinstance(result["bw_limit_hz"], float) + assert isinstance(result["channel_spacing_hz"], float) + assert isinstance(result["spacing_ok"], bool) + assert isinstance(result["spur_limit_dbc"], float) + assert isinstance(result["spurs"], list) + assert isinstance(result["n_spur_fail"], int) + assert isinstance(result["spurs_ok"], bool) + assert isinstance(result["overall_ok"], bool) + for spur in result["spurs"]: + assert {"idx", "freq_hz", "power_db", "dbc", "pass"} <= spur.keys() + + +# --------------------------------------------------------------------------- +# 2. Flask route -- /api/fcc_check +# --------------------------------------------------------------------------- + + +class TestFccCheckEndpoint: + @staticmethod + def _client(): + app.config["TESTING"] = True + return app.test_client() + + def test_clean_tone_passes(self, monkeypatch): + monkeypatch.setattr(app_module, "_capture_iq", + lambda n: _make_tone_iq(n)) + resp = self._client().get("/api/fcc_check") + assert resp.status_code == 200 + body = resp.get_json() + assert body["state"] == "pass" + assert body["pass"] is True + assert "carrier" in body and "checks" in body and "analysis" in body + assert body["checks"]["occupied_bandwidth"]["pass"] is True + assert body["checks"]["spurious_emissions"]["pass"] is True + + def test_no_signal_reports_no_signal_state(self, monkeypatch): + monkeypatch.setattr(app_module, "_capture_iq", + lambda n: _make_tone_iq(n, amp=0.0)) + resp = self._client().get("/api/fcc_check") + assert resp.status_code == 200 + body = resp.get_json() + assert body["state"] == "no_signal" + assert body["pass"] is None + + def test_busy_capture_lock_returns_409(self, monkeypatch): + monkeypatch.setattr(app_module, "_capture_iq", + lambda n: _make_tone_iq(n)) + _capture_lock.acquire() + try: + resp = self._client().get("/api/fcc_check") + assert resp.status_code == 409 + finally: + _capture_lock.release() + + def test_invalid_seconds_returns_400(self, monkeypatch): + monkeypatch.setattr(app_module, "_capture_iq", + lambda n: _make_tone_iq(n)) + client = self._client() + assert client.get("/api/fcc_check?seconds=abc").status_code == 400 + assert client.get("/api/fcc_check?seconds=0").status_code == 400