Merge branch 'fix/overflow-gaps'

This commit is contained in:
2026-10-25 11:57:25 -06:00
7 changed files with 642 additions and 178 deletions
+1
View File
@@ -6,3 +6,4 @@ __pycache__/
dist/ dist/
*.cf32 *.cf32
*.cs16 *.cs16
captures/
+9 -3
View File
@@ -83,7 +83,7 @@ claude -p "show bladerf status" --mcp-config .mcp.json --allowedTools "mcp__mcbl
### IQ Capture ### IQ Capture
- `bladerf_capture_iq` - Stream samples to file (CF32/CS16) - `bladerf_capture_iq` - Stream samples to file (CF32/CS16)
- `bladerf_capture_spectrum` - Capture and compute FFT, power in dBFS - `bladerf_capture_spectrum` - Capture and compute FFT, power in dBFS
- `bladerf_stream_status` - Check capture progress - `bladerf_stream_status` - Capture progress, or the result of a background capture
- `bladerf_stop_capture` - Abort running capture - `bladerf_stop_capture` - Abort running capture
### Advanced ### Advanced
@@ -103,7 +103,7 @@ claude -p "show bladerf status" --mcp-config .mcp.json --allowedTools "mcp__mcbl
> Set frequency to 915 MHz with 2 MHz sample rate > Set frequency to 915 MHz with 2 MHz sample rate
[Uses bladerf_set_frequency and bladerf_set_sample_rate] [Uses bladerf_set_frequency and bladerf_set_sample_rate]
> Capture 1 second of IQ data to /tmp/capture.cf32 > Capture 1 second of IQ data to ~/claude/sdr/mcbladerf/captures/capture.cf32
[Uses bladerf_capture_iq] [Uses bladerf_capture_iq]
> Show me the spectrum around 100 MHz > Show me the spectrum around 100 MHz
@@ -112,7 +112,13 @@ claude -p "show bladerf status" --mcp-config .mcp.json --allowedTools "mcp__mcbl
## Capture Files and Readings ## 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. 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 `clipped_samples` and `peak_dbfs`, so check those before trusting the data.
When the host falls behind, the driver drops samples, and not always with an overflow error (a libusb transfer timeout drops data silently). Each read carries a hardware timestamp, so every gap is measured exactly and, by default, zero-filled: sample N of the file is always N / rate seconds after the first. Such captures finish as `completed_with_gaps` with a `warning`, `samples_dropped` and a gap list in output-sample offsets. Pass `fill_gaps=false` to get a shorter file with the same report instead.
Long captures should use `background=true`: the call returns at once, `bladerf_stream_status` shows progress and then the result, and `bladerf_stop_capture` ends it early. Spectra come back as a 256-point max-hold summary with `noise_floor_db`; peak and floor are computed at full resolution. Out-of-range gain, frequency, rate or bandwidth values are rejected, not clamped. The sample rate resets to 4 MHz each time the device is opened, so always set it.
Write captures to disk, not `/tmp`. On many Linux systems `/tmp` is a tmpfs held in RAM, and 30 Msps CS16 is 120 MB per second, so a few long captures can push the machine into swap and the OOM killer. Once the host stalls, the stream drops samples. The project's `captures/` directory is gitignored for this.
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. 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.
+261 -74
View File
@@ -7,6 +7,7 @@ and computing power spectra from captured data.
from __future__ import annotations from __future__ import annotations
import asyncio import asyncio
import threading
import time import time
from contextlib import closing from contextlib import closing
from dataclasses import dataclass, field from dataclasses import dataclass, field
@@ -22,14 +23,24 @@ if TYPE_CHECKING:
SampleFormat = Literal["cf32", "cs16"] SampleFormat = Literal["cf32", "cs16"]
# Samples per readStream call (the bladeRF plugin returns at most 4096 per call) # Samples per readStream call, matching the stream buffer length
_CHUNK = 65536 _CHUNK = device.STREAM_BUFLEN
# Samples dropped after stream activation while the LMS6002D settles # Samples dropped after stream activation while the LMS6002D settles
_SETTLE_SAMPLES = 16384 _SETTLE_SAMPLES = 16384
# Give up if the stream delivers nothing for this long # Give up if the stream delivers nothing for this long
_STALL_SECONDS = 2.0 _STALL_SECONDS = 2.0
# Q11 saturates at +/-2047 (or -2048); anything at or past this is clipped # Q11 saturates at +/-2047 (or -2048); anything at or past this is clipped
_CLIP_LEVEL = 2047 / device.CS16_FULL_SCALE _CLIP_Q11 = 2047
# Gap entries kept in a result; the totals always cover every gap
_MAX_GAPS_REPORTED = 50
@dataclass
class Gap:
"""Samples the driver dropped. offset is where the gap starts in the output."""
offset_samples: int
length_samples: int
@dataclass @dataclass
@@ -45,10 +56,29 @@ class CaptureResult:
file_size_bytes: int file_size_bytes: int
gains_db: dict[str, float] = field(default_factory=dict) gains_db: dict[str, float] = field(default_factory=dict)
overflows: int = 0 overflows: int = 0
samples_dropped: int = 0
unmeasured_overflows: int = 0
gaps: list[Gap] = field(default_factory=list)
gap_count: int = 0
zero_filled: bool = True
clipped_samples: int = 0 clipped_samples: int = 0
peak_dbfs: float | None = None peak_dbfs: float | None = None
cancelled: bool = False cancelled: bool = False
@property
def warning(self) -> str | None:
if self.samples_dropped or self.unmeasured_overflows:
dropped_ms = 1000 * self.samples_dropped / self.sample_rate_hz
how = "replaced with zeros" if self.zero_filled else "missing from the file"
msg = (
f"{self.overflows} overflow(s): {self.samples_dropped} samples "
f"({dropped_ms:.1f} ms) were dropped by the host and are {how}."
)
if self.unmeasured_overflows:
msg += f" {self.unmeasured_overflows} overflow(s) had no timestamp to size the gap."
return msg
return None
@dataclass @dataclass
class SpectrumResult: class SpectrumResult:
@@ -61,8 +91,10 @@ class SpectrumResult:
power_db: list[float] power_db: list[float]
peak_frequency_hz: float peak_frequency_hz: float
peak_power_db: float peak_power_db: float
noise_floor_db: float = 0.0
gains_db: dict[str, float] = field(default_factory=dict) gains_db: dict[str, float] = field(default_factory=dict)
overflows: int = 0 overflows: int = 0
samples_dropped: int = 0
clipped_samples: int = 0 clipped_samples: int = 0
@@ -80,6 +112,10 @@ class CaptureProgress:
@dataclass @dataclass
class _StreamStats: class _StreamStats:
overflows: int = 0 overflows: int = 0
unmeasured_overflows: int = 0
dropped: int = 0
gaps: list[Gap] = field(default_factory=list)
gap_count: int = 0
clipped: int = 0 clipped: int = 0
peak: float = 0.0 # largest |I| or |Q| as a fraction of full scale peak: float = 0.0 # largest |I| or |Q| as a fraction of full scale
@@ -87,6 +123,12 @@ class _StreamStats:
def peak_dbfs(self) -> float | None: def peak_dbfs(self) -> float | None:
return round(20 * np.log10(self.peak), 2) if self.peak > 0 else None return round(20 * np.log10(self.peak), 2) if self.peak > 0 else None
def add_gap(self, offset: int, length: int) -> None:
self.dropped += length
self.gap_count += 1
if len(self.gaps) < _MAX_GAPS_REPORTED:
self.gaps.append(Gap(offset, length))
def _get_capture_state() -> dict[str, Any]: def _get_capture_state() -> dict[str, Any]:
return device.get_state().capture_info return device.get_state().capture_info
@@ -106,6 +148,12 @@ def get_capture_progress() -> CaptureProgress | None:
return _get_capture_state()["progress"] return _get_capture_state()["progress"]
def get_last_capture() -> tuple[CaptureResult | None, str | None]:
"""Result or error of the most recent background capture."""
state = _get_capture_state()
return state.get("last_result"), state.get("last_error")
def cancel_capture() -> bool: def cancel_capture() -> bool:
"""Request cancellation of the current capture. """Request cancellation of the current capture.
@@ -132,32 +180,53 @@ def _configure_rx(
""" """
device.set_sample_rate("rx", sample_rate_hz) device.set_sample_rate("rx", sample_rate_hz)
device.set_frequency("rx", frequency_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) if bandwidth_hz is None:
bandwidth_hz = device.default_bandwidth("rx", sample_rate_hz)
device.set_bandwidth("rx", bandwidth_hz)
for stage, gain_db in (gains or {}).items(): for stage, gain_db in (gains or {}).items():
device.set_gain("rx", stage, gain_db) device.set_gain("rx", stage, gain_db)
return device.get_rf_config("rx") return device.get_rf_config("rx")
def _components(chunk: np.ndarray, soapy_format: str) -> np.ndarray: def _chunk_stats(chunk: np.ndarray, soapy_format: str) -> tuple[float, int]:
"""|I| and |Q| as fractions of full scale, shape (n, 2).""" """Peak |I| or |Q| as a fraction of full scale, and clipped sample count.
Runs on every chunk at full rate, so CS16 stays in integers.
"""
if soapy_format == "CS16": if soapy_format == "CS16":
return np.abs(chunk.reshape(-1, 2).astype(np.float32)) / device.CS16_FULL_SCALE mag = np.abs(chunk)
return np.abs(chunk.view(np.float32).reshape(-1, 2)) per_sample = np.maximum(mag[0::2], mag[1::2])
return int(per_sample.max()) / device.CS16_FULL_SCALE, int(
np.count_nonzero(per_sample >= _CLIP_Q11)
)
mag = np.abs(chunk.view(np.float32))
per_sample = np.maximum(mag[0::2], mag[1::2])
clip = _CLIP_Q11 / device.CS16_FULL_SCALE
return float(per_sample.max()), int(np.count_nonzero(per_sample >= clip))
def _rx_chunks( def _rx_chunks(
total_samples: int, total_samples: int,
sample_rate_hz: float,
soapy_format: str, soapy_format: str,
stats: _StreamStats, stats: _StreamStats,
fill_gaps: bool = True,
cancelled: Callable[[], bool] = lambda: False, cancelled: Callable[[], bool] = lambda: False,
) -> Iterator[np.ndarray]: ) -> Iterator[np.ndarray]:
"""Stream RX samples, yielding views into a reused buffer. """Stream RX samples, yielding views into a reused buffer.
Each read carries the hardware timestamp of its first sample. A jump
between where the previous read ended and where this one starts is a gap
the driver dropped on overflow; it is sized exactly and, with fill_gaps,
yielded as zeros so sample N of the output is always N / rate seconds
after the first. total_samples counts output samples, zeros included.
Consumers must copy or write each chunk before asking for the next. CF32 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 chunks are complex64; CS16 chunks are interleaved int16 I/Q. The stream is
always closed on exit, including on error or early close. always closed on exit, including on error or early close.
""" """
buff = device.alloc_buffer(_CHUNK, soapy_format) buff = device.alloc_buffer(_CHUNK, soapy_format)
zeros = np.zeros_like(buff)
per_sample = 2 if soapy_format == "CS16" else 1 per_sample = 2 if soapy_format == "CS16" else 1
device.setup_rx_stream(soapy_format) device.setup_rx_stream(soapy_format)
@@ -165,37 +234,151 @@ def _rx_chunks(
device.activate_rx_stream() device.activate_rx_stream()
to_discard = _SETTLE_SAMPLES to_discard = _SETTLE_SAMPLES
got = 0 got = 0
# Previous read's timestamp and length. Gaps come from the difference
# between consecutive timestamps, rounded once: rounding each absolute
# timestamp separately flips by +/-1 when the device clock sits near a
# half-sample boundary, which reads as phantom 1-sample gaps.
prev_time_ns: int | None = None
prev_count = 0
pending_overflows = 0
last_data = time.monotonic() last_data = time.monotonic()
while got < total_samples and not cancelled(): while got < total_samples and not cancelled():
want = min(_CHUNK, to_discard or total_samples - got) want = min(_CHUNK, to_discard or total_samples - got)
n, overflow = device.read_stream(buff, want) r = device.read_stream(buff, want)
if overflow: if r.overflow:
last_data = time.monotonic() last_data = time.monotonic()
if not to_discard: if not to_discard:
stats.overflows += 1 stats.overflows += 1
pending_overflows += 1
continue continue
if n == 0: if r.count == 0:
if time.monotonic() - last_data > _STALL_SECONDS: if time.monotonic() - last_data > _STALL_SECONDS:
raise RuntimeError(f"RX stream stalled: no samples for {_STALL_SECONDS:.0f} s") raise RuntimeError(f"RX stream stalled: no samples for {_STALL_SECONDS:.0f} s")
continue continue
last_data = time.monotonic() last_data = time.monotonic()
if to_discard: if to_discard:
to_discard = max(0, to_discard - n) to_discard = max(0, to_discard - r.count)
continue continue
if r.time_ns is not None:
gap = 0
if prev_time_ns is not None:
elapsed = round((r.time_ns - prev_time_ns) * sample_rate_hz / 1e9)
gap = elapsed - prev_count
prev_time_ns, prev_count = r.time_ns, r.count
if gap > 0:
stats.add_gap(got, gap)
if fill_gaps:
gap = min(gap, total_samples - got)
while gap > 0:
n = min(gap, _CHUNK)
got += n
gap -= n
yield zeros[: n * per_sample]
if got >= total_samples:
break
else:
prev_time_ns = None
stats.unmeasured_overflows += pending_overflows
pending_overflows = 0
n = min(r.count, total_samples - got)
chunk = buff[: n * per_sample] chunk = buff[: n * per_sample]
comps = _components(chunk, soapy_format) peak, clipped = _chunk_stats(chunk, soapy_format)
stats.peak = max(stats.peak, float(comps.max())) stats.peak = max(stats.peak, peak)
stats.clipped += int(np.count_nonzero(comps.max(axis=1) >= _CLIP_LEVEL)) stats.clipped += clipped
got += n got += n
yield chunk yield chunk
finally: finally:
device.close_rx_stream() device.close_rx_stream()
def _claim_capture() -> dict[str, Any]:
device.ensure_connected()
state = _get_capture_state()
if state["active"]:
raise RuntimeError("A capture is already in progress")
state["active"] = True
state["cancelled"] = False
return state
def _release_capture(state: dict[str, Any]) -> None:
state["active"] = False
state["progress"] = None
state["cancelled"] = False
def _capture_iq(
state: dict[str, Any],
frequency_hz: float,
sample_rate_hz: float,
duration_seconds: float,
output_path: str | Path,
format: SampleFormat,
gains: dict[str, float] | None,
bandwidth_hz: float | None,
fill_gaps: bool,
progress_callback: Callable[[CaptureProgress], None] | None,
) -> CaptureResult:
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"
bytes_per_sample = 8 if format == "cf32" else 4
output_path = Path(output_path)
output_path.parent.mkdir(parents=True, exist_ok=True)
stats = _StreamStats()
bytes_written = 0
start_time = time.monotonic()
chunks = _rx_chunks(
total_samples, rf.sample_rate_hz, soapy_format, stats, fill_gaps, lambda: state["cancelled"]
)
with open(output_path, "wb") as f, closing(chunks):
for chunk in chunks:
f.write(chunk)
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,
elapsed_seconds=elapsed,
estimated_remaining_seconds=remaining,
bytes_written=bytes_written,
)
state["progress"] = progress
if progress_callback:
progress_callback(progress)
return CaptureResult(
file_path=str(output_path),
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,
samples_dropped=stats.dropped,
unmeasured_overflows=stats.unmeasured_overflows,
gaps=stats.gaps,
gap_count=stats.gap_count,
zero_filled=fill_gaps,
clipped_samples=stats.clipped,
peak_dbfs=stats.peak_dbfs,
cancelled=state["cancelled"],
)
def capture_iq_sync( def capture_iq_sync(
frequency_hz: float, frequency_hz: float,
sample_rate_hz: float, sample_rate_hz: float,
@@ -205,6 +388,7 @@ def capture_iq_sync(
gains: dict[str, float] | None = None, gains: dict[str, float] | None = None,
bandwidth_hz: float | None = None, bandwidth_hz: float | None = None,
progress_callback: Callable[[CaptureProgress], None] | None = None, progress_callback: Callable[[CaptureProgress], None] | None = None,
fill_gaps: bool = True,
) -> CaptureResult: ) -> CaptureResult:
"""Capture IQ samples to a file (synchronous version). """Capture IQ samples to a file (synchronous version).
@@ -220,6 +404,8 @@ def capture_iq_sync(
gains: Optional dict of gain settings {"lna": 6, "rxvga1": 25} gains: Optional dict of gain settings {"lna": 6, "rxvga1": 25}
bandwidth_hz: Optional bandwidth (defaults to sample_rate * 0.8) bandwidth_hz: Optional bandwidth (defaults to sample_rate * 0.8)
progress_callback: Optional callback for progress updates progress_callback: Optional callback for progress updates
fill_gaps: Zero-fill samples dropped on overflow so the file timeline
stays exact. If False, gaps are only reported.
Returns: Returns:
CaptureResult with file info and statistics. CaptureResult with file info and statistics.
@@ -227,67 +413,47 @@ def capture_iq_sync(
Raises: Raises:
RuntimeError: If capture fails. RuntimeError: If capture fails.
""" """
device.ensure_connected() state = _claim_capture()
state = _get_capture_state()
if state["active"]:
raise RuntimeError("A capture is already in progress")
state["active"] = True
state["cancelled"] = False
try: try:
rf = _configure_rx(frequency_hz, sample_rate_hz, bandwidth_hz, gains) return _capture_iq(
total_samples = int(duration_seconds * rf.sample_rate_hz) state, frequency_hz, sample_rate_hz, duration_seconds, output_path,
soapy_format = "CF32" if format == "cf32" else "CS16" format, gains, bandwidth_hz, fill_gaps, progress_callback,
bytes_per_sample = 8 if format == "cf32" else 4
output_path = Path(output_path)
output_path.parent.mkdir(parents=True, exist_ok=True)
stats = _StreamStats()
bytes_written = 0
start_time = time.monotonic()
chunks = _rx_chunks(total_samples, soapy_format, stats, lambda: state["cancelled"])
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,
elapsed_seconds=elapsed,
estimated_remaining_seconds=remaining,
bytes_written=bytes_written,
)
state["progress"] = progress
if progress_callback:
progress_callback(progress)
return CaptureResult(
file_path=str(output_path),
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: finally:
state["active"] = False _release_capture(state)
state["progress"] = None
state["cancelled"] = False
def start_capture(
frequency_hz: float,
sample_rate_hz: float,
duration_seconds: float,
output_path: str | Path,
format: SampleFormat = "cf32",
gains: dict[str, float] | None = None,
bandwidth_hz: float | None = None,
fill_gaps: bool = True,
) -> None:
"""Start an IQ capture on a background thread and return immediately.
Poll is_capture_active() / get_capture_progress(), then get_last_capture()
for the result. cancel_capture() stops it early.
"""
state = _claim_capture()
state["last_result"] = None
state["last_error"] = None
def run() -> None:
try:
state["last_result"] = _capture_iq(
state, frequency_hz, sample_rate_hz, duration_seconds, output_path,
format, gains, bandwidth_hz, fill_gaps, None,
)
except Exception as e:
state["last_error"] = str(e)
finally:
_release_capture(state)
threading.Thread(target=run, name="mcbladerf-capture", daemon=True).start()
async def capture_iq( async def capture_iq(
@@ -381,9 +547,26 @@ def compute_spectrum(
power_db=power_db.tolist(), power_db=power_db.tolist(),
peak_frequency_hz=float(frequencies[peak_idx]), peak_frequency_hz=float(frequencies[peak_idx]),
peak_power_db=float(power_db[peak_idx]), peak_power_db=float(power_db[peak_idx]),
noise_floor_db=float(np.median(power_db)),
) )
def decimate_max_hold(
frequencies_hz: list[float], power_db: list[float], max_points: int
) -> tuple[list[float], list[float]]:
"""Reduce a spectrum to at most max_points by keeping each group's maximum.
Max-hold keeps narrow carriers visible, which averaging would bury. Each
output frequency is its group's centre.
"""
if max_points <= 0 or len(power_db) <= max_points:
return frequencies_hz, power_db
groups = np.array_split(np.arange(len(power_db)), max_points)
f = np.asarray(frequencies_hz)
p = np.asarray(power_db)
return [float(f[g].mean()) for g in groups], [float(p[g].max()) for g in groups]
def capture_spectrum_sync( def capture_spectrum_sync(
frequency_hz: float, frequency_hz: float,
sample_rate_hz: float, sample_rate_hz: float,
@@ -413,8 +596,11 @@ def capture_spectrum_sync(
rf = _configure_rx(frequency_hz, sample_rate_hz, bandwidth_hz, gains) rf = _configure_rx(frequency_hz, sample_rate_hz, bandwidth_hz, gains)
# No zero-fill: zeros inside an FFT frame would distort it. A gap only
# makes frames non-contiguous, which averaging tolerates.
stats = _StreamStats() stats = _StreamStats()
with closing(_rx_chunks(fft_size * averaging, "CF32", stats)) as chunks: chunks = _rx_chunks(fft_size * averaging, rf.sample_rate_hz, "CF32", stats, fill_gaps=False)
with closing(chunks):
samples = np.concatenate([c.copy() for c in chunks]) samples = np.concatenate([c.copy() for c in chunks])
result = compute_spectrum( result = compute_spectrum(
@@ -427,6 +613,7 @@ def capture_spectrum_sync(
) )
result.gains_db = rf.gains result.gains_db = rf.gains
result.overflows = stats.overflows result.overflows = stats.overflows
result.samples_dropped = stats.dropped
result.clipped_samples = stats.clipped result.clipped_samples = stats.clipped
return result return result
+65 -14
View File
@@ -24,9 +24,16 @@ except ImportError:
SOAPY_SDR_RX = 0 SOAPY_SDR_RX = 0
SOAPY_SDR_TX = 1 SOAPY_SDR_TX = 1
# SoapySDR readStream error codes (SoapySDR/Errors.h) # SoapySDR readStream error codes (SoapySDR/Errors.h) and flags (SoapySDR/Constants.h)
SOAPY_SDR_TIMEOUT = -1 SOAPY_SDR_TIMEOUT = -1
SOAPY_SDR_OVERFLOW = -4 SOAPY_SDR_OVERFLOW = -4
SOAPY_SDR_HAS_TIME = 1 << 2
# SoapyBladeRF stream args. The default 4096-sample buffers mean ~7300 Python
# readStream calls per second at 30 Msps, and any host stall over ~4 ms
# overflows. 128 x 64K samples buffers ~280 ms at 30 Msps.
STREAM_BUFLEN = 65536
RX_STREAM_ARGS = {"buflen": str(STREAM_BUFLEN), "buffers": "128", "transfers": "32"}
# bladeRF SC16 Q11: CS16 samples are raw integers, CF32 is CS16 / 2048 # bladeRF SC16 Q11: CS16 samples are raw integers, CF32 is CS16 / 2048
CS16_FULL_SCALE = 2048.0 CS16_FULL_SCALE = 2048.0
@@ -333,8 +340,24 @@ def set_frequency(direction: DirectionType, frequency_hz: float, channel: int =
Actual frequency set (may differ due to PLL resolution). Actual frequency set (may differ due to PLL resolution).
""" """
device = get_device() device = get_device()
device.setFrequency(direction_to_soapy(direction), channel, frequency_hz) soapy_dir = direction_to_soapy(direction)
return device.getFrequency(direction_to_soapy(direction), channel) check_range("frequency", frequency_hz, device.getFrequencyRange(soapy_dir, channel), "Hz")
device.setFrequency(soapy_dir, channel, frequency_hz)
return device.getFrequency(soapy_dir, channel)
def check_range(name: str, value: float, ranges: Any, unit: str) -> None:
"""Raise if value is outside a SoapySDR range or RangeList.
The plugin clamps out-of-range values silently, so asking for rxvga1=99
would quietly give 30.
"""
span = range_span([ranges] if hasattr(ranges, "minimum") else ranges)
if span and not span["min_hz"] <= value <= span["max_hz"]:
raise RuntimeError(
f"{name} {value:g} {unit} is outside the supported range "
f"{span['min_hz']:g} to {span['max_hz']:g} {unit}"
)
def get_sample_rate(direction: DirectionType, channel: int = 0) -> float: def get_sample_rate(direction: DirectionType, channel: int = 0) -> float:
@@ -363,8 +386,10 @@ def set_sample_rate(direction: DirectionType, sample_rate_hz: float, channel: in
Actual sample rate set. Actual sample rate set.
""" """
device = get_device() device = get_device()
device.setSampleRate(direction_to_soapy(direction), channel, sample_rate_hz) soapy_dir = direction_to_soapy(direction)
return device.getSampleRate(direction_to_soapy(direction), channel) check_range("sample rate", sample_rate_hz, device.getSampleRateRange(soapy_dir, channel), "Hz")
device.setSampleRate(soapy_dir, channel, sample_rate_hz)
return device.getSampleRate(soapy_dir, channel)
def get_bandwidth(direction: DirectionType, channel: int = 0) -> float: def get_bandwidth(direction: DirectionType, channel: int = 0) -> float:
@@ -393,8 +418,19 @@ def set_bandwidth(direction: DirectionType, bandwidth_hz: float, channel: int =
Actual bandwidth set. Actual bandwidth set.
""" """
device = get_device() device = get_device()
device.setBandwidth(direction_to_soapy(direction), channel, bandwidth_hz) soapy_dir = direction_to_soapy(direction)
return device.getBandwidth(direction_to_soapy(direction), channel) check_range("bandwidth", bandwidth_hz, device.getBandwidthRange(soapy_dir, channel), "Hz")
device.setBandwidth(soapy_dir, channel, bandwidth_hz)
return device.getBandwidth(soapy_dir, channel)
def default_bandwidth(direction: DirectionType, sample_rate_hz: float, channel: int = 0) -> float:
"""80% of the sample rate, clamped to the filter range (1.5 to 28 MHz on bladeRF1)."""
span = range_span(get_device().getBandwidthRange(direction_to_soapy(direction), channel))
bw = sample_rate_hz * 0.8
if span:
bw = min(max(bw, span["min_hz"]), span["max_hz"])
return bw
def list_gains(direction: DirectionType, channel: int = 0) -> list[dict[str, Any]]: def list_gains(direction: DirectionType, channel: int = 0) -> list[dict[str, Any]]:
@@ -453,6 +489,10 @@ def set_gain(direction: DirectionType, stage: str, gain_db: float, channel: int
""" """
device = get_device() device = get_device()
soapy_dir = direction_to_soapy(direction) soapy_dir = direction_to_soapy(direction)
stages = list(device.listGains(soapy_dir, channel))
if stage not in stages:
raise RuntimeError(f"Unknown {direction} gain stage {stage!r}. Stages: {stages}")
check_range(f"{stage} gain", gain_db, device.getGainRange(soapy_dir, channel, stage), "dB")
device.setGain(soapy_dir, channel, stage, gain_db) device.setGain(soapy_dir, channel, stage, gain_db)
return device.getGain(soapy_dir, channel, stage) return device.getGain(soapy_dir, channel, stage)
@@ -536,7 +576,7 @@ def setup_rx_stream(format_str: str = "CF32", channel: int = 0) -> Any:
except Exception: except Exception:
pass pass
stream = device.setupStream(SOAPY_SDR_RX, format_str, [channel]) stream = device.setupStream(SOAPY_SDR_RX, format_str, [channel], RX_STREAM_ARGS)
state.active_rx_stream = stream state.active_rx_stream = stream
return stream return stream
@@ -573,7 +613,16 @@ def alloc_buffer(num_samples: int, format_str: str) -> np.ndarray:
return np.zeros(num_samples, dtype=np.complex64) return np.zeros(num_samples, dtype=np.complex64)
def read_stream(buff: np.ndarray, num_samples: int, timeout_us: int = 100000) -> tuple[int, bool]: @dataclass
class ReadResult:
"""One readStream call. time_ns is the hardware timestamp of the first sample."""
count: int
overflow: bool = False
time_ns: int | None = None
def read_stream(buff: np.ndarray, num_samples: int, timeout_us: int = 200000) -> ReadResult:
"""Read samples from the active RX stream into a preallocated buffer. """Read samples from the active RX stream into a preallocated buffer.
Args: Args:
@@ -582,8 +631,9 @@ def read_stream(buff: np.ndarray, num_samples: int, timeout_us: int = 100000) ->
timeout_us: Timeout in microseconds timeout_us: Timeout in microseconds
Returns: Returns:
Tuple of (samples read, overflow). Timeouts and overflows return 0 ReadResult. Timeouts and overflows return 0 samples rather than
samples rather than raising; an overflow means samples were dropped. raising; an overflow means the driver dropped samples, and the next
read's timestamp shows how many.
""" """
state = get_state() state = get_state()
device = get_device() device = get_device()
@@ -594,13 +644,14 @@ def read_stream(buff: np.ndarray, num_samples: int, timeout_us: int = 100000) ->
sr = device.readStream(state.active_rx_stream, [buff], num_samples, timeoutUs=timeout_us) sr = device.readStream(state.active_rx_stream, [buff], num_samples, timeoutUs=timeout_us)
if sr.ret == SOAPY_SDR_TIMEOUT: if sr.ret == SOAPY_SDR_TIMEOUT:
return 0, False return ReadResult(0)
if sr.ret == SOAPY_SDR_OVERFLOW: if sr.ret == SOAPY_SDR_OVERFLOW:
return 0, True return ReadResult(0, overflow=True)
if sr.ret < 0: if sr.ret < 0:
raise RuntimeError(f"Stream read error: {SoapySDR.errToStr(sr.ret)} ({sr.ret})") raise RuntimeError(f"Stream read error: {SoapySDR.errToStr(sr.ret)} ({sr.ret})")
return sr.ret, False time_ns = sr.timeNs if sr.flags & SOAPY_SDR_HAS_TIME else None
return ReadResult(sr.ret, time_ns=time_ns)
def close_rx_stream() -> None: def close_rx_stream() -> None:
+152 -76
View File
@@ -22,6 +22,44 @@ mcp = FastMCP(
) )
def _capture_busy() -> dict | None:
"""Error response if a background capture owns the device."""
if capture.is_capture_active():
return {
"status": "error",
"message": "A capture is running. Wait for it (bladerf_stream_status) "
"or stop it (bladerf_stop_capture) first.",
}
return None
def _capture_result_dict(result: capture.CaptureResult) -> dict:
warning = result.warning
return {
"status": "completed_with_gaps" if warning else "completed",
"warning": warning,
"file_path": result.file_path,
"sample_count": result.sample_count,
"duration_seconds": result.duration_seconds,
"sample_rate_hz": result.sample_rate_hz,
"frequency_hz": result.frequency_hz,
"format": result.format,
"file_size_bytes": result.file_size_bytes,
"gains_db": result.gains_db,
"overflows": result.overflows,
"samples_dropped": result.samples_dropped,
"zero_filled": result.zero_filled,
"gap_count": result.gap_count,
"gaps": [
{"offset_samples": g.offset_samples, "length_samples": g.length_samples}
for g in result.gaps
],
"clipped_samples": result.clipped_samples,
"peak_dbfs": result.peak_dbfs,
"cancelled": result.cancelled,
}
def _with_cli(fn: Callable[..., Any], serial: str | None, *args: Any) -> Any: def _with_cli(fn: Callable[..., Any], serial: str | None, *args: Any) -> Any:
"""Run a bladeRF-cli helper against the device. """Run a bladeRF-cli helper against the device.
@@ -60,15 +98,13 @@ def bladerf_status() -> dict:
rx_config = device.get_rf_config("rx") rx_config = device.get_rf_config("rx")
tx_config = device.get_rf_config("tx") tx_config = device.get_rf_config("tx")
# SoapySDR does not expose USB speed; bladerf_get_hardware_info has it
known = {k: v for k, v in hw_info.items() if v != "unknown"}
return { return {
"connected": True, "connected": True,
"serial": hw_info.get("serial"), **known,
"product": hw_info.get("product"), "capture_active": capture.is_capture_active(),
"vendor": hw_info.get("vendor"),
"fw_version": hw_info.get("fw_version"),
"fpga_version": hw_info.get("fpga_version"),
"fpga_size": hw_info.get("fpga_size"),
"usb_speed": hw_info.get("usb_speed"),
"rx_config": { "rx_config": {
"frequency_hz": rx_config.frequency_hz, "frequency_hz": rx_config.frequency_hz,
"sample_rate_hz": rx_config.sample_rate_hz, "sample_rate_hz": rx_config.sample_rate_hz,
@@ -129,6 +165,8 @@ def bladerf_connect(
Returns: Returns:
Dict with connection status and device info. Dict with connection status and device info.
""" """
if busy := _capture_busy():
return busy
try: try:
info = device.connect(serial=serial, device_index=device_index) info = device.connect(serial=serial, device_index=device_index)
return { return {
@@ -150,6 +188,8 @@ def bladerf_disconnect() -> dict:
Returns: Returns:
Dict with disconnection status. Dict with disconnection status.
""" """
if busy := _capture_busy():
return busy
was_connected = device.is_connected() was_connected = device.is_connected()
device.disconnect() device.disconnect()
return { return {
@@ -202,6 +242,8 @@ def bladerf_set_frequency(
Returns: Returns:
Dict with actual frequency set (may differ due to PLL resolution). Dict with actual frequency set (may differ due to PLL resolution).
""" """
if busy := _capture_busy():
return busy
try: try:
actual = device.set_frequency(direction, frequency_hz, channel) actual = device.set_frequency(direction, frequency_hz, channel)
return { return {
@@ -253,6 +295,8 @@ def bladerf_set_sample_rate(
Returns: Returns:
Dict with actual sample rate set. Dict with actual sample rate set.
""" """
if busy := _capture_busy():
return busy
try: try:
actual = device.set_sample_rate(direction, sample_rate_hz, channel) actual = device.set_sample_rate(direction, sample_rate_hz, channel)
return { return {
@@ -305,6 +349,8 @@ def bladerf_set_bandwidth(
Returns: Returns:
Dict with actual bandwidth set. Dict with actual bandwidth set.
""" """
if busy := _capture_busy():
return busy
try: try:
actual = device.set_bandwidth(direction, bandwidth_hz, channel) actual = device.set_bandwidth(direction, bandwidth_hz, channel)
return { return {
@@ -387,6 +433,8 @@ def bladerf_set_gain(
Returns: Returns:
Dict with actual gain set. Dict with actual gain set.
""" """
if busy := _capture_busy():
return busy
try: try:
actual = device.set_gain(direction, stage, gain_db, channel) actual = device.set_gain(direction, stage, gain_db, channel)
return { return {
@@ -439,6 +487,15 @@ def bladerf_get_sample_rate_range(
# ============================================================================= # =============================================================================
def _gains(lna: float | None, rxvga1: float | None, rxvga2: float | None) -> dict | None:
gains = {
stage: value
for stage, value in (("lna", lna), ("rxvga1", rxvga1), ("rxvga2", rxvga2))
if value is not None
}
return gains or None
@mcp.tool() @mcp.tool()
def bladerf_capture_iq( def bladerf_capture_iq(
frequency_hz: Annotated[float, "Center frequency in Hz"], frequency_hz: Annotated[float, "Center frequency in Hz"],
@@ -449,7 +506,9 @@ def bladerf_capture_iq(
lna_gain_db: Annotated[float | None, "LNA gain in dB (0, 3, or 6)"] = None, lna_gain_db: Annotated[float | None, "LNA gain in dB (0, 3, or 6)"] = None,
rxvga1_gain_db: Annotated[float | None, "RXVGA1 gain in dB (5-30)"] = None, rxvga1_gain_db: Annotated[float | None, "RXVGA1 gain in dB (5-30)"] = None,
rxvga2_gain_db: Annotated[float | None, "RXVGA2 gain in dB (0-30)"] = None, rxvga2_gain_db: Annotated[float | None, "RXVGA2 gain in dB (0-30)"] = None,
bandwidth_hz: Annotated[float | None, "Analog bandwidth (defaults to 0.8 * sample_rate)"] = None, bandwidth_hz: Annotated[float | None, "Analog bandwidth (defaults to 0.8 * sample_rate, max 28 MHz)"] = None,
background: Annotated[bool, "Return immediately; poll bladerf_stream_status for the result"] = False,
fill_gaps: Annotated[bool, "Zero-fill samples dropped on overflow to keep the timeline exact"] = True,
) -> dict: ) -> dict:
"""Capture IQ samples to a file. """Capture IQ samples to a file.
@@ -457,10 +516,23 @@ def bladerf_capture_iq(
float32 (CF32, 8 bytes/sample, scaled to +/-1.0) or complex int16 (CS16, float32 (CF32, 8 bytes/sample, scaled to +/-1.0) or complex int16 (CS16,
4 bytes/sample, raw SC16 Q11 with full scale 2048). 4 bytes/sample, raw SC16 Q11 with full scale 2048).
If the host falls behind, the driver drops samples. Each gap is measured
from the hardware timestamps and, by default, zero-filled so sample N is
always N / sample_rate seconds after the first. Status is then
"completed_with_gaps" with a warning, samples_dropped and a gap list
(offsets in output samples). With fill_gaps=False the file is shorter
than the requested duration instead.
Files are large (30 Msps CS16 is 120 MB/s). Prefer a path on disk over
/tmp, which is often a RAM-backed tmpfs.
Use background=True for anything longer than a few seconds: the call
returns at once, bladerf_stream_status reports progress and then the
result, and bladerf_stop_capture can end it early.
Gains left as None keep their current device value, which may come from 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 an earlier call or another process; gains_db in the result shows what was
used. Check overflows (dropped-sample events) and clipped_samples before used. Out-of-range values are rejected rather than clamped.
trusting the data.
Args: Args:
frequency_hz: Center frequency frequency_hz: Center frequency
@@ -472,45 +544,31 @@ def bladerf_capture_iq(
rxvga1_gain_db: RXVGA1 gain (5-30 dB) rxvga1_gain_db: RXVGA1 gain (5-30 dB)
rxvga2_gain_db: RXVGA2 gain (0-30 dB) rxvga2_gain_db: RXVGA2 gain (0-30 dB)
bandwidth_hz: Analog bandwidth bandwidth_hz: Analog bandwidth
background: Run on a background thread and return immediately
fill_gaps: Zero-fill dropped samples
Returns: Returns:
Dict with file path, sample count, duration, and file size. Dict with file path, sample count, gap accounting, and signal stats.
""" """
# Build gains dict from individual parameters kwargs = {
gains = {} "frequency_hz": frequency_hz,
if lna_gain_db is not None: "sample_rate_hz": sample_rate_hz,
gains["lna"] = lna_gain_db "duration_seconds": duration_seconds,
if rxvga1_gain_db is not None: "output_path": output_path,
gains["rxvga1"] = rxvga1_gain_db "format": format,
if rxvga2_gain_db is not None: "gains": _gains(lna_gain_db, rxvga1_gain_db, rxvga2_gain_db),
gains["rxvga2"] = rxvga2_gain_db "bandwidth_hz": bandwidth_hz,
"fill_gaps": fill_gaps,
}
try: try:
result = capture.capture_iq_sync( if background:
frequency_hz=frequency_hz, capture.start_capture(**kwargs)
sample_rate_hz=sample_rate_hz, return {
duration_seconds=duration_seconds, "status": "started",
output_path=output_path, "message": "Capture running. Poll bladerf_stream_status for progress and the result.",
format=format, }
gains=gains if gains else None, return _capture_result_dict(capture.capture_iq_sync(**kwargs))
bandwidth_hz=bandwidth_hz, except (RuntimeError, OSError) as e:
)
return {
"status": "completed",
"file_path": result.file_path,
"sample_count": result.sample_count,
"duration_seconds": result.duration_seconds,
"sample_rate_hz": result.sample_rate_hz,
"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)} return {"status": "error", "message": str(e)}
@@ -524,68 +582,78 @@ def bladerf_capture_spectrum(
rxvga1_gain_db: Annotated[float | None, "RXVGA1 gain in dB"] = None, rxvga1_gain_db: Annotated[float | None, "RXVGA1 gain in dB"] = None,
rxvga2_gain_db: Annotated[float | None, "RXVGA2 gain in dB"] = None, rxvga2_gain_db: Annotated[float | None, "RXVGA2 gain in dB"] = None,
window: Annotated[Literal["hann", "blackman", "hamming", "none"], "Window function"] = "hann", window: Annotated[Literal["hann", "blackman", "hamming", "none"], "Window function"] = "hann",
max_points: Annotated[int, "Max points returned (max-hold decimation); 0 returns no arrays"] = 256,
) -> dict: ) -> dict:
"""Capture and compute power spectrum (FFT). """Capture and compute power spectrum (FFT).
Captures IQ samples and computes an averaged power spectrum. Captures IQ samples and computes an averaged power spectrum in dBFS
Returns frequency bins and power levels in dBFS (0 dBFS is a full-scale (0 dBFS is a full-scale tone). Peak and noise floor (median bin) are
tone). Gains left as None keep their current device value. A nonzero computed at full FFT resolution. The returned arrays are reduced to
max_points by keeping each group's strongest bin, so narrow carriers stay
visible; use max_points=0 for just the summary. Large arrays can exceed
what the client will show.
Gains left as None keep their current device value. A nonzero
clipped_samples means the ADC saturated; lower the gain. clipped_samples means the ADC saturated; lower the gain.
Args: Args:
frequency_hz: Center frequency frequency_hz: Center frequency
sample_rate_hz: Sample rate (determines visible bandwidth) sample_rate_hz: Sample rate (determines visible bandwidth)
fft_size: FFT size (more bins = finer frequency resolution) fft_size: FFT size (more bins = finer frequency resolution)
averaging: Number of FFTs to average (reduces noise floor) averaging: Number of FFTs to average (reduces noise floor variance)
lna_gain_db: LNA gain lna_gain_db: LNA gain
rxvga1_gain_db: RXVGA1 gain rxvga1_gain_db: RXVGA1 gain
rxvga2_gain_db: RXVGA2 gain rxvga2_gain_db: RXVGA2 gain
window: Window function for spectral leakage reduction window: Window function for spectral leakage reduction
max_points: Maximum number of points in the returned arrays
Returns: Returns:
Dict with frequency bins, power levels, and peak detection. Dict with peak, noise floor, and a decimated spectrum.
""" """
gains = {}
if lna_gain_db is not None:
gains["lna"] = lna_gain_db
if rxvga1_gain_db is not None:
gains["rxvga1"] = rxvga1_gain_db
if rxvga2_gain_db is not None:
gains["rxvga2"] = rxvga2_gain_db
try: try:
result = capture.capture_spectrum_sync( result = capture.capture_spectrum_sync(
frequency_hz=frequency_hz, frequency_hz=frequency_hz,
sample_rate_hz=sample_rate_hz, sample_rate_hz=sample_rate_hz,
fft_size=fft_size, fft_size=fft_size,
averaging=averaging, averaging=averaging,
gains=gains if gains else None, gains=_gains(lna_gain_db, rxvga1_gain_db, rxvga2_gain_db),
window=window, window=window,
) )
return { except (RuntimeError, OSError) as e:
"status": "completed",
"center_frequency_hz": result.center_frequency_hz,
"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)} return {"status": "error", "message": str(e)}
response = {
"status": "completed",
"center_frequency_hz": result.center_frequency_hz,
"bandwidth_hz": result.bandwidth_hz,
"bin_count": result.bin_count,
"resolution_hz": result.bandwidth_hz / result.bin_count,
"power_units": "dBFS",
"peak_frequency_hz": result.peak_frequency_hz,
"peak_power_db": round(result.peak_power_db, 2),
"noise_floor_db": round(result.noise_floor_db, 2),
"gains_db": result.gains_db,
"overflows": result.overflows,
"samples_dropped": result.samples_dropped,
"clipped_samples": result.clipped_samples,
}
if max_points > 0:
freqs, power = capture.decimate_max_hold(
result.frequencies_hz, result.power_db, max_points
)
response["points"] = len(power)
response["frequencies_hz"] = [round(f) for f in freqs]
response["power_db"] = [round(p, 1) for p in power]
return response
@mcp.tool() @mcp.tool()
def bladerf_stream_status() -> dict: def bladerf_stream_status() -> dict:
"""Check if a capture is in progress. """Check capture progress, or get the result of the last background capture.
Returns: Returns:
Dict with capture status and progress if active. While running: progress. After a background capture: its full result
(same shape as bladerf_capture_iq) or its error.
""" """
if capture.is_capture_active(): if capture.is_capture_active():
progress = capture.get_capture_progress() progress = capture.get_capture_progress()
@@ -600,6 +668,12 @@ def bladerf_stream_status() -> dict:
"bytes_written": progress.bytes_written, "bytes_written": progress.bytes_written,
} }
return {"active": True} return {"active": True}
result, error = capture.get_last_capture()
if error:
return {"active": False, "last_capture": {"status": "error", "message": error}}
if result:
return {"active": False, "last_capture": _capture_result_dict(result)}
return {"active": False} return {"active": False}
@@ -749,6 +823,8 @@ def bladerf_set_loopback(
Returns: Returns:
Dict with status. Dict with status.
""" """
if busy := _capture_busy():
return busy
try: try:
state = device.get_state() state = device.get_state()
if state.device is not None and serial in (None, state.serial): if state.device is not None and serial in (None, state.serial):
+30 -11
View File
@@ -34,17 +34,6 @@ def test_cs16_file_is_raw_q11(tmp_path):
np.testing.assert_allclose(x, [1.0 - 0.5j, 0.25j]) 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: class _Range:
def __init__(self, lo, hi): def __init__(self, lo, hi):
self.lo, self.hi = 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 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(): def test_alloc_buffer_matches_stream_format():
assert device.alloc_buffer(10, "CS16").dtype == np.int16 assert device.alloc_buffer(10, "CS16").dtype == np.int16
assert device.alloc_buffer(10, "CS16").size == 20 assert device.alloc_buffer(10, "CS16").size == 20
+124
View File
@@ -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