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
89 lines
2.8 KiB
Python
89 lines
2.8 KiB
Python
import numpy as np
|
|
import pytest
|
|
|
|
from mcbladerf import capture, device
|
|
|
|
|
|
@pytest.mark.parametrize("window", ["hann", "blackman", "hamming", "none"])
|
|
def test_full_scale_tone_reads_0_dbfs(window):
|
|
fs, n = 1e6, 1024
|
|
bin_hz = fs / n
|
|
t = np.arange(n * 8) / fs
|
|
tone = np.exp(2j * np.pi * 37 * bin_hz * t).astype(np.complex64)
|
|
|
|
s = capture.compute_spectrum(tone, fs, 100e6, fft_size=n, window=window, averaging=8)
|
|
|
|
assert s.peak_power_db == pytest.approx(0.0, abs=0.01)
|
|
assert s.peak_frequency_hz == pytest.approx(100e6 + 37 * bin_hz)
|
|
|
|
|
|
def test_short_input_uses_requested_window():
|
|
x = np.ones(100, dtype=np.complex64)
|
|
s = capture.compute_spectrum(x, 1e6, 0.0, fft_size=1024, window="none")
|
|
assert s.bin_count == 100
|
|
assert s.peak_power_db == pytest.approx(0.0, abs=0.01)
|
|
|
|
|
|
def test_cs16_file_is_raw_q11(tmp_path):
|
|
raw = np.array([2048, -1024, 0, 512], dtype=np.int16)
|
|
path = tmp_path / "x.cs16"
|
|
raw.tofile(path)
|
|
|
|
x = capture.read_iq_file(path, "cs16")
|
|
|
|
np.testing.assert_allclose(x, [1.0 - 0.5j, 0.25j])
|
|
|
|
|
|
class _Range:
|
|
def __init__(self, lo, hi):
|
|
self.lo, self.hi = lo, hi
|
|
|
|
def minimum(self):
|
|
return self.lo
|
|
|
|
def maximum(self):
|
|
return self.hi
|
|
|
|
|
|
def test_range_span_covers_every_range():
|
|
# What SoapyBladeRF 0.4.1 reports for a bladeRF x40
|
|
ranges = [_Range(80e3, 10e6), _Range(10e6, 20e6), _Range(20e6, 40e6)]
|
|
assert device.range_span(ranges) == {"min_hz": 80e3, "max_hz": 40e6}
|
|
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
|
|
assert device.alloc_buffer(10, "CF32").dtype == np.complex64
|