From 1290c729413ee59fff902854e6dea883afc84363 Mon Sep 17 00:00:00 2001 From: hunterhubble Date: Thu, 9 Apr 2026 20:14:49 +0000 Subject: [PATCH 1/2] stream_web: Added API call for IQ sample capture Helpful for: - Extra signal processing through numpy or matplotlib - Recording samples with pyhubblenetwork - Testing Satellite hardware with IQ samples from real devices - Determining the frequency of a continuous tone Signed-off-by: hunterhubble --- src/stream_web/app.py | 98 +++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 98 insertions(+) diff --git a/src/stream_web/app.py b/src/stream_web/app.py index 50afc57..849bc1e 100644 --- a/src/stream_web/app.py +++ b/src/stream_web/app.py @@ -466,6 +466,104 @@ def api_packets(): return Response(payload, mimetype="application/x-ndjson") +_IQ_CAPTURE_MAX_SECONDS = 60 + +# 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. +_iq_capture_lock = threading.Lock() + + +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 + + +@app.route("/api/iq_capture", methods=["GET"]) +def api_iq_capture(): + raw = flask_request.args.get("seconds") + if raw is None: + return jsonify(error="'seconds' query parameter is required"), 400 + try: + seconds = int(raw) + except (ValueError, TypeError): + return jsonify(error="'seconds' must be an integer"), 400 + + if seconds < 1: + return jsonify(error="'seconds' must be >= 1"), 400 + + if seconds > _IQ_CAPTURE_MAX_SECONDS: + return jsonify(error=f"'seconds' must be <= {_IQ_CAPTURE_MAX_SECONDS}"), 400 + + if not _iq_capture_lock.acquire(blocking=False): + return jsonify(error="Another IQ capture is already in progress"), 409 + try: + n_samples = seconds * config.SAMPLE_RATE + buf_size = config.IQ_BUFFER_SIZE + chunk_n = buf_size // 2 + + # Copy straight into one pre-allocated array -- no intermediate chunk + # list or concatenate -- so peak memory is a single copy of the + # capture rather than several. + 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: + return jsonify( + error="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 [chunk_start, +this_n) is read lock- + # free while the RX keeps writing. The RX only overwrites + # chunk_start after its write index laps a full buffer past it; if + # the copy stalled that long, the head is now < this_n ahead of + # chunk_start (it wrapped past it) and the samples are corrupt. + if _samples_advanced(chunk_start, int(state.buf_write_idx), + buf_size) < this_n: + return jsonify( + error="SDR overran the capture buffer (host too slow); " + "capture aborted"), 500 + + chunk_start = e + filled += this_n + + 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 + finally: + _iq_capture_lock.release() + + # =========================================================================== # TX API routes # =========================================================================== From 0226cd2bbe57aac440f25933954fef0af728acb6 Mon Sep 17 00:00:00 2001 From: hunterhubble Date: Thu, 2 Jul 2026 23:18:50 +0000 Subject: [PATCH 2/2] feat: Added analysis option Signed-off-by: hunterhubble --- CLAUDE.md | 2 + src/stream_web/analysis.py | 371 ++++++++++++++++++++++++++++++++++ src/stream_web/app.py | 207 ++++++++++++------- src/stream_web/config.py | 25 +++ src/stream_web/processor.py | 6 +- src/stream_web/spectrogram.py | 31 +-- src/stream_web/timing.py | 38 ++++ 7 files changed, 579 insertions(+), 101 deletions(-) create mode 100644 src/stream_web/analysis.py diff --git a/CLAUDE.md b/CLAUDE.md index 0b8d8ba..a9048f9 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–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":""}`) diff --git a/src/stream_web/analysis.py b/src/stream_web/analysis.py new file mode 100644 index 0000000..6040cc3 --- /dev/null +++ b/src/stream_web/analysis.py @@ -0,0 +1,371 @@ +"""Offline diagnostics for a recorded IQ segment (the /api/record_analyze endpoint) + + 1. Packet detection -- decode_packets(): run the decoder over the recording, keep the + decoded packets, anchor symbol edges on the preamble (start_sample). + 2. Per-symbol analyses -- one function per question, each run on a single packet: + analyze_timing() duration / gap / clock drift (cf. validate_symbol_gaps) + analyze_amplitude() per-symbol RMS dBFS + SNR + analyze_channels() dominant freq vs channel window (cf. validate_channel_v2) + analyze_chipset() measured synth-res vs known chipsets + expected_hops() the expected channel-hop schedule + 3. Orchestration -- analyze_recording(): decode, pick a representative packet, run + the analyses on it, and summarise the capture. + 4. Report formatting -- build_report(): render the plain-text file the customer returns. + +Note: Currently, this only analyzes packets that successfully decode +""" +import numpy as np +from hubble_satnet_decoder import decode_signal + +from . import config +from .timing import correct_symbol_edges, dominant_symbol_freq, symbol_amplitudes_dbfs + +# Ideal symbol grid, used only as the reference the drift measurement is taken against -- +_SYM_GRID_MS, _GAP_GRID_MS = 8.0, 0.8 + +# =========================================================================== +# 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(0.6 * config.SAMPLE_RATE)) + offsets = list(range(0, n - win + 1, step)) + if offsets[-1] != n - win: + offsets.append(n - win) + return offsets + + +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 (network id, sequence number) + -- 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). + """ + sr, win = config.SAMPLE_RATE, config.DECODE_SAMPLES + seen: dict = {} + for off in _window_offsets(len(iq), win): + try: + packets, _detections, attempts = decode_signal(iq[off:off + win]) + except Exception: # one bad window shouldn't abort the whole capture + continue + 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: + key = (p.get("ntw_id"), p.get("seq_num")) + if key in seen or starts.get(key) is None: + continue + d = dict(p) + d["abs_start"] = off + starts[key] + d["abs_time_s"] = d["abs_start"] / sr + n_sym, slot, sym_len = config.packet_symbol_grid(d) + d["edges"] = correct_symbol_edges(iq, d["abs_start"], 0, n_sym, 0, slot, sym_len) + seen[key] = d + return sorted(seen.values(), key=lambda d: d["abs_start"]) + + +# =========================================================================== +# 2. Per-symbol analyses (each answers one question about one packet) +# =========================================================================== + +def analyze_timing(pkt: dict) -> dict: + """Per-symbol duration, inter-symbol gap, and clock drift. + + Drift is the symbol's midpoint minus where an ideal 8ms-symbol/0.8ms-gap grid says it + should be; the symbol-0 offset is subtracted as a baseline so only accumulating clock + skew remains. Rate is the drift increment from the previous symbol. Mirrors + validate_symbol_gaps() in pluto_one_second_trace.py. + """ + sr, edges = config.SAMPLE_RATE, pkt["edges"] + if len(edges) < 2: + return {"per_symbol": [], "sym_mean_ms": None, "gap_mean_ms": None, + "drift_us_per_sym": None, "total_drift_us": None} + + # Create nominal template (8ms symbol + 0.8ms gap) and compare real packet to it + _n, slot, sym_len = config.packet_symbol_grid(pkt) + tmpl_mid = [edges[0][0] + i * slot + sym_len // 2 for i in range(len(edges))] + drift = [((s + e) // 2 - t) / sr * 1e6 for (s, e), t in zip(edges, tmpl_mid)] + drift = [d - drift[0] for d in drift] + + rows, durs, gaps = [], [], [] + for i, (s, e) in enumerate(edges): + dur = (e - s) / sr * 1e6 + gap = (edges[i + 1][0] - e) / sr * 1e6 if i < len(edges) - 1 else None + durs.append(dur) + if gap is not None: + gaps.append(gap) + rows.append({"idx": i, "duration_us": round(dur, 2), + "gap_us": round(gap, 2) if gap is not None else None, + "drift_us": round(drift[i], 2), + "rate_us_per_sym": round(drift[i] - drift[i - 1], 2) if i else 0.0}) + + return {"per_symbol": rows, + "sym_mean_ms": round(float(np.mean(durs)) / 1e3, 4), + "gap_mean_ms": round(float(np.mean(gaps)) / 1e3, 4) if gaps else None, + "total_drift_us": round(drift[-1], 2), + "drift_us_per_sym": round(drift[-1] / (len(edges) - 1), 2)} + + +def analyze_amplitude(iq: np.ndarray, pkt: dict) -> dict: + """Per-symbol RMS amplitude (dBFS) and SNR above the inter-symbol noise floor.""" + amps, gaps = symbol_amplitudes_dbfs(iq, pkt["edges"], config.ADC_FULL_SCALE) + if not amps: + return {"per_symbol": [], "mean_dbfs": None, "dropoff_db": None, + "noise_floor_dbfs": None, "snr_db": None} + + floor = float(np.median(gaps)) if gaps else None + rows = [{"idx": i, "amp_dbfs": round(float(a), 2), + "snr_db": round(float(a) - floor, 2) if floor is not None else None} + for i, a in enumerate(amps)] + return {"per_symbol": rows, + "mean_dbfs": round(float(np.mean(amps)), 2), + "dropoff_db": round(float(max(amps) - min(amps)), 2), + "noise_floor_dbfs": round(floor, 2) if floor is not None else None, + "snr_db": round(float(np.mean(amps)) - floor, 2) if floor is not None else None} + + +def _calibrate_spacing(slots, channel_num, base_freq, step): + """Inter-channel spacing (Hz), estimated as the median over PDU hops with enough symbols; + falls back to the configured CHANNEL_SPACING when the packet is too short to calibrate. + Returns (spacing, calibrated_from_pdu).""" + est = [] + for slot_num, ch, fs in slots: + if slot_num == 1 or ch == channel_num or len(fs) < 8: + continue + midrange = 0.5 * (min(fs) + max(fs)) + e = (midrange - 31.5 * step - base_freq) / (ch - channel_num) + if 0.7 * config.CHANNEL_SPACING < abs(e) < 1.3 * config.CHANNEL_SPACING: + est.append(e) + if est: + return float(np.median(est)), True + return float(config.CHANNEL_SPACING), False + + +def analyze_channels(iq: np.ndarray, pkt: dict) -> dict | None: + """Per-symbol dominant frequency vs the expected channel window. + + Calibrates the intra-channel FSK step from the two extreme preamble tones (symbol 0 = + value-63, symbol 1 = value-0), calibrates inter-channel spacing from the PDU hops, then + window-checks every symbol against its channel. Mirrors validate_channel_v2() in + pluto_one_second_trace.py. Returns None if hop parameters are unavailable. + """ + edges = pkt["edges"] + rotated = config.rotated_hop_sequence(pkt.get("channel_num"), pkt.get("hop_seq_idx")) + if rotated is None or len(edges) < 2: + return None + + sr, sps, hop = config.SAMPLE_RATE, config.samples_per_symbol, config.NUM_SYM_PER_HOP + ch = rotated[0] # == channel_num, by construction + freqs = [dominant_symbol_freq(iq[s:s + sps], sr) for s, _ in edges] + + f63, f0 = freqs[0], freqs[1] + step = abs(f63 - f0) / 63.0 + width = step * 64.0 + base = min(f63, f0) + + # Group symbols into hops of NUM_SYM_PER_HOP; hop h sits on channel rotated[h]. + slots = [(h + 1, rotated[h % len(rotated)], freqs[b:b + hop]) + for h, b in enumerate(range(0, len(freqs), hop))] + spacing, from_pdu = _calibrate_spacing(slots, ch, base, step) + expected = {c: base + (c - ch) * spacing for c in rotated} + + rows, gi = [], 0 + for _slot_num, c, fs in slots: + lo, hi = expected[c], expected[c] + width + for fr in fs: + ok = lo <= fr < hi + rows.append({"idx": gi, "channel": int(c), "freq_hz": round(fr, 1), + "in_window": ok, + "off_by_hz": 0.0 if ok else round(min(abs(fr - lo), abs(fr - hi)), 1)}) + gi += 1 + + return {"calibration": {"preamble_f63_hz": round(f63, 1), "preamble_f0_hz": round(f0, 1), + "step_hz": round(step, 1), "channel_width_hz": round(width, 1), + "spacing_hz": round(spacing, 1), "spacing_from_pdu": from_pdu, + "rotated_channels": [int(c) for c in rotated]}, + "per_symbol": rows, + "n_in_window": sum(r["in_window"] for r in rows), + "all_valid": all(r["in_window"] for r in rows)} + + +def analyze_chipset(pkt: dict) -> dict: + """Chipset and synthesizer resolution as reported by the decoder: the measured value and + the chipset's nominal table value, printed as-is (no matching / threshold).""" + meas = pkt.get("measured_synth_res") + name = pkt.get("chipset") + return {"chipset": name, + "measured_synth_res": round(float(meas), 2) if meas is not None else None, + "nominal_synth_res": config.SYNTH_RES.get(name)} + + +def expected_hops(pkt: dict) -> dict | None: + """Expected channel-hop schedule (start channel + rotated sequence). None if unavailable.""" + rotated = config.rotated_hop_sequence(pkt.get("channel_num"), pkt.get("hop_seq_idx")) + if rotated is None: + return None + n_sym, _, _ = config.packet_symbol_grid(pkt) + n_hops = -(-n_sym // config.NUM_SYM_PER_HOP) # ceil division + return {"hop_seq_idx": int(pkt["hop_seq_idx"]), "start_channel": int(rotated[0]), + "expected_channels": [int(rotated[h % len(rotated)]) for h in range(n_hops)]} + + +# =========================================================================== +# 3. Capture-level summary +# =========================================================================== + +def _rs_corrections(packets: list[dict]) -> dict: + """Reed-Solomon correction effort across packets (printed as-is; higher = less margin).""" + 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} + + +# =========================================================================== +# 4. Orchestration +# =========================================================================== + +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(iq: np.ndarray, pkt: dict) -> dict: + """One-line-per-packet summary for the all-packets table.""" + t, a = analyze_timing(pkt), analyze_amplitude(iq, pkt) + return {**_packet_info(pkt), "sym_mean_ms": t["sym_mean_ms"], "gap_mean_ms": t["gap_mean_ms"], + "total_drift_us": t["total_drift_us"], "amp_mean_dbfs": a["mean_dbfs"], + "snr_db": a["snr_db"]} + + +def analyze_recording(iq: np.ndarray) -> dict: + """Analyze a recorded IQ segment: decode the packets, then run the per-symbol timing, + amplitude, frequency/channel, and chipset analyses on the representative packet. Returns + the report dict consumed by build_report (the endpoint adds the 'capture' block).""" + packets = decode_packets(iq) + if not packets: + return {"summary": {"packets": 0}, "representative": None, "packets": []} + + rep = max(packets, key=lambda p: p.get("num_pdu_symbols") or 0) # fullest = richest tables + return { + "summary": {"packets": len(packets), "rs_corrections": _rs_corrections(packets), + "chipset": analyze_chipset(rep), + "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": analyze_timing(rep), + "amplitude": analyze_amplitude(iq, rep), + "channels": analyze_channels(iq, rep), "hops": expected_hops(rep)}, + "packets": [_packet_line(iq, p) for p in packets], + } + + +# =========================================================================== +# 5. 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 ==="] + 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']}") + + 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" + + 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) + + 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" + + +def _timing_section(lines: list, t: dict) -> None: + lines += ["", f"PER-SYMBOL TIMING (duration, gap; drift measured vs an ideal " + f"{_SYM_GRID_MS * 1e3:.0f} us symbol / {_GAP_GRID_MS * 1e3:.0f} us gap 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 "" + lines.append(f" Symbol {r['idx']:>2}: duration={r['duration_us']:.2f} us{gap}, " + f"drift={r['drift_us']:+.2f} us, rate={r['rate_us_per_sym']:+.2f} us/sym") + 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: preamble sym63={c['preamble_f63_hz']} Hz " + f"sym0={c['preamble_f0_hz']} Hz, FSK step={c['step_hz']} Hz, " + f"channel width={c['channel_width_hz']} Hz", + f" channel spacing: {c['spacing_hz']} Hz " + f"({'calibrated from PDU hops' if c['spacing_from_pdu'] else 'default'})"] + 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: + 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: + 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 849bc1e..343262d 100644 --- a/src/stream_web/app.py +++ b/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 @@ -467,11 +467,21 @@ def api_packets(): _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. -_iq_capture_lock = threading.Lock() +# 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: @@ -481,87 +491,132 @@ def _samples_advanced(start: int, end: int, buf_size: int) -> int: return buf_size - start + end -@app.route("/api/iq_capture", methods=["GET"]) -def api_iq_capture(): +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: - return jsonify(error="'seconds' query parameter is required"), 400 + 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 jsonify(error="'seconds' must be an integer"), 400 - + return None, (jsonify(error="'seconds' must be an integer"), 400) if seconds < 1: - return jsonify(error="'seconds' must be >= 1"), 400 + 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 - if seconds > _IQ_CAPTURE_MAX_SECONDS: - return jsonify(error=f"'seconds' must be <= {_IQ_CAPTURE_MAX_SECONDS}"), 400 - if not _iq_capture_lock.acquire(blocking=False): - return jsonify(error="Another IQ capture is already in progress"), 409 +@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: - n_samples = seconds * config.SAMPLE_RATE - buf_size = config.IQ_BUFFER_SIZE - chunk_n = buf_size // 2 - - # Copy straight into one pre-allocated array -- no intermediate chunk - # list or concatenate -- so peak memory is a single copy of the - # capture rather than several. - 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: - return jsonify( - error="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 [chunk_start, +this_n) is read lock- - # free while the RX keeps writing. The RX only overwrites - # chunk_start after its write index laps a full buffer past it; if - # the copy stalled that long, the head is now < this_n ahead of - # chunk_start (it wrapped past it) and the samples are corrupt. - if _samples_advanced(chunk_start, int(state.buf_write_idx), - buf_size) < this_n: - return jsonify( - error="SDR overran the capture buffer (host too slow); " - "capture aborted"), 500 - - chunk_start = e - filled += this_n - - 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 + segment = _capture_iq(seconds * config.SAMPLE_RATE) + except CaptureError as ce: + return jsonify(error=str(ce)), ce.status finally: - _iq_capture_lock.release() + _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 # =========================================================================== diff --git a/src/stream_web/config.py b/src/stream_web/config.py index a19510f..9d03ac6 100644 --- a/src/stream_web/config.py +++ b/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] diff --git a/src/stream_web/processor.py b/src/stream_web/processor.py index e618421..bc60e14 100644 --- a/src/stream_web/processor.py +++ b/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) diff --git a/src/stream_web/spectrogram.py b/src/stream_web/spectrogram.py index f85f074..973c31c 100644 --- a/src/stream_web/spectrogram.py +++ b/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) diff --git a/src/stream_web/timing.py b/src/stream_web/timing.py index 051f925..7f3a5cd 100644 --- a/src/stream_web/timing.py +++ b/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