Sat scan improvements

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")

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}")

src/stream_web/spectrogram.py

@@ -363,48 +363,75 @@ 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)
 
     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

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,
+    }