diff --git a/CLAUDE.md b/CLAUDE.md index 0b8d8ba..f6b13a5 100644 --- a/CLAUDE.md +++ b/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–30, default 10) 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, default 10) and return a plain-text diagnostic report file: for a representative decoded packet, per-symbol timing (duration/gap/drift), per-symbol frequency vs expected channel window, and per-symbol amplitude/SNR, plus chipset/synth-res, frequency offset, and RS corrections; analyzes decoded packets only; backs `hubblenetwork sat record` ### TX - `POST /api/tx/start` — start TX (`{"mode":"tone"}` or `{"mode":"packet","file":""}`) diff --git a/Dockerfile b/Dockerfile index e4c083f..45f5f79 100644 --- a/Dockerfile +++ b/Dockerfile @@ -73,18 +73,21 @@ COPY entrypoint.sh /app/ ARG USE_LOCAL_DECODER=0 COPY decoder-src* /tmp/decoder-src/ -# Install the python package -# GNU Radio from the Ubuntu PPA is compiled against NumPy 1.x; -# pip must not upgrade numpy beyond 1.x or gnuradio will fail to import. -# --ignore-installed is needed because some system distutils packages -# (blinker, etc.) can't be pip-uninstalled cleanly. +# Install the python package (pulls hubble-satnet-decoder from PyPI as a dependency). +# GNU Radio from the Ubuntu PPA is compiled against NumPy 1.x; pip must not upgrade numpy +# beyond 1.x or gnuradio fails to import. --ignore-installed is needed because some system +# distutils packages (blinker, etc.) can't be pip-uninstalled cleanly. RUN python3 -m pip install --upgrade pip setuptools wheel +# NOTE: a non-local build pulls hubble-satnet-decoder from PyPI. Until 1.2.0 (which adds +# analyze_packet) is published, the app imports will fail at startup on a non-local build -- +# use --build-arg USE_LOCAL_DECODER=1 in the meantime. +RUN python3 -m pip install --ignore-installed -e . "numpy>=1.26,<2" + +# For a local decoder build, override the PyPI decoder that `-e .` just pulled with the copied +# source (--no-deps, so only the decoder package is replaced) so the local version wins. RUN if [ "$USE_LOCAL_DECODER" = "1" ]; then \ - python3 -m pip install --ignore-installed /tmp/decoder-src "numpy>=1.26,<2"; \ - else \ - python3 -m pip install --ignore-installed "hubble-satnet-decoder>=1.1.1" "numpy>=1.26,<2"; \ + python3 -m pip install --ignore-installed --no-deps /tmp/decoder-src; \ fi -RUN python3 -m pip install --ignore-installed -e . "numpy>=1.26,<2" # Start the live spectrogram + decoder web server EXPOSE 8050 diff --git a/src/stream_web/analysis.py b/src/stream_web/analysis.py new file mode 100644 index 0000000..cbb9910 --- /dev/null +++ b/src/stream_web/analysis.py @@ -0,0 +1,260 @@ +"""Offline diagnostics for a recorded IQ segment (the /api/record_analyze endpoint). + +Decodes the packets in the recording via hubble_satnet_decoder, runs the decoder's per-packet +signal analysis on each, and renders a plain-text diagnostic report. + +Note: currently only packets that successfully decode are analyzed. +""" +import logging + +import numpy as np +from hubble_satnet_decoder import analyze_packet, decode_signal, packet_symbol_grid + +from . import config +from .timing import correct_symbol_edges + +log = logging.getLogger(__name__) + +# 1. Packet detection + +def _window_offsets(n: int, win: int) -> list[int]: + """Decoder-window start offsets covering the recording, overlapping by more than the + longest packet (~0.53 s) so every packet lands fully inside at least one window.""" + if n <= win: + return [0] + step = max(1, win - int(config.DECODE_INTERVAL_S * config.SAMPLE_RATE)) + offsets = list(range(0, n - win + 1, step)) + if offsets[-1] != n - win: + offsets.append(n - win) + return offsets + + +def _is_duplicate(pkt: dict, seen: dict[tuple, list[float]]) -> bool: + """True if *pkt* is the same packet as one already accepted -- same sequence number, + auth tag, and payload, within config.DEDUP_START_TOL_S. This collapses a packet caught in two + overlapping windows while keeping genuinely distinct packets (which differ in payload/ + auth or are further apart in time, even if a short seq_num happens to repeat). + + *seen* buckets accepted packets' abs_time_s by (seq_num, auth_tag, payload_val), so only + packets that could possibly match are compared -- O(1) on average instead of scanning every + packet accepted so far. If *pkt* is new, its time is recorded in the bucket.""" + key = (pkt.get("seq_num"), pkt.get("auth_tag"), pkt.get("payload_val")) + times = seen.setdefault(key, []) + if any(abs(pkt["abs_time_s"] - t) < config.DEDUP_START_TOL_S for t in times): + return True + times.append(pkt["abs_time_s"]) + return False + + +def decode_packets(iq: np.ndarray) -> list[dict]: + """Run the decoder over the recording and return each decoded packet once, in time order. + + The decoder works on a fixed-size window, so we slide overlapping windows across the + recording and keep the fully-decoded packets, deduped by _is_duplicate (a packet caught + in two overlapping windows is reported once). The decoded *attempt* supplies start_sample + (the preamble location); the *packet* supplies the decode-only fields (freq offset, PDU + RS corrections). One failed window is logged and skipped -- it never aborts the capture. + """ + sr, win = config.SAMPLE_RATE, config.DECODE_SAMPLES + accepted: list[dict] = [] + seen: dict[tuple, list[float]] = {} + for offset in _window_offsets(len(iq), win): + try: + packets, _detections, attempts = decode_signal(iq[offset:offset + win]) + except Exception: + log.warning("decode_signal failed on window at offset %d; skipping", offset, + exc_info=True) + continue + # The 'start sample' we need lives only on the decoded attempts. + starts = {(a.get("ntw_id"), a.get("seq_num")): a.get("start_sample") + for a in attempts + if a.get("decoded") and a.get("start_sample") is not None} + + for p in packets: + # Searching through attempted decodes with successful decodes to find + # the start sample for this packet (by ntw_id + seq_num) + start = starts.get((p.get("ntw_id"), p.get("seq_num"))) + if start is None: + continue + d = dict(p) + d["abs_start"] = offset + start + d["abs_time_s"] = d["abs_start"] / sr + if _is_duplicate(d, seen): # If duplicate, don't save + continue + n_sym, slot = packet_symbol_grid(d) + + # Find nominal start and end of packet, then correct the symbol edges + lo = max(0, d["abs_start"] - slot) + hi = min(len(iq), d["abs_start"] + n_sym * slot + slot) + local = correct_symbol_edges(iq[lo:hi], d["abs_start"] - lo, 0, n_sym, 0, slot, + config.samples_per_symbol) + d["edges"] = [(s + lo, e + lo) for s, e in local] # back to absolute indices + accepted.append(d) + accepted.sort(key=lambda d: d["abs_start"]) + return accepted + + +# 2. Capture-level summary + orchestration + +def _rs_corrections(packets: list[dict]) -> dict: + """Return RS corrections, which are data members of the packet dict.""" + def col(key): + return [float(p[key]) for p in packets + if isinstance(p.get(key), (int, float)) and not isinstance(p.get(key), bool)] + hdr, pdu = col("header_n_corr"), col("pdu_n_corr") + return {"header_mean": round(float(np.mean(hdr)), 2) if hdr else None, + "pdu_mean": round(float(np.mean(pdu)), 2) if pdu else None, + "pdu_max": int(max(pdu)) if pdu else None} + + +def _packet_info(pkt: dict) -> dict: + return {"abs_time_s": round(float(pkt["abs_time_s"]), 4), + "channel_num": pkt.get("channel_num"), "chipset": pkt.get("chipset")} + + +def _packet_line(pkt: dict, timing: dict, amplitude: dict) -> dict: + """Report packet info in one line.""" + return {**_packet_info(pkt), + "sym_mean_ms": timing["sym_mean_ms"], "gap_mean_ms": timing["gap_mean_ms"], + "total_drift_us": timing["total_drift_us"], + "amp_mean_dbfs": amplitude["mean_dbfs"], "snr_db": amplitude["snr_db"]} + + +def analyze_recording(iq: np.ndarray) -> dict: + """Analyze a recorded IQ segment: decode the packets, run the decoder's per-packet analysis + on each, pick the healthiest as representative, and summarise. Returns the report dict + consumed by build_report (the endpoint adds the 'capture' block).""" + + # Decode packets from IQ samples + packets = decode_packets(iq) + if not packets: + return {"summary": {"packets": 0}, "representative": None, "packets": []} + + # Invoke hubble-satnet-decoder's analysis + analyses = [analyze_packet(iq, p, p["edges"]) for p in packets] + + # Define the best packet based on snr_db + def _quality(i): + snr = analyses[i]["amplitude"]["snr_db"] + return (snr if snr is not None else float("-inf"), packets[i].get("num_pdu_symbols") or 0) + + rep_i = max(range(len(packets)), key=_quality) + rep, rep_an = packets[rep_i], analyses[rep_i] + + return { + "summary": {"packets": len(packets), "rs_corrections": _rs_corrections(packets), + "chipset": rep_an["chipset"], + "freq_delta_hz": round(float(rep["freq_delta_hz"]), 1) + if rep.get("freq_delta_hz") is not None else None}, + "representative": {"info": _packet_info(rep), "timing": rep_an["timing"], + "amplitude": rep_an["amplitude"], "channels": rep_an["channels"], + "hops": rep_an["hops"]}, + "packets": [_packet_line(packets[i], analyses[i]["timing"], analyses[i]["amplitude"]) + for i in range(len(packets))], + } + + +# 3. Report formatting + +def build_report(report: dict) -> str: + """Render the analysis dict into the plain-text diagnostic file the customer returns.""" + s = report["summary"] + lines = ["=== Hubble sat-record diagnostic ==="] + # Show capture statistics + cap = report.get("capture") + if cap: + lines.append(f"capture: {cap['seconds']} s @ {cap['sample_rate_hz']} Hz, " + f"center {cap['center_freq_hz'] / 1e6:.4f} MHz, {cap['n_samples']} samples") + lines.append(f"decoded packets: {s['packets']}") + + # Show representative packet + rep = report.get("representative") + if not rep: + lines.append("No packets decoded -- check the device is transmitting on the " + "expected channel.") + return "\n".join(lines) + "\n" + + # Show information about the representative packet + info = rep["info"] + lines += ["", "=== representative packet ===", + f" t={info['abs_time_s']}s channel {info['channel_num']} " + f"chipset {info['chipset']}"] + _timing_section(lines, rep["timing"]) + _channel_section(lines, rep["channels"], rep["hops"]) + _amplitude_section(lines, rep["amplitude"]) + _frequency_section(lines, s) + + # Show RS corrections and per-packet table + rs = s["rs_corrections"] + lines += ["", f"RS corrections: header_mean={rs['header_mean']} " + f"pdu_mean={rs['pdu_mean']} pdu_max={rs['pdu_max']}"] + _packet_table(lines, report["packets"]) + return "\n".join(lines) + "\n" + +# Write up timing section, channel section, amplitude section, frequency section for each symbol +def _timing_section(lines: list, t: dict) -> None: + lines += ["", "PER-SYMBOL TIMING (duration, gap; drift = slot midpoint vs. ideal grid)"] + if not t["per_symbol"]: + lines.append(" n/a (not enough symbols)") + return + for r in t["per_symbol"]: + gap = f", gap={r['gap_us']:.2f} us" if r["gap_us"] is not None else "" + if r["drift_us"] is not None: + drift = f", drift={r['drift_us']:+.2f} us, rate={r['rate_us_per_sym']:+.2f} us/sym" + else: + drift = ", drift=n/a" + lines.append(f" Symbol {r['idx']:>2}: duration={r['duration_us']:.2f} us{gap}{drift}") + lines.append(f" Overall drift: {t['total_drift_us']:+.2f} us " + f"({t['drift_us_per_sym']:+.2f} us/symbol over {len(t['per_symbol'])} symbols)") + + +def _channel_section(lines: list, ch: dict | None, hops: dict | None) -> None: + lines += ["", "PER-SYMBOL FREQUENCY / CHANNEL HOPPING"] + if hops: + lines.append(f" hop sequence idx {hops['hop_seq_idx']}, start channel " + f"{hops['start_channel']}, expected channels {hops['expected_channels']}") + if not ch: + lines.append(" n/a (channel/hop info unavailable for this packet)") + return + c = ch["calibration"] + lines += [f" calibration: F0={c['f0_hz']} Hz, FSK step={c['step_hz']} Hz, " + f"channel width={c['channel_width_hz']} Hz", + f" channel spacing: {c['spacing_hz']} Hz ({c['spacing_source']})"] + for r in ch["per_symbol"]: + mark = "ok" if r["in_window"] else f"OUT by {r['off_by_hz']} Hz" + lines.append(f" Symbol {r['idx']:>2}: ch {r['channel']:>2}, " + f"freq={r['freq_hz']:>10.1f} Hz [{mark}]") + lines.append(f" channel validation: {'PASS' if ch['all_valid'] else 'FAIL'} " + f"({ch['n_in_window']}/{len(ch['per_symbol'])} symbols in window)") + + +def _amplitude_section(lines: list, a: dict) -> None: + lines += ["", "PER-SYMBOL AMPLITUDE (RMS dBFS; SNR above the inter-symbol noise floor)"] + if not a["per_symbol"]: + lines.append(" n/a (not enough symbols)") + return + for r in a["per_symbol"]: + snr = f", snr={r['snr_db']:.2f} dB" if r["snr_db"] is not None else "" + lines.append(f" Symbol {r['idx']:>2}: amp={r['amp_dbfs']:.2f} dBFS{snr}") + lines.append(f" summary: mean {a['mean_dbfs']} dBFS, dropoff (p-p) {a['dropoff_db']} dB, " + f"noise floor {a['noise_floor_dbfs']} dBFS, SNR {a['snr_db']} dB") + + +def _frequency_section(lines: list, s: dict) -> None: + # Prints actual vs expected synth resolution and chipset info + ch = s["chipset"] + lines += ["", "FREQUENCY / CHIPSET", + f" offset from center: {s['freq_delta_hz']} Hz", + f" chipset (decoder): {ch['chipset']}", + f" synth-res: measured {ch['measured_synth_res']} Hz " + f"(chipset nominal {ch['nominal_synth_res']} Hz)"] + + +def _packet_table(lines: list, packets: list[dict]) -> None: + # Prints general packet info for that recording + lines += ["", "PER-PACKET (all packets)"] + for p in packets: + lines.append(f" t={p['abs_time_s']}s ch{p['channel_num']} {p['chipset']}: " + f"sym={p['sym_mean_ms']}ms gap={p['gap_mean_ms']}ms " + f"drift={p['total_drift_us']}us amp={p['amp_mean_dbfs']}dBFS " + f"snr={p['snr_db']}dB") diff --git a/src/stream_web/app.py b/src/stream_web/app.py index 50afc57..b650148 100644 --- a/src/stream_web/app.py +++ b/src/stream_web/app.py @@ -16,6 +16,7 @@ import logging import multiprocessing as mp import os +import tempfile import threading import time from collections import deque @@ -26,7 +27,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 +467,164 @@ def api_packets(): return Response(payload, mimetype="application/x-ndjson") +_CAPTURE_DEFAULT_SECONDS = 10 +_CAPTURE_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=_CAPTURE_DEFAULT_SECONDS, maximum=_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() + + # Spill the .npy to disk past 10 MB so a large capture isn't duplicated in RAM: the + # segment plus its serialized bytes would otherwise roughly double peak memory + # (~375 MB for a 30 s capture) and can OOM small containers. + buf = tempfile.SpooledTemporaryFile(max_size=10 * 1024 * 1024) + 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() + 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)) + resp.headers["X-Duration-S"] = str(seconds) + resp.headers["X-N-Samples"] = str(int(len(segment))) + return resp + + # =========================================================================== # TX API routes # =========================================================================== diff --git a/src/stream_web/config.py b/src/stream_web/config.py index a19510f..4a9ab13 100644 --- a/src/stream_web/config.py +++ b/src/stream_web/config.py @@ -100,6 +100,10 @@ DECODE_INTERVAL_S = 0.6 DECODE_SAMPLES = int(DECODE_WINDOW_S * SAMPLE_RATE) +# Two decodes of the same packet (same device id, auth tag, payload) within this window are +# treated as one -- shared by the live processor and the offline record-analyze sweep. +DEDUP_START_TOL_S = 1.0 + # -- Web server & app behaviour -------------------------------------------- FLASK_PORT = 8050 VERBOSE = False @@ -118,3 +122,4 @@ CHANNEL_SPACING = _fdc.CHANNEL_SPACING DEVICE_CHANNEL_SPACING = _fdc.DEVICE_CHANNEL_SPACING _fdc.configure(SAMPLE_RATE) + diff --git a/src/stream_web/processor.py b/src/stream_web/processor.py index e618421..05c8af2 100644 --- a/src/stream_web/processor.py +++ b/src/stream_web/processor.py @@ -19,11 +19,6 @@ from .spectrogram import render_spec_image, render_symbol_zoom_plot, render_td_plot from .timing import correct_symbol_edges, edges_to_timing_stats -# Two packets with the same device ID, auth tag, and payload that -# have preambles 1s within each other are considered the same packet -# and are de-duplicated. -_DEDUP_START_TOL_S = 1.0 - def processor_main(shm_name, buf_write_idx_val, rx_peak_frac_val, rx_overflows_val, rx_gain_dB_val, td_running_val, @@ -106,7 +101,7 @@ def _extract_last(n): start_t = t0 - (config.DECODE_WINDOW_S - p.get("time_s", 0.0)) key = (p.get("ntw_id"), p.get("auth_tag"), p.get("payload_val")) prev = recent_decodes.get(key) - if prev is not None and abs(start_t - prev) < _DEDUP_START_TOL_S: + if prev is not None and abs(start_t - prev) < config.DEDUP_START_TOL_S: continue # same packet within the tolerance -> drop the repeat recent_decodes[key] = start_t deduped.append(p)