Measure dropped samples from stream timestamps instead of hiding them

Overflows were counted but the samples the driver dropped were simply missing, so a 5 s capture with 300 overflows still read as a clean 5.000 s file. Each read's hardware timestamp is now compared with where the previous read ended; gaps are sized exactly, zero-filled by default so the file timeline stays true, and reported as completed_with_gaps with a warning, samples_dropped and a gap list. This also catches drops the plugin never flags as overflows (libusb transfer timeouts). Timestamps are differenced before rounding: rounding each absolute timestamp to a sample index flips by +/-1 near half-sample boundaries and produced phantom 1-sample gaps. - stream buffers 128 x 64K samples (~280 ms at 30 Msps, was ~4 ms) - reject out-of-range gain, frequency, rate and bandwidth instead of letting the plugin clamp silently; default bandwidth clamps to range - spectrum returns a max-hold decimated summary (256 points) plus noise_floor_db; full arrays were 139 KB at fft_size 4096 - capture_iq background=True with stream_status polling, so stop_capture is reachable; device-changing tools refuse while it runs - OSError returned in the error shape; status drops "unknown" fields - README: gap accounting, keep captures off tmpfs; captures/ ignored
2026-10-25 11:57:25 -06:00
parent 067067e1e7
commit 9db76af626
7 changed files with 642 additions and 178 deletions
@@ -34,17 +34,6 @@ def test_cs16_file_is_raw_q11(tmp_path):
    np.testing.assert_allclose(x, [1.0 - 0.5j, 0.25j])


-def test_components_and_clip_level_agree_across_formats():
-    q11 = np.array([2047, -100, 50, -2048], dtype=np.int16)
-    cf32 = (q11[0::2] + 1j * q11[1::2]).astype(np.complex64) / device.CS16_FULL_SCALE
-
-    a = capture._components(q11, "CS16")
-    b = capture._components(cf32, "CF32")
-
-    np.testing.assert_allclose(a, b)
-    assert np.count_nonzero(a.max(axis=1) >= capture._CLIP_LEVEL) == 2
-
-
 class _Range:
    def __init__(self, lo, hi):
        self.lo, self.hi = lo, hi
@@ -63,6 +52,36 @@ def test_range_span_covers_every_range():
    assert device.range_span([]) is None


+def test_chunk_stats_match_components():
+    q11 = np.array([2047, -100, 50, -2048, 10, 10], dtype=np.int16)
+    cf32 = (q11[0::2] + 1j * q11[1::2]).astype(np.complex64) / device.CS16_FULL_SCALE
+
+    for chunk, fmt in ((q11, "CS16"), (cf32, "CF32")):
+        peak, clipped = capture._chunk_stats(chunk, fmt)
+        assert peak == pytest.approx(1.0)
+        assert clipped == 2
+
+
+def test_decimate_max_hold_keeps_narrow_peak():
+    f = list(range(4096))
+    p = [-80.0] * 4096
+    p[1234] = -20.0
+
+    df, dp = capture.decimate_max_hold(f, p, 256)
+
+    assert len(dp) == 256
+    assert max(dp) == -20.0
+    assert capture.decimate_max_hold(f[:100], p[:100], 256) == (f[:100], p[:100])
+
+
+def test_check_range_rejects_instead_of_clamping():
+    device.check_range("rxvga1 gain", 30, _Range(5, 30), "dB")
+    with pytest.raises(RuntimeError, match="outside the supported range 5 to 30 dB"):
+        device.check_range("rxvga1 gain", 99, _Range(5, 30), "dB")
+    with pytest.raises(RuntimeError, match="80000 to 4e\\+07"):
+        device.check_range("sample rate", 61.44e6, [_Range(80e3, 10e6), _Range(20e6, 40e6)], "Hz")
+
+
 def test_alloc_buffer_matches_stream_format():
    assert device.alloc_buffer(10, "CS16").dtype == np.int16
    assert device.alloc_buffer(10, "CS16").size == 20
@@ -0,0 +1,124 @@
+import numpy as np
+import pytest
+
+from mcbladerf import capture, device
+from mcbladerf.device import ReadResult
+
+RATE = 1e6  # one sample per microsecond keeps timestamps readable
+
+
+@pytest.fixture
+def fake_stream(monkeypatch):
+    """Script readStream results; data reads fill the buffer with 1, 2, 3, ..."""
+    monkeypatch.setattr(capture, "_SETTLE_SAMPLES", 0)
+    for name in ("setup_rx_stream", "activate_rx_stream", "close_rx_stream"):
+        monkeypatch.setattr(device, name, lambda *a, **k: None)
+
+    def load(script):
+        it = iter(script)
+        value = 0
+
+        def read_stream(buff, num_samples, timeout_us=0):
+            nonlocal value
+            r = next(it)
+            if r.count:
+                value += 1
+                buff[: 2 * r.count] = value
+            return r
+
+        monkeypatch.setattr(device, "read_stream", read_stream)
+
+    return load
+
+
+def _us(n):
+    return int(n * 1000)
+
+
+def _run(total, fill_gaps):
+    stats = capture._StreamStats()
+    out = np.concatenate(
+        [c.copy() for c in capture._rx_chunks(total, RATE, "CS16", stats, fill_gaps)]
+    )
+    return out.reshape(-1, 2)[:, 0], stats
+
+
+SCRIPT = [
+    ReadResult(1000, time_ns=_us(0)),
+    ReadResult(1000, time_ns=_us(1000)),
+    ReadResult(0, overflow=True),
+    ReadResult(1000, time_ns=_us(3000)),  # 1000 samples dropped before this
+    ReadResult(1000, time_ns=_us(4000)),
+]
+
+
+def test_gap_is_zero_filled_at_exact_position(fake_stream):
+    fake_stream(SCRIPT)
+    i, stats = _run(5000, fill_gaps=True)
+
+    assert len(i) == 5000
+    assert (i[2000:3000] == 0).all()
+    assert (i[:1000] == 1).all() and (i[3000:4000] == 3).all()
+    assert stats.overflows == 1
+    assert stats.dropped == 1000
+    assert [(g.offset_samples, g.length_samples) for g in stats.gaps] == [(2000, 1000)]
+
+
+def test_gap_reported_without_fill(fake_stream):
+    fake_stream(SCRIPT + [ReadResult(1000, time_ns=_us(5000))])
+    i, stats = _run(5000, fill_gaps=False)
+
+    assert len(i) == 5000
+    assert not (i == 0).any()
+    assert stats.dropped == 1000
+    assert stats.gaps[0].offset_samples == 2000
+
+
+def test_zero_fill_never_exceeds_requested_length(fake_stream):
+    fake_stream([ReadResult(1000, time_ns=_us(0)), ReadResult(1000, time_ns=_us(50000))])
+    i, stats = _run(3000, fill_gaps=True)
+
+    assert len(i) == 3000
+    assert stats.dropped == 49000
+
+
+def test_overflow_without_timestamp_is_flagged(fake_stream):
+    fake_stream([ReadResult(1000), ReadResult(0, overflow=True), ReadResult(1000)])
+    _, stats = _run(2000, fill_gaps=True)
+
+    assert stats.overflows == 1
+    assert stats.unmeasured_overflows == 1
+    assert stats.dropped == 0
+
+
+def test_timestamp_rounding_is_not_a_gap(fake_stream):
+    # 30 Msps: 4096 samples is 136533.33 ns, so timestamps alternate rounding
+    rate = 30e6
+    fake_stream([ReadResult(4096, time_ns=round(k * 4096 * 1e9 / rate)) for k in range(50)])
+    stats = capture._StreamStats()
+    for _ in capture._rx_chunks(50 * 4096, rate, "CS16", stats):
+        pass
+    assert stats.gap_count == 0
+
+
+def test_real_timestamps_near_half_sample_are_not_gaps(fake_stream):
+    # Logged from the bladeRF at 30 Msps. Each absolute timestamp sits near a
+    # half-sample boundary, so rounding them one by one gave +1, 0, -1 and a
+    # phantom 1-sample gap every third read.
+    rate = 30e6
+    t0, step = 102678001063517, [2184534, 2184533, 2184533]
+    times = [t0]
+    for k in range(11):
+        times.append(times[-1] + step[k % 3])
+    fake_stream([ReadResult(65536, time_ns=t) for t in times])
+    stats = capture._StreamStats()
+    for _ in capture._rx_chunks(12 * 65536, rate, "CS16", stats):
+        pass
+    assert stats.gap_count == 0
+
+
+def test_warning_text_states_dropped_time():
+    r = capture.CaptureResult("f", 5000, 1.0, RATE, 1e9, "cs16", 20000,
+                              overflows=1, samples_dropped=1000)
+    assert "1000 samples (1.0 ms)" in r.warning
+    assert "zeros" in r.warning