rsp2k
9db76af626
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
125 lines
3.9 KiB
Python
125 lines
3.9 KiB
Python
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
|