diff --git a/src/stream_web/app.py b/src/stream_web/app.py index ba3aaa2..50afc57 100644 --- a/src/stream_web/app.py +++ b/src/stream_web/app.py @@ -427,7 +427,8 @@ def api_packets(): """Poll-and-drain: return all decodes since last call as JSONL, then clear. Each line is a JSON object with: device_id, seq_num, device_type, - timestamp, rssi_dB, channel_num, freq_offset_hz. + timestamp, rssi_dB, channel_num, freq_offset_hz, pdu_n_corr, header_n_corr + and symbol data(sym_count, sym_mean_ms, sym_std_ms, gap_count, gap_mean_ms, gap_std_ms). """ with state.lock: entries = list(state.packet_feed) @@ -452,6 +453,14 @@ def api_packets(): "channel_num": e.get("channel_num"), "freq_offset_hz": e.get("freq_delta_hz"), "payload_b64": payload_b64, + "pdu_n_corr": e.get("pdu_n_corr"), + "header_n_corr": e.get("header_n_corr"), + "sym_count": e.get("sym_count"), + "sym_mean_ms": e.get("sym_mean_ms"), + "sym_std_ms": e.get("sym_std_ms"), + "gap_count": e.get("gap_count"), + "gap_mean_ms": e.get("gap_mean_ms"), + "gap_std_ms": e.get("gap_std_ms"), })) payload = "\n".join(lines) + ("\n" if lines else "") return Response(payload, mimetype="application/x-ndjson") diff --git a/src/stream_web/processor.py b/src/stream_web/processor.py index c73861f..1f76ebf 100644 --- a/src/stream_web/processor.py +++ b/src/stream_web/processor.py @@ -17,6 +17,7 @@ from . import config from .spectrogram import render_spec_image, render_symbol_zoom_plot, render_td_plot +from .timing import correct_symbol_edges, edges_to_timing_stats def processor_main(shm_name, buf_write_idx_val, rx_peak_frac_val, @@ -150,6 +151,22 @@ def _extract_last(n): for pkt in packets: ver = pkt["phy_ver"] ntw_hex = f"0x{pkt['ntw_id']:09X}" if ver == -1 else f"0x{pkt['ntw_id']:08X}" + + pkt_start = pkt.get("start_sample") + timing: dict = { + "sym_count": None, "sym_mean_ms": None, "sym_std_ms": None, + "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)) + edges = correct_symbol_edges( + decode_chunk, pkt_start, 0, n_sym, 0, slot, config.samples_per_symbol, + ) + if edges: + timing = edges_to_timing_stats(edges, config.SAMPLE_RATE) + decode_entries.append({ "timestamp": ts, "unix_ts": unix_ts, @@ -167,6 +184,7 @@ def _extract_last(n): "header_n_corr": pkt.get("header_n_corr"), "pdu_n_corr": pkt.get("pdu_n_corr"), "num_pdu_symbols": pkt.get("num_pdu_symbols"), + **timing, }) stats = { @@ -238,7 +256,7 @@ def _extract_last(n): if not decode_info.get("energy_dB"): decode_info["energy_dB"] = td_hit.get("total_energy_dB") try: - td_img = render_td_plot(td_seg, decode_info=decode_info) + td_img, td_stats = render_td_plot(td_seg, decode_info=decode_info) status = f"t={td_hit['time_s']:.3f}s" if decode_info["decoded"]: seq = decode_info.get("seq_num") @@ -250,6 +268,7 @@ def _extract_last(n): k: v for k, v in decode_info.items() if isinstance(v, (str, int, float, bool, list, type(None))) } + td_decode_info_out.update(td_stats) td_iq_seg_out = td_seg.copy() except Exception as e: print(f"[TD] Plot error: {e}") diff --git a/src/stream_web/spectrogram.py b/src/stream_web/spectrogram.py index d94a84a..f85f074 100644 --- a/src/stream_web/spectrogram.py +++ b/src/stream_web/spectrogram.py @@ -363,8 +363,17 @@ def render_symbol_zoom_plot( return buf.read() -def render_td_plot(iq_segment: np.ndarray, decode_info: dict | None = None) -> bytes: - """Render a time-domain magnitude plot + spectrogram with annotations.""" +def render_td_plot( + iq_segment: np.ndarray, decode_info: dict | None = None +) -> tuple[bytes, dict]: + """Render a time-domain magnitude plot + spectrogram with annotations. + + Returns ``(image_bytes, stats)`` where *stats* contains symbol and gap + duration statistics derived from detected symbol edges (protocol-corrected + when ``decode_info.start_sample`` is available; otherwise envelope-threshold + based): ``sym_count``, ``sym_mean_ms``, ``sym_std_ms``, + ``gap_count``, ``gap_mean_ms``, ``gap_std_ms``. + """ n = len(iq_segment) t_ms = np.arange(n) / config.SAMPLE_RATE * 1e3 mag = np.abs(iq_segment) @@ -372,39 +381,57 @@ def render_td_plot(iq_segment: np.ndarray, decode_info: dict | None = None) -> b mag_dbfs = 20.0 * np.log10(np.clip(mag, 1e-12, None) / config.ADC_FULL_SCALE) DBFS_FLOOR = -80.0 - # Envelope-based symbol edge detection: smooth with a 0.3ms boxcar to merge - # intra-symbol ripple without smearing the 800us inter-symbol gaps - win_samples = max(1, int(0.3e-3 * config.SAMPLE_RATE)) - envelope = np.convolve(mag, np.ones(win_samples) / win_samples, mode="same") - - # Threshold at 40% of the noise-to-peak dynamic range so weak packets still - # register without false triggers from noise - noise_floor = np.percentile(envelope, 10) - signal_peak = np.percentile(envelope, 95) - thresh = noise_floor + 0.4 * (signal_peak - noise_floor) - above = envelope > thresh - - padded = np.concatenate([[False], above, [False]]) - edges = np.diff(padded.astype(np.int8)) - starts = np.where(edges == 1)[0] - ends = np.where(edges == -1)[0] - - # Drop glitches shorter than 2ms — real FSK symbols are 8ms - min_sym = int(2e-3 * config.SAMPLE_RATE) - mask = (ends - starts) >= min_sym - starts, ends = starts[mask], ends[mask] - - # Merge segments separated by less than 0.1ms; the boxcar smoothing can - # split a single symbol into multiple runs if there's a deep mid-symbol dip - min_gap = int(0.1e-3 * config.SAMPLE_RATE) - m_starts, m_ends = [], [] - for s, e in zip(starts, ends): - if m_ends and (s - m_ends[-1]) < min_gap: - m_ends[-1] = e - else: - m_starts.append(s) - m_ends.append(e) - starts, ends = np.array(m_starts), np.array(m_ends) + # Symbol edge detection: use protocol-aware correction when decode_info + # has start_sample (same method as the zoomed symbol plot), otherwise fall + # back to envelope threshold detection. + starts = np.array([], dtype=int) + 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)) + corrected = correct_symbol_edges( + iq_segment, decode_info["start_sample"], 0, n_sym, 0, slot, sym_len, + ) + if corrected: + starts = np.array([s for s, _e in corrected]) + ends = np.array([e for _s, e in corrected]) + + if len(starts) == 0: + # Fallback: envelope threshold detection + # Smooth with a 0.3ms boxcar to merge intra-symbol ripple without + # smearing the 800us inter-symbol gaps + win_samples = max(1, int(0.3e-3 * config.SAMPLE_RATE)) + envelope = np.convolve(mag, np.ones(win_samples) / win_samples, mode="same") + # Threshold at 40% of the noise-to-peak dynamic range so weak packets + # still register without false triggers from noise + noise_floor = np.percentile(envelope, 10) + signal_peak = np.percentile(envelope, 95) + thresh = noise_floor + 0.4 * (signal_peak - noise_floor) + above = envelope > thresh + padded = np.concatenate([[False], above, [False]]) + diff = np.diff(padded.astype(np.int8)) + starts = np.where(diff == 1)[0] + ends = np.where(diff == -1)[0] + # Drop glitches shorter than 2ms — real FSK symbols are 8ms + min_sym = int(2e-3 * config.SAMPLE_RATE) + mask = (ends - starts) >= min_sym + starts, ends = starts[mask], ends[mask] + # Merge segments separated by less than 0.1ms; the boxcar smoothing can + # split a single symbol into multiple runs if there's a deep mid-symbol dip + min_gap = int(0.1e-3 * config.SAMPLE_RATE) + m_starts, m_ends = [], [] + for s, e in zip(starts, ends): + if m_ends and (s - m_ends[-1]) < min_gap: + m_ends[-1] = e + else: + m_starts.append(s) + m_ends.append(e) + starts, ends = np.array(m_starts), np.array(m_ends) sym_dur_ms = (ends - starts) / config.SAMPLE_RATE * 1e3 gap_starts = ends[:-1] @@ -477,7 +504,7 @@ def render_td_plot(iq_segment: np.ndarray, decode_info: dict | None = None) -> b ax_td.axvspan(t0, t1, alpha=0.12, color="#f87171") # Stats box: symbol/gap means + std, plus cumulative drift vs expected 8.8ms slot. - # Drift = (actual first-to-last span) - (n_periods \u00d7 8.8ms); positive = TX clock fast + # Drift = (actual first-to-last span) - (n_periods × 8.8ms); positive = TX clock fast EXPECTED_PERIOD_MS = 8.8 lines = [] if len(sym_dur_ms): @@ -608,4 +635,13 @@ def render_td_plot(iq_segment: np.ndarray, decode_info: dict | None = None) -> b buf = io.BytesIO() canvas.print_png(buf) buf.seek(0) - return buf.read() + + stats: dict = { + "sym_count": int(len(sym_dur_ms)), + "sym_mean_ms": float(round(np.mean(sym_dur_ms), 4)) if len(sym_dur_ms) else None, + "sym_std_ms": float(round(np.std(sym_dur_ms), 4)) if len(sym_dur_ms) else None, + "gap_count": int(len(gap_dur_ms)), + "gap_mean_ms": float(round(np.mean(gap_dur_ms), 4)) if len(gap_dur_ms) else None, + "gap_std_ms": float(round(np.std(gap_dur_ms), 4)) if len(gap_dur_ms) else None, + } + return buf.read(), stats diff --git a/src/stream_web/timing.py b/src/stream_web/timing.py index b463b49..051f925 100644 --- a/src/stream_web/timing.py +++ b/src/stream_web/timing.py @@ -132,3 +132,28 @@ def measure_transition_us( return None i10 = i90 + int(hits_10[0]) return abs(i10 - i90) / sr * 1e6 + + +def edges_to_timing_stats(edges: list[tuple[int, int]], sr: int) -> dict: + """Convert corrected (start, end) sample pairs into timing stats. + + Returns dict with sym_count, sym_mean_ms, sym_std_ms, + gap_count, gap_mean_ms, gap_std_ms. + """ + if not edges: + return { + "sym_count": 0, "sym_mean_ms": None, "sym_std_ms": None, + "gap_count": 0, "gap_mean_ms": None, "gap_std_ms": None, + } + s_arr = np.array([s for s, _ in edges]) + e_arr = np.array([e for _, e in edges]) + sym_ms = (e_arr - s_arr) / sr * 1e3 + gap_ms = (s_arr[1:] - e_arr[:-1]) / sr * 1e3 if len(s_arr) > 1 else np.array([]) + return { + "sym_count": int(len(sym_ms)), + "sym_mean_ms": float(round(np.mean(sym_ms), 4)), + "sym_std_ms": float(round(np.std(sym_ms), 4)), + "gap_count": int(len(gap_ms)), + "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, + }