diff --git a/.gitignore b/.gitignore index d45ddae..e10cd94 100644 --- a/.gitignore +++ b/.gitignore @@ -6,3 +6,4 @@ __pycache__/ dist/ *.cf32 *.cs16 +captures/ diff --git a/README.md b/README.md index 71f21f4..77c8647 100644 --- a/README.md +++ b/README.md @@ -83,7 +83,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, 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 ### 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 [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] > 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 -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. diff --git a/src/mcbladerf/capture.py b/src/mcbladerf/capture.py index c18f472..6ecd268 100644 --- a/src/mcbladerf/capture.py +++ b/src/mcbladerf/capture.py @@ -7,6 +7,7 @@ and computing power spectra from captured data. from __future__ import annotations import asyncio +import threading import time from contextlib import closing from dataclasses import dataclass, field @@ -22,14 +23,24 @@ if TYPE_CHECKING: SampleFormat = Literal["cf32", "cs16"] -# Samples per readStream call (the bladeRF plugin returns at most 4096 per call) -_CHUNK = 65536 +# Samples per readStream call, matching the stream buffer length +_CHUNK = device.STREAM_BUFLEN # 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 +_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 @@ -45,10 +56,29 @@ class CaptureResult: file_size_bytes: int gains_db: dict[str, float] = field(default_factory=dict) 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 peak_dbfs: float | None = None 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 class SpectrumResult: @@ -61,8 +91,10 @@ class SpectrumResult: power_db: list[float] peak_frequency_hz: float peak_power_db: float + noise_floor_db: float = 0.0 gains_db: dict[str, float] = field(default_factory=dict) overflows: int = 0 + samples_dropped: int = 0 clipped_samples: int = 0 @@ -80,6 +112,10 @@ class CaptureProgress: @dataclass class _StreamStats: overflows: int = 0 + unmeasured_overflows: int = 0 + dropped: int = 0 + gaps: list[Gap] = field(default_factory=list) + gap_count: int = 0 clipped: int = 0 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: 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]: return device.get_state().capture_info @@ -106,6 +148,12 @@ def get_capture_progress() -> CaptureProgress | None: 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: """Request cancellation of the current capture. @@ -132,32 +180,53 @@ def _configure_rx( """ 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) + 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(): 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).""" +def _chunk_stats(chunk: np.ndarray, soapy_format: str) -> tuple[float, int]: + """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": - return np.abs(chunk.reshape(-1, 2).astype(np.float32)) / device.CS16_FULL_SCALE - return np.abs(chunk.view(np.float32).reshape(-1, 2)) + mag = np.abs(chunk) + 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( total_samples: int, + sample_rate_hz: float, soapy_format: str, stats: _StreamStats, + fill_gaps: bool = True, cancelled: Callable[[], bool] = lambda: False, ) -> Iterator[np.ndarray]: """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 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) + zeros = np.zeros_like(buff) per_sample = 2 if soapy_format == "CS16" else 1 device.setup_rx_stream(soapy_format) @@ -165,37 +234,151 @@ def _rx_chunks( device.activate_rx_stream() to_discard = _SETTLE_SAMPLES 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() while got < total_samples and not cancelled(): 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() if not to_discard: stats.overflows += 1 + pending_overflows += 1 continue - if n == 0: + if r.count == 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) + to_discard = max(0, to_discard - r.count) 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] - 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)) + peak, clipped = _chunk_stats(chunk, soapy_format) + stats.peak = max(stats.peak, peak) + stats.clipped += clipped got += n yield chunk finally: 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( frequency_hz: float, sample_rate_hz: float, @@ -205,6 +388,7 @@ def capture_iq_sync( gains: dict[str, float] | None = None, bandwidth_hz: float | None = None, progress_callback: Callable[[CaptureProgress], None] | None = None, + fill_gaps: bool = True, ) -> CaptureResult: """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} bandwidth_hz: Optional bandwidth (defaults to sample_rate * 0.8) 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: CaptureResult with file info and statistics. @@ -227,67 +413,47 @@ def capture_iq_sync( Raises: RuntimeError: If capture fails. """ - device.ensure_connected() - state = _get_capture_state() - - if state["active"]: - raise RuntimeError("A capture is already in progress") - - state["active"] = True - state["cancelled"] = False - + state = _claim_capture() try: - 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, 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"], + return _capture_iq( + state, frequency_hz, sample_rate_hz, duration_seconds, output_path, + format, gains, bandwidth_hz, fill_gaps, progress_callback, ) - finally: - state["active"] = False - state["progress"] = None - state["cancelled"] = False + _release_capture(state) + + +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( @@ -381,9 +547,26 @@ def compute_spectrum( power_db=power_db.tolist(), peak_frequency_hz=float(frequencies[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( frequency_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) + # No zero-fill: zeros inside an FFT frame would distort it. A gap only + # makes frames non-contiguous, which averaging tolerates. 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]) result = compute_spectrum( @@ -427,6 +613,7 @@ def capture_spectrum_sync( ) result.gains_db = rf.gains result.overflows = stats.overflows + result.samples_dropped = stats.dropped result.clipped_samples = stats.clipped return result diff --git a/src/mcbladerf/device.py b/src/mcbladerf/device.py index 5d27c3f..d240078 100644 --- a/src/mcbladerf/device.py +++ b/src/mcbladerf/device.py @@ -24,9 +24,16 @@ except ImportError: SOAPY_SDR_RX = 0 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_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 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). """ device = get_device() - device.setFrequency(direction_to_soapy(direction), channel, frequency_hz) - return device.getFrequency(direction_to_soapy(direction), channel) + soapy_dir = direction_to_soapy(direction) + 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: @@ -363,8 +386,10 @@ def set_sample_rate(direction: DirectionType, sample_rate_hz: float, channel: in Actual sample rate set. """ device = get_device() - device.setSampleRate(direction_to_soapy(direction), channel, sample_rate_hz) - return device.getSampleRate(direction_to_soapy(direction), channel) + soapy_dir = direction_to_soapy(direction) + 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: @@ -393,8 +418,19 @@ def set_bandwidth(direction: DirectionType, bandwidth_hz: float, channel: int = Actual bandwidth set. """ device = get_device() - device.setBandwidth(direction_to_soapy(direction), channel, bandwidth_hz) - return device.getBandwidth(direction_to_soapy(direction), channel) + soapy_dir = direction_to_soapy(direction) + 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]]: @@ -453,6 +489,10 @@ def set_gain(direction: DirectionType, stage: str, gain_db: float, channel: int """ device = get_device() 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) 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: 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 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) -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. Args: @@ -582,8 +631,9 @@ def read_stream(buff: np.ndarray, num_samples: int, timeout_us: int = 100000) -> timeout_us: Timeout in microseconds Returns: - Tuple of (samples read, overflow). Timeouts and overflows return 0 - samples rather than raising; an overflow means samples were dropped. + ReadResult. Timeouts and overflows return 0 samples rather than + raising; an overflow means the driver dropped samples, and the next + read's timestamp shows how many. """ state = get_state() 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) if sr.ret == SOAPY_SDR_TIMEOUT: - return 0, False + return ReadResult(0) if sr.ret == SOAPY_SDR_OVERFLOW: - return 0, True + return ReadResult(0, overflow=True) if sr.ret < 0: 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: diff --git a/src/mcbladerf/server.py b/src/mcbladerf/server.py index 78f459c..2213fdd 100644 --- a/src/mcbladerf/server.py +++ b/src/mcbladerf/server.py @@ -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: """Run a bladeRF-cli helper against the device. @@ -60,15 +98,13 @@ def bladerf_status() -> dict: rx_config = device.get_rf_config("rx") 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 { "connected": True, - "serial": hw_info.get("serial"), - "product": hw_info.get("product"), - "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"), + **known, + "capture_active": capture.is_capture_active(), "rx_config": { "frequency_hz": rx_config.frequency_hz, "sample_rate_hz": rx_config.sample_rate_hz, @@ -129,6 +165,8 @@ def bladerf_connect( Returns: Dict with connection status and device info. """ + if busy := _capture_busy(): + return busy try: info = device.connect(serial=serial, device_index=device_index) return { @@ -150,6 +188,8 @@ def bladerf_disconnect() -> dict: Returns: Dict with disconnection status. """ + if busy := _capture_busy(): + return busy was_connected = device.is_connected() device.disconnect() return { @@ -202,6 +242,8 @@ def bladerf_set_frequency( Returns: Dict with actual frequency set (may differ due to PLL resolution). """ + if busy := _capture_busy(): + return busy try: actual = device.set_frequency(direction, frequency_hz, channel) return { @@ -253,6 +295,8 @@ def bladerf_set_sample_rate( Returns: Dict with actual sample rate set. """ + if busy := _capture_busy(): + return busy try: actual = device.set_sample_rate(direction, sample_rate_hz, channel) return { @@ -305,6 +349,8 @@ def bladerf_set_bandwidth( Returns: Dict with actual bandwidth set. """ + if busy := _capture_busy(): + return busy try: actual = device.set_bandwidth(direction, bandwidth_hz, channel) return { @@ -387,6 +433,8 @@ def bladerf_set_gain( Returns: Dict with actual gain set. """ + if busy := _capture_busy(): + return busy try: actual = device.set_gain(direction, stage, gain_db, channel) 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() def bladerf_capture_iq( 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, 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, - 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: """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, 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 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. + used. Out-of-range values are rejected rather than clamped. Args: frequency_hz: Center frequency @@ -472,45 +544,31 @@ def bladerf_capture_iq( rxvga1_gain_db: RXVGA1 gain (5-30 dB) rxvga2_gain_db: RXVGA2 gain (0-30 dB) bandwidth_hz: Analog bandwidth + background: Run on a background thread and return immediately + fill_gaps: Zero-fill dropped samples 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 - 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 - + kwargs = { + "frequency_hz": frequency_hz, + "sample_rate_hz": sample_rate_hz, + "duration_seconds": duration_seconds, + "output_path": output_path, + "format": format, + "gains": _gains(lna_gain_db, rxvga1_gain_db, rxvga2_gain_db), + "bandwidth_hz": bandwidth_hz, + "fill_gaps": fill_gaps, + } try: - result = capture.capture_iq_sync( - frequency_hz=frequency_hz, - sample_rate_hz=sample_rate_hz, - duration_seconds=duration_seconds, - output_path=output_path, - format=format, - gains=gains if gains else None, - bandwidth_hz=bandwidth_hz, - ) - 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: + if background: + capture.start_capture(**kwargs) + return { + "status": "started", + "message": "Capture running. Poll bladerf_stream_status for progress and the result.", + } + return _capture_result_dict(capture.capture_iq_sync(**kwargs)) + except (RuntimeError, OSError) as 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, rxvga2_gain_db: Annotated[float | None, "RXVGA2 gain in dB"] = None, 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: """Capture and compute power spectrum (FFT). - Captures IQ samples and computes an averaged power spectrum. - 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 + Captures IQ samples and computes an averaged power spectrum in dBFS + (0 dBFS is a full-scale tone). Peak and noise floor (median bin) are + 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. Args: frequency_hz: Center frequency sample_rate_hz: Sample rate (determines visible bandwidth) 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 rxvga1_gain_db: RXVGA1 gain rxvga2_gain_db: RXVGA2 gain window: Window function for spectral leakage reduction + max_points: Maximum number of points in the returned arrays 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: result = capture.capture_spectrum_sync( frequency_hz=frequency_hz, sample_rate_hz=sample_rate_hz, fft_size=fft_size, averaging=averaging, - gains=gains if gains else None, + gains=_gains(lna_gain_db, rxvga1_gain_db, rxvga2_gain_db), window=window, ) - return { - "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: + except (RuntimeError, OSError) as 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() 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: - 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(): progress = capture.get_capture_progress() @@ -600,6 +668,12 @@ def bladerf_stream_status() -> dict: "bytes_written": progress.bytes_written, } 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} @@ -749,6 +823,8 @@ def bladerf_set_loopback( Returns: Dict with status. """ + if busy := _capture_busy(): + return busy try: state = device.get_state() if state.device is not None and serial in (None, state.serial): diff --git a/tests/test_capture.py b/tests/test_capture.py index 7ba3818..11de8bb 100644 --- a/tests/test_capture.py +++ b/tests/test_capture.py @@ -34,17 +34,6 @@ def test_cs16_file_is_raw_q11(tmp_path): 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 @@ -63,6 +52,36 @@ def test_range_span_covers_every_range(): assert device.range_span([]) is None +def test_chunk_stats_match_components(): + q11 = np.array([2047, -100, 50, -2048, 10, 10], dtype=np.int16) + cf32 = (q11[0::2] + 1j * q11[1::2]).astype(np.complex64) / device.CS16_FULL_SCALE + + for chunk, fmt in ((q11, "CS16"), (cf32, "CF32")): + peak, clipped = capture._chunk_stats(chunk, fmt) + assert peak == pytest.approx(1.0) + assert clipped == 2 + + +def test_decimate_max_hold_keeps_narrow_peak(): + f = list(range(4096)) + p = [-80.0] * 4096 + p[1234] = -20.0 + + df, dp = capture.decimate_max_hold(f, p, 256) + + assert len(dp) == 256 + assert max(dp) == -20.0 + assert capture.decimate_max_hold(f[:100], p[:100], 256) == (f[:100], p[:100]) + + +def test_check_range_rejects_instead_of_clamping(): + device.check_range("rxvga1 gain", 30, _Range(5, 30), "dB") + with pytest.raises(RuntimeError, match="outside the supported range 5 to 30 dB"): + device.check_range("rxvga1 gain", 99, _Range(5, 30), "dB") + with pytest.raises(RuntimeError, match="80000 to 4e\\+07"): + device.check_range("sample rate", 61.44e6, [_Range(80e3, 10e6), _Range(20e6, 40e6)], "Hz") + + def test_alloc_buffer_matches_stream_format(): assert device.alloc_buffer(10, "CS16").dtype == np.int16 assert device.alloc_buffer(10, "CS16").size == 20 diff --git a/tests/test_gaps.py b/tests/test_gaps.py new file mode 100644 index 0000000..7b529fe --- /dev/null +++ b/tests/test_gaps.py @@ -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