feat: /api/iq_capture + /api/record_analyze signal diagnostics

CLAUDE.md

@@ -43,6 +43,8 @@ docker run -p 8050:8050 sdr-docker
 ### RX
 - `GET /api/status` — system status, RX metrics, peak power; includes `sdr_connected` (bool) indicating whether the SDR hardware is currently open
 - `GET /api/packets` — poll-and-drain decoded packets (NDJSON)
+- `GET /api/iq_capture?seconds=N` — record N s (1–60) of raw IQ forward from now; returns a `.npy` (complex64) download with `X-Sample-Rate-Hz` / `X-Center-Freq-Hz` / `X-N-Samples` headers
+- `GET /api/record_analyze?seconds=N` — record N s (1–30) and return a plain-text diagnostic report file: per-symbol timing (drift/rate/duration/gap) and per-symbol frequency vs expected channel window for the representative packet, plus synth-res/chipset, amplitude/SNR, RS corrections, regime + diagnosis; noise/mis-hits filtered by preamble strength; backs `hubblenetwork sat record`
 
 ### TX
 - `POST /api/tx/start` — start TX (`{"mode":"tone"}` or `{"mode":"packet","file":"<name>"}`)

src/stream_web/app.py

@@ -26,7 +26,7 @@
 from flask import request as flask_request
 from hubble_satnet_decoder import reset_chipset_stats
 
-from . import config
+from . import analysis, config
 from .processor import processor_main
 
 # GNU Radio imports — deferred so the app can be imported without GNU Radio
@@ -466,6 +466,159 @@ def api_packets():
     return Response(payload, mimetype="application/x-ndjson")
 
 
+_IQ_CAPTURE_MAX_SECONDS = 60
+_RECORD_ANALYZE_MAX_SECONDS = 30
+
+# Only one capture may run at a time. Each holds a Flask worker for up to its full
+# duration, and concurrent captures would contend for both workers and buffer
+# bandwidth; a second request is rejected rather than queued. Shared by
+# /api/iq_capture and /api/record_analyze.
+_capture_lock = threading.Lock()
+
+
+class CaptureError(Exception):
+    """Raised by _capture_iq; carries the HTTP status the caller should return."""
+
+    def __init__(self, message: str, status: int):
+        super().__init__(message)
+        self.status = status
+
+
+def _samples_advanced(start: int, end: int, buf_size: int) -> int:
+    """Number of samples written from *start* to *end* in a circular buffer."""
+    if end >= start:
+        return end - start
+    return buf_size - start + end
+
+
+def _capture_iq(n_samples: int) -> np.ndarray:
+    """Capture *n_samples* of IQ forward from now out of the shared ring buffer.
+
+    Copies straight into one pre-allocated array (single copy). Raises CaptureError on
+    SDR stall (504) or if the RX laps the region mid-copy (500). Caller must hold
+    _capture_lock.
+    """
+    buf_size = config.IQ_BUFFER_SIZE
+    chunk_n = buf_size // 2
+    segment = np.empty(n_samples, dtype=np.complex64)
+    filled = 0
+    chunk_start = int(state.buf_write_idx)
+
+    while filled < n_samples:
+        this_n = min(chunk_n, n_samples - filled)
+        deadline = time.monotonic() + (this_n / config.SAMPLE_RATE) + 5.0
+
+        while _samples_advanced(chunk_start, int(state.buf_write_idx), buf_size) < this_n:
+            if time.monotonic() > deadline:
+                raise CaptureError("Timed out waiting for IQ samples from SDR", 504)
+            time.sleep(0.05)
+
+        e = (chunk_start + this_n) % buf_size
+        if chunk_start < e:
+            segment[filled:filled + this_n] = state.iq_buffer[chunk_start:e]
+        else:
+            n1 = buf_size - chunk_start
+            segment[filled:filled + n1] = state.iq_buffer[chunk_start:]
+            segment[filled + n1:filled + this_n] = state.iq_buffer[:e]
+
+        # Integrity guard: the region was read lock-free while the RX kept writing. The
+        # RX only overwrites chunk_start after lapping a full buffer past it; if the copy
+        # stalled that long, the head is now < this_n ahead (it wrapped past chunk_start)
+        # and the samples are corrupt.
+        if _samples_advanced(chunk_start, int(state.buf_write_idx), buf_size) < this_n:
+            raise CaptureError(
+                "SDR overran the capture buffer (host too slow); capture aborted", 500)
+
+        chunk_start = e
+        filled += this_n
+
+    return segment
+
+
+def _parse_seconds(default=None, maximum=_IQ_CAPTURE_MAX_SECONDS):
+    """Parse/validate the 'seconds' query param.
+
+    Returns ``(seconds, None)`` on success or ``(None, (response, status))`` on error.
+    """
+    raw = flask_request.args.get("seconds")
+    if raw is None:
+        if default is not None:
+            return default, None
+        return None, (jsonify(error="'seconds' query parameter is required"), 400)
+    try:
+        seconds = int(raw)
+    except (ValueError, TypeError):
+        return None, (jsonify(error="'seconds' must be an integer"), 400)
+    if seconds < 1:
+        return None, (jsonify(error="'seconds' must be >= 1"), 400)
+    if seconds > maximum:
+        return None, (jsonify(error=f"'seconds' must be <= {maximum}"), 400)
+    return seconds, None
+
+
+@app.route("/api/iq_capture", methods=["GET"])
+def api_iq_capture():
+    seconds, err = _parse_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()
+
+    buf = io.BytesIO()
+    np.save(buf, segment)
+    buf.seek(0)
+    fname = f"iq_{int(time.time())}_{seconds}s.npy"
+    resp = send_file(buf, mimetype="application/octet-stream",
+                     as_attachment=True, download_name=fname)
+    resp.headers["X-Sample-Rate-Hz"] = str(config.SAMPLE_RATE)
+    resp.headers["X-Center-Freq-Hz"] = str(int(state.lo_freq_hz))
+    resp.headers["X-Duration-S"] = str(seconds)
+    resp.headers["X-N-Samples"] = str(int(len(segment)))
+    return resp
+
+
+@app.route("/api/record_analyze", methods=["GET"])
+def api_record_analyze():
+    """Record N seconds of IQ and return a plain-text signal-diagnostic report file
+    (timing, frequency, channel hopping, amplitude) for the customer to send back."""
+    seconds, err = _parse_seconds(default=10, maximum=_RECORD_ANALYZE_MAX_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()
+
+    # Analysis is CPU-only on the captured copy; the lock is already released.
+    report = analysis.analyze_recording(segment)
+    report["capture"] = {
+        "seconds": seconds,
+        "sample_rate_hz": config.SAMPLE_RATE,
+        "center_freq_hz": int(state.lo_freq_hz),
+        "n_samples": int(len(segment)),
+    }
+    text = analysis.build_report(report)
+    buf = io.BytesIO(text.encode("utf-8"))
+    buf.seek(0)
+    fname = f"sat_record_{int(time.time())}_{seconds}s.txt"
+    resp = send_file(buf, mimetype="text/plain", as_attachment=True, download_name=fname)
+    resp.headers["X-Sample-Rate-Hz"] = str(config.SAMPLE_RATE)
+    resp.headers["X-Center-Freq-Hz"] = str(int(state.lo_freq_hz))
+    return resp
+
+
 # ===========================================================================
 # TX API routes
 # ===========================================================================

src/stream_web/config.py

@@ -118,3 +118,28 @@
 CHANNEL_SPACING = _fdc.CHANNEL_SPACING
 DEVICE_CHANNEL_SPACING = _fdc.DEVICE_CHANNEL_SPACING
 _fdc.configure(SAMPLE_RATE)
+
+
+# -- Packet symbol geometry (shared by analysis / processor / spectrogram) -------------
+# These derive a packet's symbol layout from the decoder's output, so the same expression
+# isn't re-inlined at every correct_symbol_edges() call site.
+
+def packet_symbol_grid(decode_info: dict) -> tuple[int, int, int]:
+    """Return ``(n_sym, slot, sym_len)`` for a decoded-packet / decode_info dict:
+    total symbols (preamble + header + PDU), samples per symbol slot, and symbol length."""
+    ver = decode_info.get("phy_ver", 1)
+    slot = slot_samples.get(ver, slot_samples[1])["slot"]
+    n_sym = PREAMBLE_LEN + NUM_HEADER_SYMS + (decode_info.get("num_pdu_symbols") or 0)
+    return n_sym, slot, samples_per_symbol
+
+
+def rotated_hop_sequence(channel_num, hop_seq_idx) -> list[int] | None:
+    """Channel-hop sequence rotated to start at ``channel_num`` (so index h is the channel
+    for hop h). Returns None if the hop parameters are missing or inconsistent."""
+    if channel_num is None or hop_seq_idx is None or hop_seq_idx >= len(HOPPING_SEQS):
+        return None
+    seq = HOPPING_SEQS[hop_seq_idx]
+    if channel_num not in seq:
+        return None
+    i = seq.index(channel_num)
+    return seq[i:] + seq[:i]

src/stream_web/processor.py

@@ -181,11 +181,9 @@ def _extract_last(n):
                 "gap_count": None, "gap_mean_ms": None, "gap_std_ms": None,
             }
             if pkt_start is not None:
-                slot = config.slot_samples.get(ver, config.slot_samples[1])["slot"]
-                n_sym = (config.PREAMBLE_LEN + config.NUM_HEADER_SYMS
-                         + (pkt.get("num_pdu_symbols") or 0))
+                n_sym, slot, sym_len = config.packet_symbol_grid(pkt)
                 edges = correct_symbol_edges(
-                    decode_chunk, pkt_start, 0, n_sym, 0, slot, config.samples_per_symbol,
+                    decode_chunk, pkt_start, 0, n_sym, 0, slot, sym_len,
                 )
                 if edges:
                     timing = edges_to_timing_stats(edges, config.SAMPLE_RATE)

src/stream_web/spectrogram.py

@@ -12,7 +12,11 @@
 from scipy.signal import spectrogram as scipy_spectrogram  # noqa: E402
 
 from . import config  # noqa: E402
-from .timing import correct_symbol_edges, measure_transition_us  # noqa: E402
+from .timing import (  # noqa: E402
+    correct_symbol_edges,
+    dominant_symbol_freq,
+    measure_transition_us,
+)
 
 # -- Pre-computed LUT and font ----------------------------------------------
 
@@ -388,12 +392,7 @@ def render_td_plot(
     ends = np.array([], dtype=int)
 
     if decode_info is not None and decode_info.get("start_sample") is not None:
-        sym_len = config.samples_per_symbol
-        phy_ver = decode_info.get("phy_ver", 1)
-        slot = config.slot_samples[phy_ver]["slot"]
-        n_sym = (config.PREAMBLE_LEN
-                 + config.NUM_HEADER_SYMS
-                 + (decode_info.get("num_pdu_symbols") or 0))
+        n_sym, slot, sym_len = config.packet_symbol_grid(decode_info)
         corrected = correct_symbol_edges(
             iq_segment, decode_info["start_sample"], 0, n_sym, 0, slot, sym_len,
         )
@@ -452,20 +451,10 @@ def render_td_plot(
     y_floor = max(DBFS_FLOOR, sig_peak_dbfs - 60)
     ax_td.set_ylim(y_floor, 0)
 
-    # FFT-based per-symbol tone frequency: blank the DC bin (bottom 2%) so
-    # IQ imbalance spurs don't eclipse the real FSK carrier
-    sym_freqs = []
-    for s, e in zip(starts, ends):
-        sym_iq = iq_segment[s:e]
-        spec = np.fft.fft(sym_iq)
-        psd = np.abs(spec) ** 2
-        freqs = np.fft.fftfreq(len(sym_iq), d=1.0 / config.SAMPLE_RATE)
-        dc_zone = int(len(psd) * 0.02)
-        if dc_zone > 0:
-            psd[:dc_zone] = 0
-            psd[-dc_zone:] = 0
-        pk = np.argmax(psd)
-        sym_freqs.append(freqs[pk])
+    # FFT-based per-symbol tone frequency (DC bins blanked so IQ-imbalance spurs don't
+    # eclipse the real FSK carrier).
+    sym_freqs = [dominant_symbol_freq(iq_segment[s:e], config.SAMPLE_RATE)
+                 for s, e in zip(starts, ends)]
 
     # F0 reference: use the decoder's measured value if available (most accurate),
     # otherwise fall back to the second detected symbol (first is the F63 preamble tone)

src/stream_web/timing.py

@@ -157,3 +157,41 @@ def edges_to_timing_stats(edges: list[tuple[int, int]], sr: int) -> dict:
         "gap_mean_ms": float(round(np.mean(gap_ms), 4)) if len(gap_ms) else None,
         "gap_std_ms": float(round(np.std(gap_ms), 4)) if len(gap_ms) else None,
     }
+
+
+def dominant_symbol_freq(iq_seg: np.ndarray, sr: int, blank_dc_frac: float = 0.02) -> float:
+    """Dominant baseband frequency (Hz) of a symbol-length slice via FFT.
+
+    The DC bins (bottom/top ``blank_dc_frac``) are zeroed first so IQ-imbalance spurs near
+    0 Hz don't eclipse the real FSK carrier. Returns 0.0 for an empty slice.
+    """
+    if not len(iq_seg):
+        return 0.0
+    psd = np.abs(np.fft.fft(iq_seg)) ** 2
+    freqs = np.fft.fftfreq(len(iq_seg), d=1.0 / sr)
+    dc = int(len(psd) * blank_dc_frac)
+    if dc > 0:
+        psd[:dc] = 0
+        psd[-dc:] = 0
+    return float(freqs[int(np.argmax(psd))])
+
+
+def symbol_amplitudes_dbfs(iq_seg: np.ndarray, edges: list[tuple[int, int]],
+                           full_scale: float = 1.0):
+    """Per-symbol RMS amplitude and per-gap noise-floor RMS, both in dBFS.
+
+    Returns ``(sym_dbfs, gap_dbfs)`` -- one amplitude per symbol body, one noise-floor
+    level per inter-symbol gap. Drives amplitude-dropoff and per-symbol SNR.
+    """
+    sym_dbfs, gap_dbfs = [], []
+    for i, (s, e) in enumerate(edges):
+        body = iq_seg[s:e]
+        if len(body):
+            sym_dbfs.append(20.0 * np.log10(np.sqrt(np.mean(np.abs(body) ** 2))
+                                            / full_scale + 1e-30))
+        if i < len(edges) - 1:
+            gap = iq_seg[e:edges[i + 1][0]]
+            if len(gap):
+                gap_dbfs.append(20.0 * np.log10(np.sqrt(np.mean(np.abs(gap) ** 2))
+                                                / full_scale + 1e-30))
+    return sym_dbfs, gap_dbfs