warehack.ing/mcbladerf
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge branch 'fix/streaming-and-cli-bugs'

Browse Source
This commit is contained in:
rsp2k
2026-10-25 10:53:53 -06:00
parent 31d771e742 492796c34e
commit 067067e1e7
10 changed files with 524 additions and 249 deletions
Download Patch File Download Diff File Expand all files Collapse all files
README.md
+17 -1
View File
@@ -82,7 +82,7 @@ claude -p "show bladerf status" --mcp-config .mcp.json --allowedTools "mcp__mcbl
### IQ Capture
- `bladerf_capture_iq` - Stream samples to file (CF32/CS16)
- `bladerf_capture_spectrum` - Capture and compute FFT
- `bladerf_capture_spectrum` - Capture and compute FFT, power in dBFS
- `bladerf_stream_status` - Check capture progress
- `bladerf_stop_capture` - Abort running capture
@@ -110,6 +110,22 @@ claude -p "show bladerf status" --mcp-config .mcp.json --allowedTools "mcp__mcbl
[Uses bladerf_capture_spectrum]
```
## Capture Files and Readings
CF32 files are complex64 scaled to ±1.0. CS16 files are the device's raw SC16 Q11 integers, interleaved I/Q with full scale 2048, the same layout `bladeRF-cli` writes with `format=bin`. Spectrum power is dBFS, so a full-scale tone reads 0. Every capture reports `overflows` (dropped-sample events), `clipped_samples` and `peak_dbfs`, so check those before trusting the data.
Gains you leave out keep whatever the device last had, including values set by another process. The radio is often shared, so set all three RX gains explicitly.
While connected, the server holds the USB device and nothing else can open it. The bladeRF-cli backed tools (probe, calibrate, FPGA load, firmware flash, hardware info) release it for the duration and restore the RX/TX settings afterwards. Call `bladerf_disconnect` when you are done so other tools can use the radio.
## Troubleshooting
**`setupStream() -3` / `bladerf_sync_config() returned -3`.** The SoapySDR plugin was built against an older libbladeRF. libbladeRF 2025.10 inserted `BLADERF_FORMAT_SC16_Q11_PACKED` into the `bladerf_format` enum without changing the soname, so a stale `soapybladerf` requests the wrong format and bladeRF1 rejects it. Rebuild `soapybladerf` against the installed headers (the Arch PKGBUILD needs `-DCMAKE_POLICY_VERSION_MINIMUM=3.5` under CMake 4).
**`No device(s) available` (-7) with the board plugged in.** Another process has the device open. Check `pgrep -xa bladeRF-cli`; `pgrep -f` can match your own shell.
**`No module named 'SoapySDR'`.** The venv must see system site-packages. Recreate it with `make setup`.
## BladeRF 1.0 x40 Specifications
- Frequency: 237.5 MHz - 3.8 GHz
pyproject.toml
+1 -1
View File
@@ -1,6 +1,6 @@
[project]
name = "mcbladerf"
version = "2026.02.08"
version = "2026.10.25"
description = "MCP server for Nuand BladeRF SDR control via SoapySDR"
authors = [{name = "Ryan Malloy", email = "ryan@supported.systems"}]
readme = "README.md"
src/mcbladerf/__init__.py
+1 -1
View File
@@ -1,3 +1,3 @@
"""MCP server for Nuand BladeRF SDR control via SoapySDR."""
__version__ = "2026.02.08"
__version__ = "2026.10.25"
src/mcbladerf/capture.py
+178 -170
View File
@@ -8,7 +8,8 @@ from __future__ import annotations
import asyncio
import time
from dataclasses import dataclass
from contextlib import closing
from dataclasses import dataclass, field
from pathlib import Path
from typing import TYPE_CHECKING, Any, Literal
@@ -17,10 +18,19 @@ import numpy as np
from . import device
if TYPE_CHECKING:
from collections.abc import Callable
from collections.abc import Callable, Iterator
SampleFormat = Literal["cf32", "cs16"]
# Samples per readStream call (the bladeRF plugin returns at most 4096 per call)
_CHUNK = 65536
# Samples dropped after stream activation while the LMS6002D settles
_SETTLE_SAMPLES = 16384
# Give up if the stream delivers nothing for this long
_STALL_SECONDS = 2.0
# Q11 saturates at +/-2047 (or -2048); anything at or past this is clipped
_CLIP_LEVEL = 2047 / device.CS16_FULL_SCALE
@dataclass
class CaptureResult:
@@ -33,11 +43,16 @@ class CaptureResult:
frequency_hz: float
format: SampleFormat
file_size_bytes: int
gains_db: dict[str, float] = field(default_factory=dict)
overflows: int = 0
clipped_samples: int = 0
peak_dbfs: float | None = None
cancelled: bool = False
@dataclass
class SpectrumResult:
"""Result from a spectrum capture operation."""
"""Result from a spectrum capture operation. Power is in dBFS."""
center_frequency_hz: float
bandwidth_hz: float
@@ -46,6 +61,9 @@ class SpectrumResult:
power_db: list[float]
peak_frequency_hz: float
peak_power_db: float
gains_db: dict[str, float] = field(default_factory=dict)
overflows: int = 0
clipped_samples: int = 0
@dataclass
@@ -59,22 +77,24 @@ class CaptureProgress:
bytes_written: int
@dataclass
class _StreamStats:
overflows: int = 0
clipped: int = 0
peak: float = 0.0 # largest |I| or |Q| as a fraction of full scale
@property
def peak_dbfs(self) -> float | None:
return round(20 * np.log10(self.peak), 2) if self.peak > 0 else None
def _get_capture_state() -> dict[str, Any]:
"""Get or create capture state in device state."""
state = device.get_state()
if not hasattr(state, "capture_info"):
state.capture_info = {
"active": False,
"progress": None,
"cancelled": False,
}
return state.capture_info
return device.get_state().capture_info
def is_capture_active() -> bool:
"""Check if a capture is currently in progress."""
state = _get_capture_state()
return state.get("active", False)
return _get_capture_state()["active"]
def get_capture_progress() -> CaptureProgress | None:
@@ -83,8 +103,7 @@ def get_capture_progress() -> CaptureProgress | None:
Returns:
CaptureProgress if capture is active, None otherwise.
"""
state = _get_capture_state()
return state.get("progress")
return _get_capture_state()["progress"]
def cancel_capture() -> bool:
@@ -94,12 +113,89 @@ def cancel_capture() -> bool:
True if a capture was active and cancellation was requested.
"""
state = _get_capture_state()
if state.get("active"):
if state["active"]:
state["cancelled"] = True
return True
return False
def _configure_rx(
frequency_hz: float,
sample_rate_hz: float,
bandwidth_hz: float | None,
gains: dict[str, float] | None,
) -> device.RFConfig:
"""Apply RX settings and return what the hardware actually took.
Gains not given keep their current value, which may be left over from an
earlier call or another process.
"""
device.set_sample_rate("rx", sample_rate_hz)
device.set_frequency("rx", frequency_hz)
device.set_bandwidth("rx", bandwidth_hz if bandwidth_hz is not None else sample_rate_hz * 0.8)
for stage, gain_db in (gains or {}).items():
device.set_gain("rx", stage, gain_db)
return device.get_rf_config("rx")
def _components(chunk: np.ndarray, soapy_format: str) -> np.ndarray:
"""|I| and |Q| as fractions of full scale, shape (n, 2)."""
if soapy_format == "CS16":
return np.abs(chunk.reshape(-1, 2).astype(np.float32)) / device.CS16_FULL_SCALE
return np.abs(chunk.view(np.float32).reshape(-1, 2))
def _rx_chunks(
total_samples: int,
soapy_format: str,
stats: _StreamStats,
cancelled: Callable[[], bool] = lambda: False,
) -> Iterator[np.ndarray]:
"""Stream RX samples, yielding views into a reused buffer.
Consumers must copy or write each chunk before asking for the next. CF32
chunks are complex64; CS16 chunks are interleaved int16 I/Q. The stream is
always closed on exit, including on error or early close.
"""
buff = device.alloc_buffer(_CHUNK, soapy_format)
per_sample = 2 if soapy_format == "CS16" else 1
device.setup_rx_stream(soapy_format)
try:
device.activate_rx_stream()
to_discard = _SETTLE_SAMPLES
got = 0
last_data = time.monotonic()
while got < total_samples and not cancelled():
want = min(_CHUNK, to_discard or total_samples - got)
n, overflow = device.read_stream(buff, want)
if overflow:
last_data = time.monotonic()
if not to_discard:
stats.overflows += 1
continue
if n == 0:
if time.monotonic() - last_data > _STALL_SECONDS:
raise RuntimeError(f"RX stream stalled: no samples for {_STALL_SECONDS:.0f} s")
continue
last_data = time.monotonic()
if to_discard:
to_discard = max(0, to_discard - n)
continue
chunk = buff[: n * per_sample]
comps = _components(chunk, soapy_format)
stats.peak = max(stats.peak, float(comps.max()))
stats.clipped += int(np.count_nonzero(comps.max(axis=1) >= _CLIP_LEVEL))
got += n
yield chunk
finally:
device.close_rx_stream()
def capture_iq_sync(
frequency_hz: float,
sample_rate_hz: float,
@@ -112,6 +208,9 @@ def capture_iq_sync(
) -> CaptureResult:
"""Capture IQ samples to a file (synchronous version).
CF32 files hold complex64 scaled to +/-1.0. CS16 files hold the device's
raw SC16 Q11 integers (full scale 2048), interleaved I/Q.
Args:
frequency_hz: Center frequency in Hz
sample_rate_hz: Sample rate in Hz
@@ -138,73 +237,27 @@ def capture_iq_sync(
state["cancelled"] = False
try:
# Configure RF parameters
device.set_frequency("rx", frequency_hz)
device.set_sample_rate("rx", sample_rate_hz)
if bandwidth_hz is None:
bandwidth_hz = sample_rate_hz * 0.8
device.set_bandwidth("rx", bandwidth_hz)
# Set gains if provided
if gains:
for stage, gain_db in gains.items():
device.set_gain("rx", stage, gain_db)
# Get actual configured values
actual_freq = device.get_frequency("rx")
actual_rate = device.get_sample_rate("rx")
# Calculate sample count
total_samples = int(duration_seconds * actual_rate)
# Set up stream
rf = _configure_rx(frequency_hz, sample_rate_hz, bandwidth_hz, gains)
total_samples = int(duration_seconds * rf.sample_rate_hz)
soapy_format = "CF32" if format == "cf32" else "CS16"
device.setup_rx_stream(soapy_format)
device.activate_rx_stream()
bytes_per_sample = 8 if format == "cf32" else 4
output_path = Path(output_path)
output_path.parent.mkdir(parents=True, exist_ok=True)
# Buffer size: 64K samples per read (good balance of latency/efficiency)
buffer_size = 65536
samples_captured = 0
stats = _StreamStats()
bytes_written = 0
start_time = time.time()
start_time = time.monotonic()
chunks = _rx_chunks(total_samples, soapy_format, stats, lambda: state["cancelled"])
with open(output_path, "wb") as f:
while samples_captured < total_samples:
if state["cancelled"]:
break
# Read chunk
samples_to_read = min(buffer_size, total_samples - samples_captured)
samples, flags = device.read_stream(samples_to_read)
if len(samples) == 0:
continue
# Write to file
if format == "cf32":
# Complex float32: 8 bytes per sample (real + imag as float32)
f.write(samples.astype(np.complex64).tobytes())
bytes_written += len(samples) * 8
else:
# Complex int16: 4 bytes per sample (real + imag as int16)
# Scale and convert
scaled = samples * 2047 # Scale to int16 range
int_samples = np.zeros(len(samples) * 2, dtype=np.int16)
int_samples[0::2] = scaled.real.astype(np.int16)
int_samples[1::2] = scaled.imag.astype(np.int16)
f.write(int_samples.tobytes())
bytes_written += len(samples) * 4
samples_captured += len(samples)
# Update progress
elapsed = time.time() - start_time
remaining = (elapsed / samples_captured) * (total_samples - samples_captured) if samples_captured > 0 else 0
with open(output_path, "wb") as f, closing(chunks):
for chunk in chunks:
f.write(chunk.tobytes())
bytes_written += chunk.nbytes
samples_captured = bytes_written // bytes_per_sample
elapsed = time.monotonic() - start_time
remaining = (elapsed / samples_captured) * (total_samples - samples_captured)
progress = CaptureProgress(
samples_captured=samples_captured,
samples_total=total_samples,
@@ -213,24 +266,22 @@ def capture_iq_sync(
bytes_written=bytes_written,
)
state["progress"] = progress
if progress_callback:
progress_callback(progress)
# Clean up stream
device.deactivate_rx_stream()
device.close_rx_stream()
actual_duration = time.time() - start_time
return CaptureResult(
file_path=str(output_path),
sample_count=samples_captured,
duration_seconds=actual_duration,
sample_rate_hz=actual_rate,
frequency_hz=actual_freq,
sample_count=bytes_written // bytes_per_sample,
duration_seconds=time.monotonic() - start_time,
sample_rate_hz=rf.sample_rate_hz,
frequency_hz=rf.frequency_hz,
format=format,
file_size_bytes=bytes_written,
gains_db=rf.gains,
overflows=stats.overflows,
clipped_samples=stats.clipped,
peak_dbfs=stats.peak_dbfs,
cancelled=state["cancelled"],
)
finally:
@@ -267,6 +318,16 @@ async def capture_iq(
)
def _window(name: str, size: int) -> np.ndarray:
if name == "hann":
return np.hanning(size)
if name == "blackman":
return np.blackman(size)
if name == "hamming":
return np.hamming(size)
return np.ones(size)
def compute_spectrum(
samples: np.ndarray,
sample_rate_hz: float,
@@ -277,8 +338,11 @@ def compute_spectrum(
) -> SpectrumResult:
"""Compute power spectrum from IQ samples.
Power is normalized to dBFS: a full-scale complex tone (amplitude 1.0)
centred in a bin reads 0 dBFS whatever the window or FFT size.
Args:
samples: Complex IQ samples (numpy array)
samples: Complex IQ samples scaled to +/-1.0
sample_rate_hz: Sample rate of the data
center_frequency_hz: Center frequency
fft_size: FFT size (power of 2 recommended)
@@ -288,49 +352,26 @@ def compute_spectrum(
Returns:
SpectrumResult with frequency bins and power levels.
"""
# Apply window
if window == "hann":
win = np.hanning(fft_size)
elif window == "blackman":
win = np.blackman(fft_size)
elif window == "hamming":
win = np.hamming(fft_size)
else:
win = np.ones(fft_size)
# Compute averaged spectrum
num_ffts = min(averaging, len(samples) // fft_size)
if num_ffts == 0:
num_ffts = 1
fft_size = len(samples)
win = np.hanning(fft_size) if window != "none" else np.ones(fft_size)
win = _window(window, fft_size)
power_sum = np.zeros(fft_size)
for i in range(num_ffts):
start = i * fft_size
chunk = samples[start : start + fft_size]
if len(chunk) < fft_size:
chunk = np.pad(chunk, (0, fft_size - len(chunk)))
chunk = samples[i * fft_size : (i + 1) * fft_size]
spectrum = np.fft.fftshift(np.fft.fft(chunk * win))
power_sum += np.abs(spectrum) ** 2
# Apply window and FFT
windowed = chunk * win
spectrum = np.fft.fftshift(np.fft.fft(windowed))
power = np.abs(spectrum) ** 2
power_sum += power
power_avg = power_sum / num_ffts / np.sum(win) ** 2
power_db = 10 * np.log10(power_avg + 1e-20)
# Average and convert to dB
power_avg = power_sum / num_ffts
power_db = 10 * np.log10(power_avg + 1e-12) # Add small value to avoid log(0)
# Frequency axis
freq_bins = np.fft.fftshift(np.fft.fftfreq(fft_size, 1.0 / sample_rate_hz))
frequencies = center_frequency_hz + freq_bins
# Find peak
peak_idx = np.argmax(power_db)
peak_freq = frequencies[peak_idx]
peak_power = power_db[peak_idx]
return SpectrumResult(
center_frequency_hz=center_frequency_hz,
@@ -338,8 +379,8 @@ def compute_spectrum(
bin_count=fft_size,
frequencies_hz=frequencies.tolist(),
power_db=power_db.tolist(),
peak_frequency_hz=float(peak_freq),
peak_power_db=float(peak_power),
peak_frequency_hz=float(frequencies[peak_idx]),
peak_power_db=float(power_db[peak_idx]),
)
@@ -364,56 +405,30 @@ def capture_spectrum_sync(
window: Window function
Returns:
SpectrumResult with frequency bins and power levels.
SpectrumResult with frequency bins and power levels in dBFS.
"""
device.ensure_connected()
if is_capture_active():
raise RuntimeError("A capture is already in progress")
# Configure RF
device.set_frequency("rx", frequency_hz)
device.set_sample_rate("rx", sample_rate_hz)
rf = _configure_rx(frequency_hz, sample_rate_hz, bandwidth_hz, gains)
if bandwidth_hz is None:
bandwidth_hz = sample_rate_hz * 0.8
device.set_bandwidth("rx", bandwidth_hz)
stats = _StreamStats()
with closing(_rx_chunks(fft_size * averaging, "CF32", stats)) as chunks:
samples = np.concatenate([c.copy() for c in chunks])
if gains:
for stage, gain_db in gains.items():
device.set_gain("rx", stage, gain_db)
# Get actual values
actual_freq = device.get_frequency("rx")
actual_rate = device.get_sample_rate("rx")
# Capture enough samples for averaging
total_samples = fft_size * averaging
device.setup_rx_stream("CF32")
device.activate_rx_stream()
try:
# Collect samples
all_samples = []
remaining = total_samples
while remaining > 0:
samples, flags = device.read_stream(min(remaining, 65536))
all_samples.append(samples)
remaining -= len(samples)
samples = np.concatenate(all_samples)[:total_samples]
finally:
device.deactivate_rx_stream()
device.close_rx_stream()
return compute_spectrum(
result = compute_spectrum(
samples=samples,
sample_rate_hz=actual_rate,
center_frequency_hz=actual_freq,
sample_rate_hz=rf.sample_rate_hz,
center_frequency_hz=rf.frequency_hz,
fft_size=fft_size,
window=window,
averaging=averaging,
)
result.gains_db = rf.gains
result.overflows = stats.overflows
result.clipped_samples = stats.clipped
return result
async def capture_spectrum(
@@ -446,7 +461,7 @@ def read_iq_file(
format: SampleFormat = "cf32",
max_samples: int | None = None,
) -> np.ndarray:
"""Read IQ samples from a file.
"""Read IQ samples from a file, scaled to +/-1.0.
Args:
file_path: Path to the IQ file
@@ -456,18 +471,11 @@ def read_iq_file(
Returns:
Complex numpy array of samples.
"""
file_path = Path(file_path)
with open(file_path, "rb") as f:
data = f.read()
if format == "cf32":
# Complex float32: 8 bytes per sample
samples = np.frombuffer(data, dtype=np.complex64)
samples = np.fromfile(file_path, dtype=np.complex64)
else:
# Complex int16: 4 bytes per sample
int_data = np.frombuffer(data, dtype=np.int16)
samples = (int_data[0::2] + 1j * int_data[1::2]).astype(np.complex64) / 2047.0
int_data = np.fromfile(file_path, dtype=np.int16)
samples = (int_data[0::2] + 1j * int_data[1::2]).astype(np.complex64) / device.CS16_FULL_SCALE
if max_samples is not None:
samples = samples[:max_samples]
src/mcbladerf/cli_fallback.py
+10 -3
View File
@@ -133,8 +133,8 @@ def get_device_info(serial: str | None = None) -> CLIDeviceInfo:
if match:
info["fpga_version"] = match.group(1)
# Parse firmware version
match = re.search(r"FW version:\s*(\S+)", output)
# Parse firmware version ("Firmware version:" in bladeRF-cli 1.10)
match = re.search(r"(?:FW|Firmware) version:\s*(\S+)", output)
if match:
info["fw_version"] = match.group(1)
@@ -255,13 +255,20 @@ def flash_firmware(img_path: str | Path, serial: str | None = None) -> dict[str,
def calibrate_dc(serial: str | None = None) -> dict[str, str]:
"""Run DC offset calibration.
Runs the LMS6002D internal auto-calibrations, then the RX/TX I/Q DC
correction for the current frequency and gain settings. Results live in
device registers, so they persist after bladeRF-cli exits.
Args:
serial: Device serial
Returns:
Dict with calibration status.
"""
output = _run_cli(["cal dc"], serial=serial, timeout=60)
# A bare "cal dc" prints the command help and exits 0 without calibrating
output = _run_cli(["calibrate lms", "calibrate dc rxtx"], serial=serial, timeout=120)
if "Usage: calibrate" in output or "Description of calibrate" in output:
raise RuntimeError(f"bladeRF-cli rejected the calibrate command: {output.strip()[:300]}")
return {
"status": "completed",
src/mcbladerf/device.py
+112 -22
View File
@@ -7,6 +7,8 @@ device connection and stream state across MCP tool calls.
from __future__ import annotations
import asyncio
from collections.abc import Iterator
from contextlib import contextmanager
from dataclasses import dataclass, field
from typing import Any, Literal
@@ -22,6 +24,18 @@ except ImportError:
SOAPY_SDR_RX = 0
SOAPY_SDR_TX = 1
# SoapySDR readStream error codes (SoapySDR/Errors.h)
SOAPY_SDR_TIMEOUT = -1
SOAPY_SDR_OVERFLOW = -4
# bladeRF SC16 Q11: CS16 samples are raw integers, CF32 is CS16 / 2048
CS16_FULL_SCALE = 2048.0
LOOPBACK_MODES = (
"none", "firmware", "bb_txlpf_rxvga2", "bb_txvga1_rxvga2",
"bb_txlpf_rxlpf", "bb_txvga1_rxlpf", "rf_lna1", "rf_lna2", "rf_lna3",
)
DirectionType = Literal["rx", "tx"]
@@ -71,7 +85,9 @@ class BladeRFState:
active_rx_stream: Any = None
active_tx_stream: Any = None
capture_task: asyncio.Task | None = None
lock: asyncio.Lock = field(default_factory=asyncio.Lock)
capture_info: dict[str, Any] = field(
default_factory=lambda: {"active": False, "progress": None, "cancelled": False}
)
# Global state singleton
@@ -202,6 +218,62 @@ def ensure_connected() -> None:
raise RuntimeError("No BladeRF device connected. Use bladerf_connect first.")
@contextmanager
def released() -> Iterator[str | None]:
"""Temporarily close the SoapySDR handle so another process can open the device.
libusb lets only one process claim the bladeRF, so bladeRF-cli fails with
"No devices available" while we hold it. RX/TX settings are restored on
reconnect. Yields the serial of the released device, or None if nothing
was connected.
"""
state = get_state()
if state.device is None:
yield None
return
if state.capture_info["active"]:
raise RuntimeError("Cannot release the device while a capture is running")
serial = state.serial
saved = {d: get_rf_config(d) for d in ("rx", "tx")}
disconnect()
try:
yield serial
finally:
try:
connect(serial=serial)
except RuntimeError as e:
raise RuntimeError(
f"{e}. The device was released for bladeRF-cli and could not be "
"reopened; another process may have claimed it."
) from e
for direction, cfg in saved.items():
_restore_rf_config(direction, cfg)
def _restore_rf_config(direction: DirectionType, cfg: RFConfig) -> None:
device = get_device()
soapy_dir = direction_to_soapy(direction)
device.setSampleRate(soapy_dir, 0, cfg.sample_rate_hz)
device.setFrequency(soapy_dir, 0, cfg.frequency_hz)
device.setBandwidth(soapy_dir, 0, cfg.bandwidth_hz)
for stage, gain in cfg.gains.items():
device.setGain(soapy_dir, 0, stage, gain)
def set_loopback(mode: str) -> str:
"""Set loopback mode through the plugin's settings API.
Returns:
Loopback mode read back from the device.
"""
if mode not in LOOPBACK_MODES:
raise RuntimeError(f"Invalid loopback mode. Valid modes: {list(LOOPBACK_MODES)}")
device = get_device()
device.writeSetting("loopback", mode)
return device.readSetting("loopback")
def get_device() -> Any:
"""Get the SoapySDR device object.
@@ -393,13 +465,22 @@ def get_frequency_range(direction: DirectionType, channel: int = 0) -> dict[str,
"""
device = get_device()
ranges = device.getFrequencyRange(direction_to_soapy(direction), channel)
# SoapySDR returns list of ranges, usually just one for BladeRF
if ranges:
return range_span(ranges) or {"min_hz": 237.5e6, "max_hz": 3.8e9} # BladeRF 1.0 defaults
def range_span(ranges: Any) -> dict[str, float] | None:
"""Overall min/max of a SoapySDR RangeList.
The bladeRF plugin splits sample rates into several ranges with different
step sizes (80k-10M, 10M-20M, 20M-40M), so the first range alone is not the
device limit.
"""
if not ranges:
return None
return {
"min_hz": ranges[0].minimum(),
"max_hz": ranges[0].maximum(),
"min_hz": min(r.minimum() for r in ranges),
"max_hz": max(r.maximum() for r in ranges),
}
return {"min_hz": 237.5e6, "max_hz": 3.8e9} # BladeRF 1.0 defaults
def get_sample_rate_range(direction: DirectionType, channel: int = 0) -> dict[str, float]:
@@ -410,12 +491,7 @@ def get_sample_rate_range(direction: DirectionType, channel: int = 0) -> dict[st
"""
device = get_device()
ranges = device.getSampleRateRange(direction_to_soapy(direction), channel)
if ranges:
return {
"min_hz": ranges[0].minimum(),
"max_hz": ranges[0].maximum(),
}
return {"min_hz": 80e3, "max_hz": 40e6} # BladeRF 1.0 defaults
return range_span(ranges) or {"min_hz": 80e3, "max_hz": 40e6} # BladeRF 1.0 defaults
def get_rf_config(direction: DirectionType, channel: int = 0) -> RFConfig:
@@ -485,17 +561,29 @@ def deactivate_rx_stream() -> None:
device.deactivateStream(state.active_rx_stream)
def read_stream(
num_samples: int, timeout_us: int = 100000
) -> tuple[np.ndarray, int]:
"""Read samples from the active RX stream.
def alloc_buffer(num_samples: int, format_str: str) -> np.ndarray:
"""Allocate a stream buffer matching the SoapySDR stream format.
CF32 is one complex64 per sample. CS16 is interleaved int16 I/Q, two per
sample; SoapySDR writes raw bytes, so a complex64 buffer would be filled
with integers it then misreads as floats.
"""
if format_str == "CS16":
return np.zeros(2 * num_samples, dtype=np.int16)
return np.zeros(num_samples, dtype=np.complex64)
def read_stream(buff: np.ndarray, num_samples: int, timeout_us: int = 100000) -> tuple[int, bool]:
"""Read samples from the active RX stream into a preallocated buffer.
Args:
num_samples: Number of complex samples to read
buff: Buffer from alloc_buffer()
num_samples: Maximum complex samples to read (may return fewer)
timeout_us: Timeout in microseconds
Returns:
Tuple of (samples array, flags).
Tuple of (samples read, overflow). Timeouts and overflows return 0
samples rather than raising; an overflow means samples were dropped.
"""
state = get_state()
device = get_device()
@@ -503,14 +591,16 @@ def read_stream(
if state.active_rx_stream is None:
raise RuntimeError("No RX stream active.")
# Allocate buffer - SoapySDR expects list of buffers, one per channel
buff = np.zeros(num_samples, dtype=np.complex64)
sr = device.readStream(state.active_rx_stream, [buff], num_samples, timeoutUs=timeout_us)
if sr.ret == SOAPY_SDR_TIMEOUT:
return 0, False
if sr.ret == SOAPY_SDR_OVERFLOW:
return 0, True
if sr.ret < 0:
raise RuntimeError(f"Stream read error: {sr.ret}")
raise RuntimeError(f"Stream read error: {SoapySDR.errToStr(sr.ret)} ({sr.ret})")
return buff[:sr.ret], sr.flags
return sr.ret, False
def close_rx_stream() -> None:
src/mcbladerf/server.py
+58 -48
View File
@@ -6,19 +6,35 @@ and IQ data capture capabilities through the Model Context Protocol.
from __future__ import annotations
from typing import Annotated, Literal
import sys
from collections.abc import Callable
from typing import Annotated, Any, Literal
from fastmcp import FastMCP
from . import capture, cli_fallback, device
from . import __version__, capture, cli_fallback, device
# Create FastMCP server instance
mcp = FastMCP(
"mcbladerf",
instructions="MCP server for Nuand BladeRF SDR control via SoapySDR",
version=__version__,
)
def _with_cli(fn: Callable[..., Any], serial: str | None, *args: Any) -> Any:
"""Run a bladeRF-cli helper against the device.
If the target is the device we hold open, release our USB claim for the
duration (bladeRF-cli cannot open it otherwise) and restore settings after.
"""
state = device.get_state()
if state.device is None or serial not in (None, state.serial):
return fn(*args, serial=serial)
with device.released() as held_serial:
return fn(*args, serial=held_serial)
# =============================================================================
# Device Management Tools
# =============================================================================
@@ -438,7 +454,13 @@ def bladerf_capture_iq(
"""Capture IQ samples to a file.
Configures the receiver and streams samples to disk. Supports complex
float32 (CF32, 8 bytes/sample) or complex int16 (CS16, 4 bytes/sample).
float32 (CF32, 8 bytes/sample, scaled to +/-1.0) or complex int16 (CS16,
4 bytes/sample, raw SC16 Q11 with full scale 2048).
Gains left as None keep their current device value, which may come from
an earlier call or another process; gains_db in the result shows what was
used. Check overflows (dropped-sample events) and clipped_samples before
trusting the data.
Args:
frequency_hz: Center frequency
@@ -482,6 +504,11 @@ def bladerf_capture_iq(
"frequency_hz": result.frequency_hz,
"format": result.format,
"file_size_bytes": result.file_size_bytes,
"gains_db": result.gains_db,
"overflows": result.overflows,
"clipped_samples": result.clipped_samples,
"peak_dbfs": result.peak_dbfs,
"cancelled": result.cancelled,
}
except RuntimeError as e:
return {"status": "error", "message": str(e)}
@@ -501,7 +528,9 @@ def bladerf_capture_spectrum(
"""Capture and compute power spectrum (FFT).
Captures IQ samples and computes an averaged power spectrum.
Returns frequency bins and power levels in dB.
Returns frequency bins and power levels in dBFS (0 dBFS is a full-scale
tone). Gains left as None keep their current device value. A nonzero
clipped_samples means the ADC saturated; lower the gain.
Args:
frequency_hz: Center frequency
@@ -539,9 +568,13 @@ def bladerf_capture_spectrum(
"bandwidth_hz": result.bandwidth_hz,
"bin_count": result.bin_count,
"frequencies_hz": result.frequencies_hz,
"power_units": "dBFS",
"power_db": result.power_db,
"peak_frequency_hz": result.peak_frequency_hz,
"peak_power_db": result.peak_power_db,
"gains_db": result.gains_db,
"overflows": result.overflows,
"clipped_samples": result.clipped_samples,
}
except RuntimeError as e:
return {"status": "error", "message": str(e)}
@@ -597,7 +630,7 @@ def bladerf_load_fpga(
"""Load FPGA bitstream from file (RAM, not flash).
Loads the FPGA configuration to RAM. This is NOT persistent across
power cycles. Use bladerf_flash_fpga for persistent storage.
power cycles.
Args:
rbf_path: Path to .rbf FPGA bitstream file
@@ -607,12 +640,7 @@ def bladerf_load_fpga(
Dict with load status.
"""
try:
# Use connected device's serial if not specified
if serial is None and device.is_connected():
serial = device.get_state().serial
result = cli_fallback.load_fpga(rbf_path, serial)
return result
return _with_cli(cli_fallback.load_fpga, serial, rbf_path)
except RuntimeError as e:
return {"status": "error", "message": str(e)}
@@ -635,11 +663,7 @@ def bladerf_flash_firmware(
Dict with flash status.
"""
try:
if serial is None and device.is_connected():
serial = device.get_state().serial
result = cli_fallback.flash_firmware(img_path, serial)
return result
return _with_cli(cli_fallback.flash_firmware, serial, img_path)
except RuntimeError as e:
return {"status": "error", "message": str(e)}
@@ -650,7 +674,9 @@ def bladerf_calibrate_dc(
) -> dict:
"""Run DC offset calibration.
Calibrates the DC offset to reduce LO leakage and improve signal quality.
Runs the LMS6002D auto-calibrations, then RX/TX I/Q DC correction for the
current frequency and gains, to reduce LO leakage and the centre spike.
Tune first, then calibrate.
Args:
serial: Device serial
@@ -659,11 +685,7 @@ def bladerf_calibrate_dc(
Dict with calibration status.
"""
try:
if serial is None and device.is_connected():
serial = device.get_state().serial
result = cli_fallback.calibrate_dc(serial)
return result
return _with_cli(cli_fallback.calibrate_dc, serial)
except RuntimeError as e:
return {"status": "error", "message": str(e)}
@@ -684,10 +706,7 @@ def bladerf_get_hardware_info(
Dict with detailed hardware information.
"""
try:
if serial is None and device.is_connected():
serial = device.get_state().serial
info = cli_fallback.get_device_info(serial)
info = _with_cli(cli_fallback.get_device_info, serial)
return {
"serial": info.serial,
"fpga_size": info.fpga_size,
@@ -720,6 +739,9 @@ def bladerf_set_loopback(
- bb_*: Baseband analog loopback
- rf_*: RF loopback through LNA
Loopback persists on the device after this server disconnects, and the
radio may be shared with other tools, so set it back to "none" when done.
Args:
mode: Loopback mode name
serial: Device serial
@@ -728,11 +750,10 @@ def bladerf_set_loopback(
Dict with status.
"""
try:
if serial is None and device.is_connected():
serial = device.get_state().serial
result = cli_fallback.set_loopback(mode, serial)
return result
state = device.get_state()
if state.device is not None and serial in (None, state.serial):
return {"status": "success", "mode": device.set_loopback(mode)}
return cli_fallback.set_loopback(mode, serial)
except RuntimeError as e:
return {"status": "error", "message": str(e)}
@@ -752,11 +773,7 @@ def bladerf_probe(
Dict with probe output.
"""
try:
if serial is None and device.is_connected():
serial = device.get_state().serial
result = cli_fallback.probe_device(serial)
return result
return _with_cli(cli_fallback.probe_device, serial)
except RuntimeError as e:
return {"error": str(e)}
@@ -768,19 +785,12 @@ def bladerf_probe(
def main():
"""Start the mcbladerf MCP server."""
try:
from importlib.metadata import version
package_version = version("mcbladerf")
except Exception:
from . import __version__
package_version = __version__
# stdout is the stdio JSON-RPC channel; the banner goes to stderr
print(f"📡 mcbladerf v{__version__}", file=sys.stderr)
print(" MCP server for Nuand BladeRF SDR control", file=sys.stderr)
print(" Using SoapySDR for device access", file=sys.stderr)
print(f"📡 mcbladerf v{package_version}")
print(" MCP server for Nuand BladeRF SDR control")
print(" Using SoapySDR for device access")
print()
mcp.run()
mcp.run(show_banner=False)
if __name__ == "__main__":
tests/test_capture.py
+69
View File
@@ -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
tests/test_cli_and_server.py
+75
View File
@@ -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"
uv.lock
Generated
+1 -1
View File
@@ -682,7 +682,7 @@ wheels = [
[[package]]
name = "mcbladerf"
version = "2026.2.8"
version = "2026.10.25"
source = { editable = "." }
dependencies = [
{ name = "fastmcp" },