Feature soapy stream buffering

src/CMakeLists.txt

@@ -60,6 +60,7 @@ add_subdirectory(lib)
 add_subdirectory(tools)
 
 if(ENABLE_TESTS)
+    enable_testing()
     add_subdirectory(tests)
     add_subdirectory(utests)
 endif(ENABLE_TESTS)

src/soapysdr/CMakeLists.txt

@@ -17,14 +17,24 @@ SOAPY_SDR_MODULE_UTIL(
     DESTINATION
         ${CMAKE_INSTALL_LIBDIR}/SoapySDR/modules${SOAPY_SDR_ABI_VERSION}/
     SOURCES
-        usdr_soapy.cpp usdr_soapy_reg.cpp
+        usdr_soapy.cpp usdr_soapy_reg.cpp rx_packet_buffer.cpp
     LIBRARIES
         usdr
 )
 
 if(ENABLE_TESTS)
     add_executable(test_usdr_soapy tests/test_usdr_soapy.c)
     target_link_libraries(test_usdr_soapy ${SoapySDR_LIBRARIES})
+    set_target_properties(test_usdr_soapy PROPERTIES
+        RUNTIME_OUTPUT_DIRECTORY ${CMAKE_CURRENT_BINARY_DIR}/tests
+    )
+
+    add_executable(test_rx_packet_buffer tests/test_rx_packet_buffer.cpp rx_packet_buffer.cpp)
+    target_link_libraries(test_rx_packet_buffer usdr)
+    set_target_properties(test_rx_packet_buffer PROPERTIES
+        RUNTIME_OUTPUT_DIRECTORY ${CMAKE_CURRENT_BINARY_DIR}/tests
+    )
+    add_test(NAME rx_packet_buffer COMMAND $<TARGET_FILE:test_rx_packet_buffer>)
 endif(ENABLE_TESTS)
 
 option(ENABLE_SOAPY_HIL_TESTS "Enable SoapySDR hardware-in-the-loop tests" OFF)

src/soapysdr/README.md

@@ -0,0 +1,72 @@
+# SoapySDR USDR support
+
+This directory contains the `usdr` SoapySDR module.
+
+## Device discovery
+
+Use the `driver=usdr` filter when listing USDR devices. Without the driver key,
+SoapySDR asks every installed module to enumerate, and unrelated modules such
+as audio devices may appear in the output.
+
+USB devices can be selected by their topology path, for example
+`bus=usb@3/1/6`: `3` is the USB bus, `1` is the downstream port, and `6` is the
+device address. This is useful when several USDR boards are connected to stable
+USB ports. It is not a serial number and may change if the board is moved to
+another port or hub, and the device/address part may also change after
+unplugging and plugging the board back into the same port.
+
+```sh
+SoapySDRUtil --find="driver=usdr"
+SoapySDRUtil --find="driver=usdr,bus=usb@3/1/6"
+SoapySDRUtil --find="driver=usdr,bus=pci,device=usdr0"
+```
+
+## Soapy Parameters
+
+Device arguments are passed when enumerating or opening the device, for example
+`SoapySDRUtil --find="driver=usdr,bus=usb@3/1/6"` or
+`SoapySDR.Device({"driver": "usdr", "bus": "usb@3/1/6"})`.
+
+| Argument | Scope | Description |
+| --- | --- | --- |
+| `driver=usdr` | discovery/open | Selects the USDR Soapy module. Use this to avoid unrelated devices from other Soapy modules. |
+| `bus=<path>` | discovery/open | Selects a USB or PCI bus path, for example `usb@3/1/6` or `pci`. |
+| `device=<name>` | discovery/open | Selects a lower-level device name, for example `usdr0` with `bus=pci`. |
+| `dev=<kwargs>` | open | Packed lower-level device string. Explicit `bus`, `device`, `fe`, `extclk`, and `extref` arguments override values from `dev`. |
+| `fe=<name>` | discovery/open | Selects a frontend when supported by the lower level. |
+| `extclk=<value>` | discovery/open | Passes external clock selection to the lower level. |
+| `extref=<value>` | discovery/open | Passes external reference selection to the lower level. |
+| `rxGapFill=none` | open | Default RX timestamp gap fill mode for streams created by this device. |
+| `rxGapFill=zero` | open | Default RX timestamp gap fill mode that fills timestamp gaps with zero samples. |
+
+Stream arguments are passed to `setupStream()`.
+
+| Argument | Direction | Default | Description |
+| --- | --- | --- | --- |
+| `bufferLength=<samples>` | RX/TX | automatic | Hardware packet size over the link. RX values must be either `0`/automatic or in the supported range checked by the driver. |
+| `linkFormat=CS16` | RX/TX | `CS16` | Complex int16 link format. TX currently supports `CS16` only. |
+| `linkFormat=CS12` | RX | `CS16` | Complex int12 link format for RX when the user stream format is `CF32`. |
+| `floatScale=1.0` | RX/TX | `1.0` | Stream float scaling. Values other than `1.0` are currently rejected. |
+| `rxGapFill=none` | RX | device default | Keeps only real samples and exposes packet loss as a timestamp jump. |
+| `rxGapFill=zero` | RX | device default | Keeps only real samples internally, but fills timestamp gaps with zero samples when data is returned from `readStream()`. |
+
+Advanced and debug open arguments are intended for development and diagnostics:
+
+| Argument | Description |
+| --- | --- |
+| `loglevel=<n>` | Overrides USDR log level. The `SOAPY_USDR_LOGLEVEL` environment variable can also set the default on Linux. |
+| `calls=1` | Enables verbose Soapy call logging. |
+| `desired_rx_pkt=<samples>` | Overrides the default RX packet size used when `bufferLength` is not specified. |
+| `rx12bit=1` | Forces RX wire format to 12-bit mode. |
+| `rxdump=<file>` | Dumps received samples to a file for debugging. |
+| `txcorr=<value>` | Applies the existing TX correction/debug path. |
+| `refclk=<value>` | Recognized for reference clock selection; currently logs the request. |
+| `rx_bw=<hz>` / `tx_bw=<hz>` | Applies a bandwidth value at open time. Prefer standard `setBandwidth()` for normal applications. |
+
+`SOAPY_USDR_ARGS` may be used to override open arguments from the environment
+with a packed Soapy kwargs string.
+
+## Testing
+
+See [tests/README.md](tests/README.md) for Python and C hardware-in-the-loop
+smoke tests, stream buffer checks, and CTest integration.

src/soapysdr/rx_packet_buffer.cpp

@@ -0,0 +1,257 @@
+// Copyright (c) 2026 Wavelet Lab
+// SPDX-License-Identifier: MIT
+
+#include "rx_packet_buffer.h"
+
+#include <algorithm>
+#include <cstring>
+#include <stdexcept>
+
+RxPacketBuffer::RxPacketBuffer(pusdr_dms_t stream,
+                               unsigned channels,
+                               size_t samples_per_packet,
+                               size_t bytes_per_packet,
+                               GapFill gap_fill,
+                               RecvFunction recv_function)
+    : _stream(stream)
+    , _recv(recv_function ? recv_function : &usdr_dms_recv)
+    , _channels(channels)
+    , _samples_per_packet(samples_per_packet)
+    , _bytes_per_packet(bytes_per_packet)
+    , _bytes_per_sample(bytes_per_packet / samples_per_packet)
+    , _gap_fill(gap_fill)
+    , _capacity(0)
+    , _read_pos(0)
+    , _write_pos(0)
+    , _available(0)
+    , _next_output_time(0)
+    , _time_valid(false)
+{
+    if (_stream == nullptr || _channels == 0 || _samples_per_packet == 0 ||
+        _bytes_per_packet == 0 || _bytes_per_sample == 0 ||
+        (_bytes_per_packet % _samples_per_packet) != 0) {
+        throw std::invalid_argument("RxPacketBuffer: invalid stream packet configuration");
+    }
+
+    _packet_buffers.resize(_channels);
+    _packet_ptrs.resize(_channels);
+    for (unsigned i = 0; i < _channels; i++) {
+        _packet_buffers[i].resize(_bytes_per_packet);
+        _packet_ptrs[i] = _packet_buffers[i].data();
+    }
+
+    ensureCapacity(_bytes_per_packet * 16);
+}
+
+size_t RxPacketBuffer::bytesPerElems(size_t elems) const
+{
+    return elems * _bytes_per_sample;
+}
+
+size_t RxPacketBuffer::elemsPerBytes(size_t bytes) const
+{
+    return bytes / _bytes_per_sample;
+}
+
+void RxPacketBuffer::ensureCapacity(size_t requested_bytes)
+{
+    if (requested_bytes <= _capacity) {
+        return;
+    }
+
+    size_t new_capacity = std::max(_bytes_per_packet * 16, _capacity);
+    if (new_capacity == 0) {
+        new_capacity = _bytes_per_packet * 16;
+    }
+    while (new_capacity < requested_bytes) {
+        new_capacity *= 2;
+    }
+
+    std::vector<std::vector<unsigned char>> new_buffers(_channels);
+    for (unsigned ch = 0; ch < _channels; ch++) {
+        new_buffers[ch].resize(new_capacity);
+        if (_available == 0) {
+            continue;
+        }
+
+        const size_t first = std::min(_available, _capacity - _read_pos);
+        std::memcpy(new_buffers[ch].data(), _buffers[ch].data() + _read_pos, first);
+        if (first < _available) {
+            std::memcpy(new_buffers[ch].data() + first, _buffers[ch].data(), _available - first);
+        }
+    }
+
+    _buffers.swap(new_buffers);
+    _capacity = new_capacity;
+    _read_pos = 0;
+    _write_pos = _available;
+}
+
+void RxPacketBuffer::appendPacket(const usdr_dms_recv_nfo_t &nfo)
+{
+    const size_t reported_samples = (nfo.totsyms != 0) ? nfo.totsyms : _samples_per_packet;
+    const size_t packet_samples = std::min(reported_samples, _samples_per_packet);
+    const size_t packet_bytes = bytesPerElems(packet_samples);
+
+    if (!_time_valid) {
+        _next_output_time = nfo.fsymtime;
+        _time_valid = true;
+    }
+
+    if (!_segments.empty()) {
+        const Segment &last = _segments.back();
+        const dm_time_t expected_time = last.start_time + (dm_time_t)last.samples;
+        if (expected_time == nfo.fsymtime) {
+            _segments.back().samples += packet_samples;
+        } else if (_gap_fill == GAP_FILL_ZERO && nfo.fsymtime > expected_time) {
+            _segments.push_back({nfo.fsymtime, packet_samples});
+        } else {
+            dropBufferedData();
+            _next_output_time = nfo.fsymtime;
+            _time_valid = true;
+            _segments.push_back({nfo.fsymtime, packet_samples});
+        }
+    } else if (nfo.fsymtime < _next_output_time ||
+               (_gap_fill == GAP_FILL_NONE && nfo.fsymtime != _next_output_time)) {
+        dropBufferedData();
+        _next_output_time = nfo.fsymtime;
+        _time_valid = true;
+        _segments.push_back({nfo.fsymtime, packet_samples});
+    } else {
+        _segments.push_back({nfo.fsymtime, packet_samples});
+    }
+
+    ensureCapacity(_available + packet_bytes);
+    for (unsigned ch = 0; ch < _channels; ch++) {
+        const size_t first = std::min(packet_bytes, _capacity - _write_pos);
+        std::memcpy(_buffers[ch].data() + _write_pos, _packet_buffers[ch].data(), first);
+        if (first < packet_bytes) {
+            std::memcpy(_buffers[ch].data(), _packet_buffers[ch].data() + first, packet_bytes - first);
+        }
+    }
+
+    _write_pos = (_write_pos + packet_bytes) % _capacity;
+    _available += packet_bytes;
+}
+
+void RxPacketBuffer::readRealBytes(void * const *buffs, size_t dst_offset_bytes, size_t bytes)
+{
+    for (unsigned ch = 0; ch < _channels; ch++) {
+        unsigned char *dst = static_cast<unsigned char*>(buffs[ch]) + dst_offset_bytes;
+        const size_t first = std::min(bytes, _capacity - _read_pos);
+        std::memcpy(dst, _buffers[ch].data() + _read_pos, first);
+        if (first < bytes) {
+            std::memcpy(dst + first, _buffers[ch].data(), bytes - first);
+        }
+    }
+
+    _read_pos = (_read_pos + bytes) % _capacity;
+    _available -= bytes;
+    size_t samples = elemsPerBytes(bytes);
+    _next_output_time += (dm_time_t)samples;
+    while (samples != 0 && !_segments.empty()) {
+        Segment &segment = _segments.front();
+        if (samples < segment.samples) {
+            segment.start_time += (dm_time_t)samples;
+            segment.samples -= samples;
+            break;
+        }
+        samples -= segment.samples;
+        _segments.pop_front();
+    }
+}
+
+void RxPacketBuffer::writeZeros(void * const *buffs, size_t dst_offset_bytes, size_t bytes)
+{
+    for (unsigned ch = 0; ch < _channels; ch++) {
+        unsigned char *dst = static_cast<unsigned char*>(buffs[ch]) + dst_offset_bytes;
+        std::memset(dst, 0, bytes);
+    }
+    _next_output_time += (dm_time_t)elemsPerBytes(bytes);
+}
+
+size_t RxPacketBuffer::outputAvailableSamples() const
+{
+    if (!_time_valid || _segments.empty()) {
+        return 0;
+    }
+
+    dm_time_t cursor = _next_output_time;
+    size_t samples = 0;
+    for (const Segment &segment: _segments) {
+        if (segment.start_time > cursor) {
+            if (_gap_fill != GAP_FILL_ZERO) {
+                break;
+            }
+            samples += (size_t)(segment.start_time - cursor);
+            cursor = segment.start_time;
+        } else if (segment.start_time < cursor) {
+            return 0;
+        }
+
+        samples += segment.samples;
+        cursor += (dm_time_t)segment.samples;
+    }
+    return samples;
+}
+
+void RxPacketBuffer::dropBufferedData()
+{
+    _read_pos = 0;
+    _write_pos = 0;
+    _available = 0;
+    _segments.clear();
+}
+
+int RxPacketBuffer::read(void * const *buffs,
+                         size_t elems,
+                         long timeout_us,
+                         dm_time_t &sample_time,
+                         usdr_dms_recv_nfo_t &nfo)
+{
+    const size_t requested_bytes = bytesPerElems(elems);
+    ensureCapacity(requested_bytes + _bytes_per_packet);
+
+    while (outputAvailableSamples() < elems) {
+        const int res = _recv(_stream, _packet_ptrs.data(), timeout_us / 1000, &nfo);
+        if (res != 0) {
+            return res;
+        }
+        appendPacket(nfo);
+    }
+
+    sample_time = _next_output_time;
+    size_t copied = 0;
+    while (copied < requested_bytes) {
+        if (_segments.empty()) {
+            break;
+        }
+
+        const Segment &first = _segments.front();
+        if (first.start_time > _next_output_time) {
+            const size_t zero_samples = std::min((size_t)(first.start_time - _next_output_time),
+                                                 elemsPerBytes(requested_bytes - copied));
+            const size_t zero_bytes = bytesPerElems(zero_samples);
+            writeZeros(buffs, copied, zero_bytes);
+            copied += zero_bytes;
+            continue;
+        }
+
+        const size_t real_bytes = std::min(bytesPerElems(first.samples), requested_bytes - copied);
+        readRealBytes(buffs, copied, real_bytes);
+        copied += real_bytes;
+    }
+    return 0;
+}
+
+void RxPacketBuffer::reset()
+{
+    dropBufferedData();
+    _next_output_time = 0;
+    _time_valid = false;
+}
+
+bool RxPacketBuffer::empty() const
+{
+    return _segments.empty() && _available == 0;
+}