Fix CS16 capture, CLI tools while connected, sample-rate range, stdout banner
- CS16 streams read into an int16 buffer; files hold raw SC16 Q11 (full scale 2048) instead of all zeros - bladeRF-cli backed tools release the USB claim while they run and restore RX/TX settings afterwards; loopback uses the plugin setting - calibrate_dc runs "calibrate lms" + "calibrate dc rxtx"; a bare "cal dc" only printed help and reported success - report the span of all SoapySDR ranges (rate was capped at 10 Msps) - parse "Firmware version" from bladeRF-cli 1.10 - startup banner goes to stderr, off the stdio JSON-RPC channel - shared RX chunk reader: settle discard, overflow and stall handling, stream always closed; captures report overflows, clipping, peak dBFS - spectrum power normalized to dBFS - unit tests, README troubleshooting, version 2026.10.25
This commit is contained in:
@@ -0,0 +1,69 @@
|
||||
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])
|
||||
|
||||
|
||||
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
|
||||
|
||||
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_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
|
||||
@@ -0,0 +1,75 @@
|
||||
import json
|
||||
import subprocess
|
||||
import sys
|
||||
|
||||
from mcbladerf import cli_fallback
|
||||
|
||||
# bladeRF-cli 1.10 output for `-e info -e version`
|
||||
INFO_OUTPUT = """
|
||||
Board: Nuand bladeRF (bladerf1)
|
||||
Serial #: 0123456789abcdef0123456789abcdef
|
||||
VCTCXO DAC calibration: 0x8f68
|
||||
FPGA size: 40 KLE
|
||||
FPGA loaded: yes
|
||||
Flash size: 32 Mbit
|
||||
USB bus: 4
|
||||
USB address: 16
|
||||
USB speed: SuperSpeed
|
||||
Backend: libusb
|
||||
Instance: 0
|
||||
|
||||
bladeRF-cli version: 1.10.0-git-fcf94233-dirty
|
||||
libbladeRF version: 2.6.0-git-fcf94233-dirty
|
||||
|
||||
Firmware version: 2.6.0-git-09c82087
|
||||
FPGA version: 0.16.0 (configured from SPI flash)
|
||||
"""
|
||||
|
||||
|
||||
def test_device_info_parses_firmware_version(monkeypatch):
|
||||
monkeypatch.setattr(cli_fallback, "_run_cli", lambda *a, **k: INFO_OUTPUT)
|
||||
|
||||
info = cli_fallback.get_device_info()
|
||||
|
||||
assert info.fw_version == "2.6.0-git-09c82087"
|
||||
assert info.fpga_version == "0.16.0"
|
||||
assert info.fpga_loaded is True
|
||||
assert info.usb_speed == "SuperSpeed"
|
||||
|
||||
|
||||
def test_calibrate_dc_rejects_help_output(monkeypatch):
|
||||
monkeypatch.setattr(
|
||||
cli_fallback, "_run_cli", lambda *a, **k: "\nDescription of calibrate command:\n"
|
||||
)
|
||||
try:
|
||||
cli_fallback.calibrate_dc()
|
||||
except RuntimeError as e:
|
||||
assert "rejected" in str(e)
|
||||
else:
|
||||
raise AssertionError("help text was reported as a successful calibration")
|
||||
|
||||
|
||||
def test_stdout_carries_only_json_rpc():
|
||||
request = {
|
||||
"jsonrpc": "2.0",
|
||||
"id": 1,
|
||||
"method": "initialize",
|
||||
"params": {
|
||||
"protocolVersion": "2025-06-18",
|
||||
"capabilities": {},
|
||||
"clientInfo": {"name": "test", "version": "0"},
|
||||
},
|
||||
}
|
||||
proc = subprocess.run(
|
||||
[sys.executable, "-c", "from mcbladerf.server import main; main()"],
|
||||
input=json.dumps(request) + "\n",
|
||||
capture_output=True,
|
||||
text=True,
|
||||
timeout=30,
|
||||
)
|
||||
|
||||
lines = [line for line in proc.stdout.splitlines() if line.strip()]
|
||||
assert lines, proc.stderr
|
||||
for line in lines:
|
||||
json.loads(line)
|
||||
assert json.loads(lines[0])["result"]["serverInfo"]["name"] == "mcbladerf"
|
||||
Reference in New Issue
Block a user