mirror of
https://github.com/gnss-sdr/gnss-sdr
synced 2026-11-01 20:24:30 +00:00
Fuse per-band frequency translation into EVK1029_Signal_Source
With enable_freq_xlating=true, Evk1029Source translates, filters and decimates each band itself and exposes one gr_complex output port per band, so Bypass Signal Conditioners can connect the channels straight to the source. Each band's FIR runs on its own worker thread, and the next chunk of the capture is decoded while the current one is filtered. Per-band filter design: band<n>_IF, _decimation_factor, _bw, _tw, _window, _attenuation_dB and _kaiser_beta. Taken from PR #1143 without its SignalConditioner Bypass commits (next already provides Bypass), plus: - Center shorter band filters on the longest one, so every band has the same group delay (mismatched tap counts biased pseudoranges between bands by up to km) - Fix source dump in fused mode: one gr_complex file per band (_ch<n>) - Reject unequal or non-positive band decimation factors - Build with GNU Radio < 3.9 (firdes window type) and VOLK < 2.5 (rotator), use volk_gnsssdr::vector - Link Gnuradio::filter privately and only with ENABLE_EVK1029 - Start worker threads after the capture file is opened
This commit is contained in:
7 files changed
+744
-102
No files matched your search
@@ -29,6 +29,16 @@ All notable changes to GNSS-SDR will be documented in this file.
|
||||
Signal Conditioner implementation connects the channels directly to the signal
|
||||
source. See the
|
||||
[Signal Conditioner documentation](https://gnss-sdr.org/docs/sp-blocks/signal-conditioner/#batching-and-buffering).
|
||||
- `EVK1029_Signal_Source` can now translate, filter, and decimate each band
|
||||
inside the source block (`enable_freq_xlating=true`), emitting one baseband
|
||||
`gr_complex` output per band and running each band's filter on its own worker
|
||||
thread, while decoding the next chunk of the capture in parallel. Combined
|
||||
with `Bypass` Signal Conditioners, this removes the per-band
|
||||
`Freq_Xlating_Fir_Filter` blocks. Each band is configured with `band<n>_IF`,
|
||||
`band<n>_decimation_factor` (the same for all bands), `band<n>_bw`,
|
||||
`band<n>_tw`, `band<n>_window`, `band<n>_attenuation_dB`, and
|
||||
`band<n>_kaiser_beta`. Bands with filters of different lengths are aligned to
|
||||
the same group delay, so they do not bias pseudoranges between bands.
|
||||
|
||||
### Improvements in Interoperability:
|
||||
|
||||
|
||||
@@ -224,6 +224,10 @@ target_link_libraries(signal_source_adapters
|
||||
Volk::volk
|
||||
)
|
||||
|
||||
if(ENABLE_EVK1029)
|
||||
target_link_libraries(signal_source_adapters PRIVATE Gnuradio::filter)
|
||||
endif()
|
||||
|
||||
if(ENABLE_GLOG_AND_GFLAGS)
|
||||
target_link_libraries(signal_source_adapters PRIVATE Glog::glog)
|
||||
target_compile_definitions(signal_source_adapters PRIVATE -DUSE_GLOG_AND_GFLAGS=1)
|
||||
|
||||
@@ -15,7 +15,9 @@
|
||||
|
||||
#include "evk1029_signal_source.h"
|
||||
#include "configuration_interface.h"
|
||||
#include <gnuradio/filter/firdes.h>
|
||||
#include <cmath>
|
||||
#include <stdexcept>
|
||||
#include <utility>
|
||||
|
||||
#if USE_GLOG_AND_GFLAGS
|
||||
@@ -26,6 +28,79 @@
|
||||
|
||||
using namespace std::string_literals;
|
||||
|
||||
namespace
|
||||
{
|
||||
// firdes takes gr::fft::window::win_type since GNU Radio 3.9, and its own
|
||||
// firdes::win_type before that.
|
||||
#if GNURADIO_USES_STD_POINTERS
|
||||
using Fir_Window = gr::fft::window::win_type;
|
||||
#else
|
||||
using Fir_Window = gr::filter::firdes::win_type;
|
||||
#endif
|
||||
|
||||
Fir_Window parse_window_type(const std::string& role, const std::string& name)
|
||||
{
|
||||
if (name == "hamming")
|
||||
{
|
||||
return Fir_Window::WIN_HAMMING;
|
||||
}
|
||||
if (name == "hann")
|
||||
{
|
||||
return Fir_Window::WIN_HANN;
|
||||
}
|
||||
if (name == "blackman")
|
||||
{
|
||||
return Fir_Window::WIN_BLACKMAN;
|
||||
}
|
||||
if (name == "blackman_harris")
|
||||
{
|
||||
return Fir_Window::WIN_BLACKMAN_HARRIS;
|
||||
}
|
||||
if (name == "rectangular")
|
||||
{
|
||||
return Fir_Window::WIN_RECTANGULAR;
|
||||
}
|
||||
if (name == "kaiser")
|
||||
{
|
||||
return Fir_Window::WIN_KAISER;
|
||||
}
|
||||
LOG(WARNING) << role << ": unknown window type '" << name << "', falling back to hamming";
|
||||
return Fir_Window::WIN_HAMMING;
|
||||
}
|
||||
|
||||
// Kaiser's own empirical formula (1974) relating a target stopband
|
||||
// attenuation to the window beta that actually achieves it. GNU Radio's
|
||||
// firdes::low_pass_2() does NOT derive beta from attenuation_dB itself --
|
||||
// attenuation_dB there only sizes the tap count, while beta independently
|
||||
// sets the real window shape -- so this block computes a consistent beta
|
||||
// unless the user overrides it explicitly.
|
||||
double kaiser_beta_from_attenuation(double attenuation_dB)
|
||||
{
|
||||
if (attenuation_dB > 50.0)
|
||||
{
|
||||
return 0.1102 * (attenuation_dB - 8.7);
|
||||
}
|
||||
if (attenuation_dB >= 21.0)
|
||||
{
|
||||
return 0.5842 * std::pow(attenuation_dB - 21.0, 0.4) + 0.07886 * (attenuation_dB - 21.0);
|
||||
}
|
||||
return 0.0;
|
||||
}
|
||||
|
||||
// "./evk1029_signal_source.dat" -> "./evk1029_signal_source_ch1.dat"
|
||||
std::string band_dump_filename(const std::string& filename, unsigned int band)
|
||||
{
|
||||
const std::string suffix = "_ch" + std::to_string(band);
|
||||
const auto dot = filename.find_last_of('.');
|
||||
const auto slash = filename.find_last_of("/\\");
|
||||
if (dot == std::string::npos || (slash != std::string::npos && dot < slash) || dot == slash + 1)
|
||||
{
|
||||
return filename + suffix;
|
||||
}
|
||||
return filename.substr(0, dot) + suffix + filename.substr(dot);
|
||||
}
|
||||
} // namespace
|
||||
|
||||
Evk1029SignalSource::Evk1029SignalSource(
|
||||
const ConfigurationInterface* configuration,
|
||||
const std::string& role,
|
||||
@@ -37,7 +112,8 @@ Evk1029SignalSource::Evk1029SignalSource(
|
||||
dump_(configuration->property(role + ".dump"s, false)),
|
||||
item_size_(sizeof(int8_t)),
|
||||
rf_channels_(static_cast<unsigned int>(getRfChannels())),
|
||||
enable_throttle_control_(configuration->property(role + ".enable_throttle_control"s, false))
|
||||
enable_throttle_control_(configuration->property(role + ".enable_throttle_control"s, false)),
|
||||
enable_freq_xlating_(configuration->property(role + ".enable_freq_xlating"s, false))
|
||||
{
|
||||
const std::string filename = configuration->property(role + ".filename"s, "data.bin"s);
|
||||
// NOTE: this must be the RAW (pre-decimation) ADC sample rate of the capture
|
||||
@@ -70,7 +146,60 @@ Evk1029SignalSource::Evk1029SignalSource(
|
||||
DLOG(INFO) << "EVK1029 Signal Source: filename=" << filename << ", fs=" << fs << ", item_size=" << item_size_
|
||||
<< ", RF_channels=" << rf_channels_ << ", seconds_to_skip=" << seconds_to_skip;
|
||||
|
||||
evk1029_source_ = evk1029_make_source(filename, queue, static_cast<double>(fs), bytes_to_skip);
|
||||
std::vector<Evk1029FreqXlatingBand> freq_xlating_bands;
|
||||
if (enable_freq_xlating_)
|
||||
{
|
||||
if (enable_throttle_control_)
|
||||
{
|
||||
LOG(ERROR) << role << ": enable_throttle_control is not supported together with enable_freq_xlating; ignoring enable_throttle_control";
|
||||
enable_throttle_control_ = false;
|
||||
}
|
||||
if (fs <= 0)
|
||||
{
|
||||
throw std::invalid_argument(role + ".enable_freq_xlating requires a positive " + role + ".sampling_frequency");
|
||||
}
|
||||
item_size_ = sizeof(gr_complex);
|
||||
for (unsigned int i = 0; i < rf_channels_; i++)
|
||||
{
|
||||
const std::string band = role + ".band"s + std::to_string(i) + "_"s;
|
||||
const double intermediate_freq = configuration->property(band + "IF"s, 0.0);
|
||||
const int decimation_factor = configuration->property(band + "decimation_factor"s, 1);
|
||||
if (decimation_factor < 1)
|
||||
{
|
||||
throw std::invalid_argument(band + "decimation_factor must be at least 1");
|
||||
}
|
||||
if (!freq_xlating_bands.empty() && decimation_factor != freq_xlating_bands.front().decimation_factor)
|
||||
{
|
||||
throw std::invalid_argument(band + "decimation_factor (" + std::to_string(decimation_factor) + ") differs from " + role + ".band0_decimation_factor (" + std::to_string(freq_xlating_bands.front().decimation_factor) + "): all bands share one output rate");
|
||||
}
|
||||
const double default_bw = (static_cast<double>(fs) / decimation_factor) / 2;
|
||||
const double bw = configuration->property(band + "bw"s, default_bw);
|
||||
const double default_tw = bw / 10.0;
|
||||
const double tw = configuration->property(band + "tw"s, default_tw);
|
||||
const std::string window_name = configuration->property(band + "window"s, "hamming"s);
|
||||
const Fir_Window window = parse_window_type(role, window_name);
|
||||
Evk1029FreqXlatingBand cfg;
|
||||
cfg.decimation_factor = decimation_factor;
|
||||
cfg.center_freq = intermediate_freq;
|
||||
if (configuration->is_present(band + "attenuation_dB"s))
|
||||
{
|
||||
const double attenuation_dB = configuration->property(band + "attenuation_dB"s, 53.0);
|
||||
const double default_kaiser_beta = kaiser_beta_from_attenuation(attenuation_dB);
|
||||
const double kaiser_beta = configuration->property(band + "kaiser_beta"s, default_kaiser_beta);
|
||||
cfg.taps = gr::filter::firdes::low_pass_2(1.0, static_cast<double>(fs), bw, tw, attenuation_dB, window, kaiser_beta);
|
||||
}
|
||||
else
|
||||
{
|
||||
const double kaiser_beta = configuration->property(band + "kaiser_beta"s, 6.76);
|
||||
cfg.taps = gr::filter::firdes::low_pass(1.0, static_cast<double>(fs), bw, tw, window, kaiser_beta);
|
||||
}
|
||||
LOG(INFO) << role << ": band " << i << " freq-xlating, IF=" << intermediate_freq
|
||||
<< " decimation=" << decimation_factor << " window=" << window_name << " taps=" << cfg.taps.size();
|
||||
freq_xlating_bands.push_back(std::move(cfg));
|
||||
}
|
||||
}
|
||||
|
||||
evk1029_source_ = evk1029_make_source(filename, queue, static_cast<double>(fs), bytes_to_skip, freq_xlating_bands);
|
||||
|
||||
if (enable_throttle_control_)
|
||||
{
|
||||
@@ -78,8 +207,14 @@ Evk1029SignalSource::Evk1029SignalSource(
|
||||
}
|
||||
if (dump_)
|
||||
{
|
||||
DLOG(INFO) << "Dumping output into file " << dump_filename_;
|
||||
file_sink_ = gr::blocks::file_sink::make(item_size_, dump_filename_.c_str());
|
||||
// One file per output port: the raw stream, or each translated band.
|
||||
const unsigned int ports = enable_freq_xlating_ ? rf_channels_ : 1;
|
||||
for (unsigned int i = 0; i < ports; i++)
|
||||
{
|
||||
const std::string filename = ports > 1 ? band_dump_filename(dump_filename_, i) : dump_filename_;
|
||||
DLOG(INFO) << "Dumping output into file " << filename;
|
||||
file_sinks_.push_back(gr::blocks::file_sink::make(item_size_, filename.c_str()));
|
||||
}
|
||||
}
|
||||
|
||||
if (in_stream > 0)
|
||||
@@ -101,14 +236,17 @@ void Evk1029SignalSource::connect(gr::top_block_sptr top_block)
|
||||
DLOG(INFO) << "connected evk1029_source to throttle";
|
||||
if (dump_)
|
||||
{
|
||||
top_block->connect(throttle_, 0, file_sink_, 0);
|
||||
top_block->connect(throttle_, 0, file_sinks_.front(), 0);
|
||||
DLOG(INFO) << "connected throttle to file sink";
|
||||
}
|
||||
}
|
||||
else if (dump_)
|
||||
else
|
||||
{
|
||||
top_block->connect(evk1029_source_, 0, file_sink_, 0);
|
||||
DLOG(INFO) << "connected evk1029_source to file sink";
|
||||
for (size_t i = 0; i < file_sinks_.size(); i++)
|
||||
{
|
||||
top_block->connect(evk1029_source_, static_cast<int>(i), file_sinks_[i], 0);
|
||||
DLOG(INFO) << "connected evk1029_source output " << i << " to file sink";
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -120,12 +258,15 @@ void Evk1029SignalSource::disconnect(gr::top_block_sptr top_block)
|
||||
top_block->disconnect(evk1029_source_, 0, throttle_, 0);
|
||||
if (dump_)
|
||||
{
|
||||
top_block->disconnect(throttle_, 0, file_sink_, 0);
|
||||
top_block->disconnect(throttle_, 0, file_sinks_.front(), 0);
|
||||
}
|
||||
}
|
||||
else if (dump_)
|
||||
else
|
||||
{
|
||||
top_block->disconnect(evk1029_source_, 0, file_sink_, 0);
|
||||
for (size_t i = 0; i < file_sinks_.size(); i++)
|
||||
{
|
||||
top_block->disconnect(evk1029_source_, static_cast<int>(i), file_sinks_[i], 0);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -22,6 +22,14 @@
|
||||
* Set role + ".seconds_to_skip" to start reading at a given time into the
|
||||
* capture (converted to a byte position using the raw sampling frequency).
|
||||
*
|
||||
* With role + ".enable_freq_xlating" = true, Evk1029Source itself translates,
|
||||
* filters and decimates each band (role + ".band<n>_*" parameters, n <
|
||||
* RF_channels) and exposes one gr_complex output port per band, which the
|
||||
* flowgraph wires to Signal Conditioner n. Pair it with Bypass Signal
|
||||
* Conditioners to connect the channels straight to those ports. All bands
|
||||
* share one decimation factor, and the throttle is not available in this
|
||||
* mode. With .dump, each band is written to its own file (suffix _ch<n>).
|
||||
*
|
||||
* -----------------------------------------------------------------------------
|
||||
*
|
||||
* GNSS-SDR is a Global Navigation Satellite System software-defined receiver.
|
||||
@@ -43,6 +51,7 @@
|
||||
#include <gnuradio/blocks/throttle.h>
|
||||
#include <cstdint>
|
||||
#include <string>
|
||||
#include <vector>
|
||||
|
||||
class ConfigurationInterface;
|
||||
|
||||
@@ -80,7 +89,7 @@ public:
|
||||
private:
|
||||
Evk1029Source_sptr evk1029_source_;
|
||||
gr::blocks::throttle::sptr throttle_;
|
||||
gr::blocks::file_sink::sptr file_sink_;
|
||||
std::vector<gr::blocks::file_sink::sptr> file_sinks_;
|
||||
|
||||
std::string dump_filename_;
|
||||
bool dump_;
|
||||
@@ -89,6 +98,7 @@ private:
|
||||
unsigned int rf_channels_;
|
||||
|
||||
bool enable_throttle_control_;
|
||||
bool enable_freq_xlating_;
|
||||
};
|
||||
|
||||
|
||||
|
||||
@@ -106,6 +106,19 @@ if(GNURADIO_USES_STD_POINTERS)
|
||||
)
|
||||
endif()
|
||||
|
||||
if(ENABLE_EVK1029)
|
||||
target_link_libraries(signal_source_gr_blocks
|
||||
PUBLIC
|
||||
Volkgnsssdr::volkgnsssdr
|
||||
PRIVATE
|
||||
Volk::volk
|
||||
)
|
||||
# volk_32fc_s32fc_x2_rotator2_32fc was introduced in VOLK 2.5
|
||||
if(VOLK_VERSION AND (VOLK_VERSION VERSION_LESS "2.5"))
|
||||
target_compile_definitions(signal_source_gr_blocks PRIVATE -DVOLK_LACKS_ROTATOR2=1)
|
||||
endif()
|
||||
endif()
|
||||
|
||||
if(ENABLE_RAW_UDP AND PCAP_FOUND)
|
||||
target_link_libraries(signal_source_gr_blocks
|
||||
PUBLIC
|
||||
|
||||
@@ -16,10 +16,22 @@
|
||||
|
||||
#include "evk1029_source.h"
|
||||
#include "command_event.h"
|
||||
#include "gnss_sdr_make_unique.h" // for std::make_unique in C++11
|
||||
#include <gnuradio/io_signature.h>
|
||||
#include <gnuradio/math.h>
|
||||
#include <volk/volk.h>
|
||||
#include <algorithm>
|
||||
#include <array>
|
||||
#include <cmath>
|
||||
#include <cstdint>
|
||||
#include <cstring>
|
||||
#include <iostream>
|
||||
#include <stdexcept>
|
||||
#include <thread>
|
||||
|
||||
#if defined(__linux__) || defined(__APPLE__)
|
||||
#include <pthread.h>
|
||||
#endif
|
||||
|
||||
#if USE_GLOG_AND_GFLAGS
|
||||
#include <glog/logging.h>
|
||||
@@ -33,14 +45,9 @@ namespace
|
||||
// requiring a huge allocation.
|
||||
constexpr std::size_t kReadChunkBytes = 1 << 20; // 1 MiB
|
||||
|
||||
// Empirically tuned (see set_output_multiple() call below): sweeping the
|
||||
// output-multiple quantum at ~178 Msps found 131072 samples (~0.74 ms)
|
||||
// measurably reduced CPU on both this block and the downstream
|
||||
// freq_xlating_fir_filter versus 8192, with 524288 showing no further
|
||||
// improvement (plateaued). Deriving the quantum from the actual sampling
|
||||
// frequency instead of hardcoding 131072 keeps that same ~0.7 ms target
|
||||
// (and so the same CPU behavior) if this is ever run at a rate other than
|
||||
// the one it was tuned against.
|
||||
// Empirically tuned output-multiple quantum (~0.7 ms); measurably reduces
|
||||
// CPU here and downstream versus a smaller one. Derived from the actual
|
||||
// sampling frequency so the same target holds at other rates.
|
||||
constexpr double kOutputMultipleTargetSeconds = 0.0007;
|
||||
constexpr int kOutputMultipleFallback = 131072; // used if sampling_frequency is unavailable/invalid
|
||||
|
||||
@@ -58,32 +65,124 @@ int output_multiple_from_sampling_frequency(double sampling_frequency)
|
||||
const double upper = std::exp2(static_cast<double>(lower_exp + 1));
|
||||
return static_cast<int>((target_samples - lower <= upper - target_samples) ? lower : upper);
|
||||
}
|
||||
|
||||
std::size_t max_history(const std::vector<Evk1029FreqXlatingBand>& bands)
|
||||
{
|
||||
std::size_t m = 0;
|
||||
for (const auto& b : bands)
|
||||
{
|
||||
m = std::max(m, b.taps.size() > 0 ? b.taps.size() - 1 : 0);
|
||||
}
|
||||
return m;
|
||||
}
|
||||
|
||||
// lut[byte] packs this byte's two decoded OBA samples (low nibble first) so
|
||||
// a single uint16_t store writes both at once on this little-endian target,
|
||||
// instead of two separate int8_t stores plus the shift/mask/scale per one.
|
||||
std::array<uint16_t, 256> make_oba_decode_lut()
|
||||
{
|
||||
std::array<uint16_t, 256> lut{};
|
||||
for (int byte = 0; byte < 256; byte++)
|
||||
{
|
||||
const auto low = static_cast<uint8_t>(static_cast<int8_t>((byte & 0x0F) * 2 - 15));
|
||||
const auto high = static_cast<uint8_t>(static_cast<int8_t>(((byte >> 4) & 0x0F) * 2 - 15));
|
||||
lut[byte] = static_cast<uint16_t>(low) | (static_cast<uint16_t>(high) << 8);
|
||||
}
|
||||
return lut;
|
||||
}
|
||||
const std::array<uint16_t, 256> kObaDecodeLut = make_oba_decode_lut();
|
||||
|
||||
// Linux and macOS's pthread_setname_np() take different arguments (the
|
||||
// macOS one can only target the calling thread), so this only ever names
|
||||
// the thread it's called from -- each BandWorker's own thread calls this
|
||||
// on itself at the top of worker_loop().
|
||||
void name_current_thread(const std::string& name)
|
||||
{
|
||||
#if defined(__linux__)
|
||||
pthread_setname_np(pthread_self(), name.c_str());
|
||||
#elif defined(__APPLE__)
|
||||
pthread_setname_np(name.c_str());
|
||||
#endif
|
||||
}
|
||||
} // namespace
|
||||
|
||||
|
||||
Evk1029Source_sptr evk1029_make_source(const std::string& filename, Concurrent_Queue<pmt::pmt_t>* queue, double sampling_frequency, uint64_t bytes_to_skip)
|
||||
Evk1029Source_sptr evk1029_make_source(const std::string& filename, Concurrent_Queue<pmt::pmt_t>* queue, double sampling_frequency, uint64_t bytes_to_skip,
|
||||
const std::vector<Evk1029FreqXlatingBand>& freq_xlating_bands)
|
||||
{
|
||||
return Evk1029Source_sptr(new Evk1029Source(filename, queue, sampling_frequency, bytes_to_skip));
|
||||
return Evk1029Source_sptr(new Evk1029Source(filename, queue, sampling_frequency, bytes_to_skip, freq_xlating_bands));
|
||||
}
|
||||
|
||||
|
||||
Evk1029Source::Evk1029Source(const std::string& filename, Concurrent_Queue<pmt::pmt_t>* queue, double sampling_frequency, uint64_t bytes_to_skip)
|
||||
: gr::sync_block("evk1029_source",
|
||||
Evk1029Source::Evk1029Source(const std::string& filename, Concurrent_Queue<pmt::pmt_t>* queue, double sampling_frequency, uint64_t bytes_to_skip,
|
||||
const std::vector<Evk1029FreqXlatingBand>& freq_xlating_bands)
|
||||
: gr::sync_decimator("evk1029_source",
|
||||
gr::io_signature::make(0, 0, 0),
|
||||
gr::io_signature::make(1, 1, sizeof(int8_t))),
|
||||
freq_xlating_bands.empty() ? gr::io_signature::make(1, 1, sizeof(int8_t)) : gr::io_signature::make(static_cast<int>(freq_xlating_bands.size()), static_cast<int>(freq_xlating_bands.size()), sizeof(gr_complex)),
|
||||
freq_xlating_bands.empty() ? 1 : freq_xlating_bands.front().decimation_factor),
|
||||
queue_(queue),
|
||||
buffer_(kReadChunkBytes),
|
||||
buffer_valid_(0),
|
||||
buffer_pos_(0)
|
||||
buffer_pos_(0),
|
||||
eof_(false),
|
||||
history_len_(max_history(freq_xlating_bands))
|
||||
{
|
||||
// Keep the scheduler from calling work() with a tiny noutput_items; each
|
||||
// unit of work produces a pair of samples (one input byte -> 2 samples).
|
||||
// This block runs at the raw, pre-decimation sample rate (up to ~180 Msps),
|
||||
// so call-frequency overhead here also drags down every downstream reader
|
||||
// (e.g. freq_xlating_fir_filter) that ends up woken at the same cadence.
|
||||
// See output_multiple_from_sampling_frequency()'s doc comment for how
|
||||
// this quantum is chosen.
|
||||
set_output_multiple(output_multiple_from_sampling_frequency(sampling_frequency));
|
||||
for (const auto& band : freq_xlating_bands)
|
||||
{
|
||||
if (band.decimation_factor != freq_xlating_bands.front().decimation_factor || band.taps.empty())
|
||||
{
|
||||
throw std::invalid_argument("EVK1029_Source: all bands need taps and the same decimation factor");
|
||||
}
|
||||
}
|
||||
|
||||
// Keep the scheduler from calling work() with a tiny noutput_items.
|
||||
// In freq_xlating mode, raising output_multiple instead deadlocks with
|
||||
// the many direct channel readers (confirmed via gdb); use a bigger
|
||||
// output buffer instead, which carries no such risk.
|
||||
if (freq_xlating_bands.empty())
|
||||
{
|
||||
set_output_multiple(output_multiple_from_sampling_frequency(sampling_frequency));
|
||||
}
|
||||
else
|
||||
{
|
||||
for (int port = 0; port < static_cast<int>(freq_xlating_bands.size()); port++)
|
||||
{
|
||||
set_min_output_buffer(port, 4000000);
|
||||
}
|
||||
}
|
||||
|
||||
for (const auto& band : freq_xlating_bands)
|
||||
{
|
||||
Band b;
|
||||
b.composite_fir.resize(band.taps.size());
|
||||
const float fwT0 = static_cast<float>(2.0 * GR_M_PI * band.center_freq / sampling_frequency);
|
||||
for (std::size_t i = 0; i < band.taps.size(); i++)
|
||||
{
|
||||
b.composite_fir[i] = band.taps[i] * std::exp(gr_complex(0, static_cast<float>(i) * fwT0));
|
||||
}
|
||||
std::reverse(b.composite_fir.begin(), b.composite_fir.end()); // match gr::filter::kernel::fir_filter::set_taps()'s internal reversal
|
||||
// Center every filter on the longest one's window, so all bands
|
||||
// get the same group delay (history_len_ / 2 input samples).
|
||||
// Aligning their newest samples instead would make a shorter
|
||||
// filter lead the others by half the length difference, which
|
||||
// biases pseudoranges between bands.
|
||||
const std::size_t slack = history_len_ - (b.composite_fir.size() - 1);
|
||||
b.history_offset = slack / 2;
|
||||
if (slack % 2 != 0)
|
||||
{
|
||||
LOG(WARNING) << "EVK1029_Source: a band filter is shorter than the longest one by an odd number of taps; "
|
||||
<< "its output lags the other bands by half an input sample (" << 0.5e9 / sampling_frequency << " ns)";
|
||||
}
|
||||
b.phase_incr = std::exp(gr_complex(0, -fwT0 * static_cast<float>(band.decimation_factor)));
|
||||
const float group_delay_phase = -fwT0 * static_cast<float>(band.taps.size() - 1) / 2.0F;
|
||||
b.phase = std::exp(gr_complex(0, group_delay_phase));
|
||||
b.samples_since_phase_renorm = 0;
|
||||
bands_.push_back(std::move(b));
|
||||
}
|
||||
decoded_.resize(history_len_); // leading history_len_ entries start at 0; harmless until the first real samples overwrite them via decode
|
||||
converted_.resize(history_len_); // same; carried-forward float history starts at 0.0F to match
|
||||
decoded_b_.resize(history_len_);
|
||||
converted_b_.resize(history_len_);
|
||||
|
||||
binary_input_file_.open(filename.c_str(), std::ios::in | std::ios::binary);
|
||||
if (binary_input_file_.is_open())
|
||||
@@ -117,11 +216,46 @@ Evk1029Source::Evk1029Source(const std::string& filename, Concurrent_Queue<pmt::
|
||||
std::cout << "EVK1029_Source: skipping the first " << aligned_offset << " bytes of the capture (requested "
|
||||
<< bytes_to_skip << ", rounded up to a 64-bit word boundary)" << std::endl;
|
||||
}
|
||||
|
||||
// One persistent FIR worker per band: work()'s own calling thread only
|
||||
// dispatches and waits.
|
||||
for (std::size_t b = 0; b < bands_.size(); b++)
|
||||
{
|
||||
workers_.push_back(std::make_unique<BandWorker>());
|
||||
BandWorker* w = workers_.back().get();
|
||||
w->thread = std::thread(&Evk1029Source::worker_loop, this, b, w);
|
||||
}
|
||||
|
||||
// Decode-ahead worker (see DecodeWorker's doc comment).
|
||||
if (!bands_.empty())
|
||||
{
|
||||
decode_worker_ = std::make_unique<DecodeWorker>();
|
||||
decode_worker_->thread = std::thread(&Evk1029Source::decode_worker_loop, this);
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
Evk1029Source::~Evk1029Source()
|
||||
{
|
||||
for (auto& w : workers_)
|
||||
{
|
||||
{
|
||||
std::lock_guard<std::mutex> lock(w->mutex);
|
||||
w->stop = true;
|
||||
}
|
||||
w->cv.notify_all();
|
||||
w->thread.join();
|
||||
}
|
||||
if (decode_worker_)
|
||||
{
|
||||
{
|
||||
std::lock_guard<std::mutex> lock(decode_worker_->mutex);
|
||||
decode_worker_->stop = true;
|
||||
}
|
||||
decode_worker_->cv.notify_all();
|
||||
decode_worker_->thread.join();
|
||||
}
|
||||
|
||||
try
|
||||
{
|
||||
if (binary_input_file_.is_open())
|
||||
@@ -140,19 +274,10 @@ Evk1029Source::~Evk1029Source()
|
||||
}
|
||||
|
||||
|
||||
int Evk1029Source::work(int noutput_items,
|
||||
gr_vector_const_void_star& input_items __attribute__((unused)),
|
||||
gr_vector_void_star& output_items)
|
||||
std::size_t Evk1029Source::decode_into_buffer(std::size_t decoded_needed, std::vector<int8_t>& target, std::size_t start_offset)
|
||||
{
|
||||
auto* out = reinterpret_cast<int8_t*>(output_items[0]);
|
||||
int produced = 0;
|
||||
bool eof = false;
|
||||
|
||||
// Each input byte unpacks to two OBA 4-bit samples (low nibble first),
|
||||
// so only ever produce an even number of items. Note noutput_items can
|
||||
// legitimately be small (even 0 or 1) on some scheduler calls, which is
|
||||
// NOT end of file -- only an actual zero-byte file read means EOF.
|
||||
while (produced + 1 < noutput_items)
|
||||
std::size_t produced = start_offset;
|
||||
while (produced + 1 < decoded_needed)
|
||||
{
|
||||
if (buffer_pos_ >= buffer_valid_)
|
||||
{
|
||||
@@ -161,25 +286,302 @@ int Evk1029Source::work(int noutput_items,
|
||||
buffer_pos_ = 0;
|
||||
if (buffer_valid_ == 0)
|
||||
{
|
||||
eof = true;
|
||||
break;
|
||||
eof_ = true;
|
||||
return produced;
|
||||
}
|
||||
}
|
||||
|
||||
const uint8_t byte = buffer_[buffer_pos_++];
|
||||
const int low_nibble = byte & 0x0F;
|
||||
const int high_nibble = (byte >> 4) & 0x0F;
|
||||
// OBA 4-bit decode: raw in [0,15] -> {-15,-13,...,-1,1,...,15}
|
||||
out[produced++] = static_cast<int8_t>(low_nibble * 2 - 15);
|
||||
out[produced++] = static_cast<int8_t>(high_nibble * 2 - 15);
|
||||
// Decode as many bytes as fit both what's left in buffer_ and what's
|
||||
// still needed, in one tight pass -- avoids re-checking the buffer
|
||||
// bound on every single byte like a naive one-byte-at-a-time loop would.
|
||||
const std::size_t bytes_wanted = (decoded_needed - produced) / 2;
|
||||
const std::size_t bytes_available = buffer_valid_ - buffer_pos_;
|
||||
const std::size_t bytes_to_decode = std::min(bytes_wanted, bytes_available);
|
||||
int8_t* dst = &target[history_len_ + produced];
|
||||
const uint8_t* src = &buffer_[buffer_pos_];
|
||||
for (std::size_t i = 0; i < bytes_to_decode; i++)
|
||||
{
|
||||
std::memcpy(dst + i * 2, &kObaDecodeLut[src[i]], sizeof(uint16_t));
|
||||
}
|
||||
buffer_pos_ += bytes_to_decode;
|
||||
produced += bytes_to_decode * 2;
|
||||
}
|
||||
|
||||
if (eof && produced == 0)
|
||||
{
|
||||
std::cout << "EVK1029_Source: EOF\n";
|
||||
queue_->push(pmt::make_any(command_event_make(200, 0)));
|
||||
return this->WORK_DONE;
|
||||
}
|
||||
|
||||
return produced;
|
||||
}
|
||||
|
||||
|
||||
void Evk1029Source::process_band(std::size_t b, int usable_noutput_items, gr_complex* out)
|
||||
{
|
||||
Band& band = bands_[b];
|
||||
const std::size_t ntaps = band.composite_fir.size();
|
||||
const int decimation = this->decimation();
|
||||
for (int i = 0; i < usable_noutput_items; i++)
|
||||
{
|
||||
volk_32fc_32f_dot_prod_32fc(&out[i], band.composite_fir.data(), converted_.data() + band.history_offset + static_cast<std::size_t>(i) * decimation, ntaps);
|
||||
}
|
||||
#if VOLK_LACKS_ROTATOR2
|
||||
volk_32fc_s32fc_x2_rotator_32fc(out, out, band.phase_incr, &band.phase, usable_noutput_items);
|
||||
#else
|
||||
volk_32fc_s32fc_x2_rotator2_32fc(out, out, &band.phase_incr, &band.phase, usable_noutput_items);
|
||||
#endif
|
||||
band.samples_since_phase_renorm += usable_noutput_items;
|
||||
if (band.samples_since_phase_renorm > 4096)
|
||||
{
|
||||
band.phase /= std::abs(band.phase);
|
||||
band.samples_since_phase_renorm = 0;
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
void Evk1029Source::worker_loop(std::size_t band_index, BandWorker* w)
|
||||
{
|
||||
name_current_thread("evk1029_wrk" + std::to_string(band_index));
|
||||
while (true)
|
||||
{
|
||||
std::unique_lock<std::mutex> lock(w->mutex);
|
||||
w->cv.wait(lock, [w] { return w->has_work || w->stop; });
|
||||
if (w->stop)
|
||||
{
|
||||
return;
|
||||
}
|
||||
const int items = w->usable_noutput_items;
|
||||
gr_complex* out = w->out;
|
||||
lock.unlock();
|
||||
|
||||
process_band(band_index, items, out);
|
||||
|
||||
lock.lock();
|
||||
w->has_work = false;
|
||||
w->finished = true;
|
||||
lock.unlock();
|
||||
w->cv.notify_all();
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
void Evk1029Source::decode_worker_loop()
|
||||
{
|
||||
name_current_thread("evk1029_dec");
|
||||
while (true)
|
||||
{
|
||||
std::unique_lock<std::mutex> lock(decode_worker_->mutex);
|
||||
decode_worker_->cv.wait(lock, [this] { return decode_worker_->has_work || decode_worker_->stop; });
|
||||
if (decode_worker_->stop)
|
||||
{
|
||||
return;
|
||||
}
|
||||
const std::size_t decoded_needed = decode_worker_->decoded_needed;
|
||||
const std::size_t start_offset = decode_worker_->start_offset;
|
||||
std::vector<int8_t>* target_decoded = decode_worker_->target_decoded;
|
||||
volk_gnsssdr::vector<float>* target_converted = decode_worker_->target_converted;
|
||||
lock.unlock();
|
||||
|
||||
if (target_decoded->size() < history_len_ + decoded_needed)
|
||||
{
|
||||
target_decoded->resize(history_len_ + decoded_needed);
|
||||
}
|
||||
const std::size_t produced = decode_into_buffer(decoded_needed, *target_decoded, start_offset);
|
||||
if (target_converted->size() < history_len_ + produced)
|
||||
{
|
||||
target_converted->resize(history_len_ + produced);
|
||||
}
|
||||
// Only the portion beyond start_offset is newly decoded -- the
|
||||
// rest (carried-forward history + leftover) already has valid,
|
||||
// previously-converted floats.
|
||||
if (produced > start_offset)
|
||||
{
|
||||
volk_8i_s32f_convert_32f(target_converted->data() + history_len_ + start_offset, target_decoded->data() + history_len_ + start_offset, 1.0F / 256.0F, produced - start_offset);
|
||||
}
|
||||
|
||||
lock.lock();
|
||||
decode_worker_->produced = produced;
|
||||
decode_worker_->has_work = false;
|
||||
decode_worker_->finished = true;
|
||||
lock.unlock();
|
||||
decode_worker_->cv.notify_all();
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
int Evk1029Source::work(int noutput_items,
|
||||
gr_vector_const_void_star& input_items __attribute__((unused)),
|
||||
gr_vector_void_star& output_items)
|
||||
{
|
||||
if (bands_.empty())
|
||||
{
|
||||
auto* out = reinterpret_cast<int8_t*>(output_items[0]);
|
||||
int produced = 0;
|
||||
bool eof = false;
|
||||
|
||||
// Each byte unpacks to two samples; only a zero-byte read is EOF.
|
||||
while (produced + 1 < noutput_items)
|
||||
{
|
||||
if (buffer_pos_ >= buffer_valid_)
|
||||
{
|
||||
binary_input_file_.read(reinterpret_cast<char*>(buffer_.data()), static_cast<std::streamsize>(buffer_.size()));
|
||||
buffer_valid_ = static_cast<std::size_t>(binary_input_file_.gcount());
|
||||
buffer_pos_ = 0;
|
||||
if (buffer_valid_ == 0)
|
||||
{
|
||||
eof = true;
|
||||
break;
|
||||
}
|
||||
}
|
||||
|
||||
const uint8_t byte = buffer_[buffer_pos_++];
|
||||
std::memcpy(&out[produced], &kObaDecodeLut[byte], sizeof(uint16_t));
|
||||
produced += 2;
|
||||
}
|
||||
|
||||
if (eof && produced == 0)
|
||||
{
|
||||
std::cout << "EVK1029_Source: EOF\n";
|
||||
queue_->push(pmt::make_any(command_event_make(200, 0)));
|
||||
return this->WORK_DONE;
|
||||
}
|
||||
|
||||
return produced;
|
||||
}
|
||||
|
||||
// Freq-xlating mode: noutput_items is post-decimation complex samples,
|
||||
// identical across every band (they share one declared decimation).
|
||||
//
|
||||
// decoded_/converted_ are always "this call's" buffers; decoded_b_/
|
||||
// converted_b_ are their ping-pong partner, prefetched by decode_worker_
|
||||
// during the PREVIOUS call's FIR work (kicked off below) so this call
|
||||
// usually just waits (near-instant) and swaps buffer identities instead
|
||||
// of decoding+converting on its own critical path. The prefetch predicts
|
||||
// "same size as this call"; noutput_items jitters slightly in practice,
|
||||
// so a mismatch is never just discarded -- undershoot is topped up
|
||||
// synchronously, overshoot's unused tail is carried forward as
|
||||
// "leftover" into the next buffer, same as FIR history already is.
|
||||
const int decimation = this->decimation();
|
||||
const std::size_t decoded_needed = static_cast<std::size_t>(noutput_items) * static_cast<std::size_t>(decimation);
|
||||
std::size_t decoded_available;
|
||||
std::size_t leftover = 0; // samples already decoded/converted beyond decoded_available, carried forward below
|
||||
|
||||
if (first_call_)
|
||||
{
|
||||
if (decoded_.size() < history_len_ + decoded_needed)
|
||||
{
|
||||
decoded_.resize(history_len_ + decoded_needed);
|
||||
}
|
||||
decoded_available = decode_into_buffer(decoded_needed, decoded_);
|
||||
if (converted_.size() < history_len_ + decoded_available)
|
||||
{
|
||||
converted_.resize(history_len_ + decoded_available);
|
||||
}
|
||||
volk_8i_s32f_convert_32f(converted_.data() + history_len_, decoded_.data() + history_len_, 1.0F / 256.0F, decoded_available);
|
||||
first_call_ = false;
|
||||
}
|
||||
else if (!decode_worker_has_pending_result_)
|
||||
{
|
||||
// No prefetch was started last time -- that means EOF was already
|
||||
// hit then (see below), so there's nothing more to give.
|
||||
decoded_available = 0;
|
||||
}
|
||||
else
|
||||
{
|
||||
std::unique_lock<std::mutex> lock(decode_worker_->mutex);
|
||||
decode_worker_->cv.wait(lock, [this] { return decode_worker_->finished; });
|
||||
std::size_t produced = decode_worker_->produced;
|
||||
lock.unlock();
|
||||
decode_worker_has_pending_result_ = false;
|
||||
|
||||
if (produced < decoded_needed && !eof_)
|
||||
{
|
||||
// Prediction (same size as the previous call) undershot --
|
||||
// should be rare given stable noutput_items; top up
|
||||
// synchronously rather than returning a short batch.
|
||||
if (decoded_b_.size() < history_len_ + decoded_needed)
|
||||
{
|
||||
decoded_b_.resize(history_len_ + decoded_needed);
|
||||
}
|
||||
const std::size_t new_produced = decode_into_buffer(decoded_needed, decoded_b_, produced);
|
||||
if (converted_b_.size() < history_len_ + new_produced)
|
||||
{
|
||||
converted_b_.resize(history_len_ + new_produced);
|
||||
}
|
||||
volk_8i_s32f_convert_32f(converted_b_.data() + history_len_ + produced, decoded_b_.data() + history_len_ + produced, 1.0F / 256.0F, new_produced - produced);
|
||||
produced = new_produced;
|
||||
}
|
||||
decoded_available = std::min(produced, decoded_needed);
|
||||
if (produced > decoded_needed)
|
||||
{
|
||||
leftover = produced - decoded_needed; // carried forward below, not discarded
|
||||
}
|
||||
|
||||
std::swap(decoded_, decoded_b_);
|
||||
std::swap(converted_, converted_b_);
|
||||
}
|
||||
|
||||
const int usable_noutput_items = static_cast<int>(decoded_available / static_cast<std::size_t>(decimation));
|
||||
|
||||
if (usable_noutput_items == 0)
|
||||
{
|
||||
if (eof_)
|
||||
{
|
||||
std::cout << "EVK1029_Source: EOF\n";
|
||||
queue_->push(pmt::make_any(command_event_make(200, 0)));
|
||||
return this->WORK_DONE;
|
||||
}
|
||||
return 0;
|
||||
}
|
||||
|
||||
// Kick off decode-ahead for the NEXT call into decoded_b_/converted_b_
|
||||
// (the now-free buffer), to overlap with this call's FIR work below.
|
||||
// Carries forward history_len_ + leftover samples: the usual FIR-history
|
||||
// tail, immediately followed by any samples decoded_ had beyond what
|
||||
// this call actually consumed (unused overshoot from the prediction),
|
||||
// so nothing decoded is ever thrown away.
|
||||
if (!eof_)
|
||||
{
|
||||
const std::size_t carry_len = history_len_ + leftover;
|
||||
if (carry_len > 0)
|
||||
{
|
||||
if (decoded_b_.size() < carry_len)
|
||||
{
|
||||
decoded_b_.resize(carry_len);
|
||||
}
|
||||
if (converted_b_.size() < carry_len)
|
||||
{
|
||||
converted_b_.resize(carry_len);
|
||||
}
|
||||
std::copy(decoded_.begin() + static_cast<std::ptrdiff_t>(decoded_available), decoded_.begin() + static_cast<std::ptrdiff_t>(decoded_available + carry_len), decoded_b_.begin());
|
||||
std::copy(converted_.begin() + static_cast<std::ptrdiff_t>(decoded_available), converted_.begin() + static_cast<std::ptrdiff_t>(decoded_available + carry_len), converted_b_.begin());
|
||||
}
|
||||
{
|
||||
std::lock_guard<std::mutex> lock(decode_worker_->mutex);
|
||||
decode_worker_->decoded_needed = std::max(decoded_needed, leftover); // predict same size as this call
|
||||
decode_worker_->start_offset = leftover;
|
||||
decode_worker_->target_decoded = &decoded_b_;
|
||||
decode_worker_->target_converted = &converted_b_;
|
||||
decode_worker_->finished = false;
|
||||
decode_worker_->has_work = true;
|
||||
}
|
||||
decode_worker_->cv.notify_all();
|
||||
decode_worker_has_pending_result_ = true;
|
||||
}
|
||||
|
||||
// Hand every band to its persistent worker, then wait for all of them.
|
||||
for (std::size_t b = 0; b < bands_.size(); b++)
|
||||
{
|
||||
BandWorker* w = workers_[b].get();
|
||||
{
|
||||
std::lock_guard<std::mutex> lock(w->mutex);
|
||||
w->usable_noutput_items = usable_noutput_items;
|
||||
w->out = reinterpret_cast<gr_complex*>(output_items[b]);
|
||||
w->finished = false;
|
||||
w->has_work = true;
|
||||
}
|
||||
w->cv.notify_all();
|
||||
}
|
||||
for (std::size_t b = 0; b < bands_.size(); b++)
|
||||
{
|
||||
BandWorker* w = workers_[b].get();
|
||||
std::unique_lock<std::mutex> lock(w->mutex);
|
||||
w->cv.wait(lock, [w] { return w->finished; });
|
||||
}
|
||||
|
||||
return usable_noutput_items;
|
||||
}
|
||||
@@ -1,30 +1,17 @@
|
||||
/*!
|
||||
* \file evk1029_source.h
|
||||
* \brief Unpacks the SAPHYRION EVK1029 raw capture files (continuous stream
|
||||
* of OBA-encoded 4-bit samples, 16 samples packed per little-endian 64-bit
|
||||
* word, no periodic block header).
|
||||
* \brief Unpacks the SAPHYRION EVK1029 raw capture files: a continuous,
|
||||
* header-less stream of OBA-encoded 4-bit samples, 16 samples packed per
|
||||
* little-endian 64-bit word.
|
||||
*
|
||||
* This is the successor of the old EVK10x9_source block, updated for the
|
||||
* current EVK1029 capture format: the old format packed two 3-bit OBA
|
||||
* samples per byte and inserted an 8-byte header every 2048-byte block;
|
||||
* the current format is a continuous stream with no headers, packing two
|
||||
* 4-bit OBA samples per byte (16 samples per 64-bit word, per the
|
||||
* capture's .sdrx metadata: quantization=4, packedbits=64, encoding=OBA).
|
||||
*
|
||||
* The block has a single output port. Multiple RF bands sharing the same
|
||||
* raw capture (e.g. several Freq_Xlating_Fir_Filter instances tuned to
|
||||
* different IFs) are each connected to that one port (see the
|
||||
* signal-source-to-conditioner wiring in gnss_flowgraph.cc, which takes
|
||||
* this path whenever output_signature()->max_streams() == 1): GNU Radio
|
||||
* supports several readers on one producer port, each with its own read
|
||||
* pointer over the same buffer. The scheduler still throttles this block
|
||||
* to the slowest of those readers (buffer::space_available() is bounded by
|
||||
* the reader with the most unconsumed items), exactly as it would with one
|
||||
* buffer per port, so the bands cannot drift apart by more than one output
|
||||
* buffer. What the single port saves is the N-1 extra output buffers and
|
||||
* the N-1 full-buffer memcpy calls per work() call, at up to the raw
|
||||
* capture rate, that a port-per-band design needs in order to duplicate
|
||||
* the same decoded samples.
|
||||
* With an empty freq_xlating_bands (the default), this block emits raw
|
||||
* decoded samples on a single output port, as before. A non-empty
|
||||
* freq_xlating_bands instead gives it one gr_complex output port per band,
|
||||
* each already frequency-translated and decimated to baseband inside this
|
||||
* block's own work() -- avoiding a separate filter block and thread per
|
||||
* band, at the cost of every band sharing one decimation factor. Bands
|
||||
* whose filters have different lengths are aligned to the center of the
|
||||
* longest one, so every band output has the same group delay.
|
||||
*
|
||||
* -----------------------------------------------------------------------------
|
||||
*
|
||||
@@ -42,11 +29,16 @@
|
||||
|
||||
#include "concurrent_queue.h"
|
||||
#include "gnss_block_interface.h"
|
||||
#include <gnuradio/sync_block.h>
|
||||
#include <gnuradio/sync_decimator.h>
|
||||
#include <pmt/pmt.h>
|
||||
#include <volk_gnsssdr/volk_gnsssdr_alloc.h> // for volk_gnsssdr::vector
|
||||
#include <condition_variable>
|
||||
#include <cstdint>
|
||||
#include <fstream>
|
||||
#include <memory>
|
||||
#include <mutex>
|
||||
#include <string>
|
||||
#include <thread>
|
||||
#include <vector>
|
||||
|
||||
/** \addtogroup Signal_Source
|
||||
@@ -59,41 +51,111 @@ class Evk1029Source;
|
||||
|
||||
using Evk1029Source_sptr = gnss_shared_ptr<Evk1029Source>;
|
||||
|
||||
// bytes_to_skip: byte position in the raw capture at which reading starts,
|
||||
// so that a long capture can be jumped straight to a time of interest
|
||||
// instead of being processed from the beginning (the adapter derives it
|
||||
// from its .seconds_to_skip parameter and the raw sampling frequency).
|
||||
// Rounded up to the next multiple of 8 bytes (64 bits) here, matching the
|
||||
// capture's native packing (16 OBA samples per little-endian 64-bit word):
|
||||
// this block itself only ever reads byte-by-byte and would decode correctly
|
||||
// from any byte offset, but seeking to a non-word-aligned byte splits a
|
||||
// 64-bit DMA word the real hardware always transferred as one unit, which
|
||||
// needlessly complicates comparing a jumped-to run against a from-the-start
|
||||
// one at the file/word level.
|
||||
Evk1029Source_sptr evk1029_make_source(const std::string& filename, Concurrent_Queue<pmt::pmt_t>* queue, double sampling_frequency, uint64_t bytes_to_skip = 0);
|
||||
//! Per-band configuration for the optional internal frequency-translation mode.
|
||||
struct Evk1029FreqXlatingBand
|
||||
{
|
||||
int decimation_factor;
|
||||
double center_freq;
|
||||
std::vector<float> taps; // real low-pass prototype, e.g. from firdes::low_pass
|
||||
};
|
||||
|
||||
// bytes_to_skip is rounded up to the next 64-bit word, matching the
|
||||
// capture's native packing, so a jumped-to run stays comparable to a
|
||||
// from-the-start one at the word level. freq_xlating_bands: see above.
|
||||
Evk1029Source_sptr evk1029_make_source(const std::string& filename, Concurrent_Queue<pmt::pmt_t>* queue, double sampling_frequency, uint64_t bytes_to_skip = 0,
|
||||
const std::vector<Evk1029FreqXlatingBand>& freq_xlating_bands = {});
|
||||
|
||||
/*!
|
||||
* \brief Reads a continuous, header-less EVK1029 raw capture file and
|
||||
* unpacks its OBA-encoded 4-bit samples (two per byte, low nibble first).
|
||||
*/
|
||||
class Evk1029Source : public gr::sync_block
|
||||
class Evk1029Source : public gr::sync_decimator
|
||||
{
|
||||
public:
|
||||
~Evk1029Source() override;
|
||||
|
||||
private:
|
||||
friend Evk1029Source_sptr evk1029_make_source(const std::string& filename, Concurrent_Queue<pmt::pmt_t>* queue, double sampling_frequency, uint64_t bytes_to_skip);
|
||||
Evk1029Source(const std::string& filename, Concurrent_Queue<pmt::pmt_t>* queue, double sampling_frequency, uint64_t bytes_to_skip = 0);
|
||||
friend Evk1029Source_sptr evk1029_make_source(const std::string& filename, Concurrent_Queue<pmt::pmt_t>* queue, double sampling_frequency, uint64_t bytes_to_skip,
|
||||
const std::vector<Evk1029FreqXlatingBand>& freq_xlating_bands);
|
||||
Evk1029Source(const std::string& filename, Concurrent_Queue<pmt::pmt_t>* queue, double sampling_frequency, uint64_t bytes_to_skip,
|
||||
const std::vector<Evk1029FreqXlatingBand>& freq_xlating_bands);
|
||||
|
||||
int work(int noutput_items,
|
||||
gr_vector_const_void_star& input_items,
|
||||
gr_vector_void_star& output_items) override;
|
||||
|
||||
// Decodes up to decoded_needed samples into target[start_offset...];
|
||||
// returns the total valid count (>= start_offset, < decoded_needed at EOF).
|
||||
std::size_t decode_into_buffer(std::size_t decoded_needed, std::vector<int8_t>& target, std::size_t start_offset = 0);
|
||||
|
||||
// Ping-pong decode-ahead: a persistent thread decodes+converts the NEXT
|
||||
// chunk into whichever of decoded_/decoded_b_ isn't the active one,
|
||||
// concurrently with this call's FIR work -- so by the time the
|
||||
// following work() call runs, decode+convert for it is usually already
|
||||
// done, taking that serial cost off the critical path. See work()'s
|
||||
// doc comment for the full protocol (bootstrap, steady state, EOF).
|
||||
struct DecodeWorker
|
||||
{
|
||||
std::thread thread;
|
||||
std::mutex mutex;
|
||||
std::condition_variable cv;
|
||||
bool has_work = false;
|
||||
bool finished = true;
|
||||
bool stop = false;
|
||||
std::size_t decoded_needed = 0;
|
||||
std::size_t start_offset = 0; // target already has this many valid samples (carried-forward leftover + history)
|
||||
std::vector<int8_t>* target_decoded = nullptr;
|
||||
volk_gnsssdr::vector<float>* target_converted = nullptr;
|
||||
std::size_t produced = 0;
|
||||
};
|
||||
void decode_worker_loop();
|
||||
std::unique_ptr<DecodeWorker> decode_worker_;
|
||||
std::vector<int8_t> decoded_b_; // decoded_'s ping-pong partner
|
||||
volk_gnsssdr::vector<float> converted_b_; // converted_'s ping-pong partner
|
||||
bool first_call_ = true;
|
||||
bool decode_worker_has_pending_result_ = false; // false only once EOF means no further prefetch was started
|
||||
|
||||
// Computes band bands_[b]'s output (FIR dot product + rotate) for
|
||||
// usable_noutput_items samples into out. Safe to run concurrently for
|
||||
// different b: only touches output_items[b] and bands_[b]'s own state.
|
||||
void process_band(std::size_t b, int usable_noutput_items, gr_complex* out);
|
||||
|
||||
// One persistent worker per band, parked on a condition variable
|
||||
// between calls rather than spawned/joined per call.
|
||||
struct BandWorker
|
||||
{
|
||||
std::thread thread;
|
||||
std::mutex mutex;
|
||||
std::condition_variable cv;
|
||||
bool has_work = false; // set by work(), cleared by the worker once it starts
|
||||
bool finished = false; // set by the worker when done, cleared by work()
|
||||
bool stop = false; // set by the destructor to end the worker's loop
|
||||
int usable_noutput_items = 0;
|
||||
gr_complex* out = nullptr;
|
||||
};
|
||||
void worker_loop(std::size_t band_index, BandWorker* w);
|
||||
std::vector<std::unique_ptr<BandWorker>> workers_;
|
||||
|
||||
std::ifstream binary_input_file_;
|
||||
Concurrent_Queue<pmt::pmt_t>* queue_;
|
||||
std::vector<uint8_t> buffer_;
|
||||
std::size_t buffer_valid_; // number of valid bytes currently in buffer_
|
||||
std::size_t buffer_pos_; // next unread byte offset within buffer_
|
||||
bool eof_;
|
||||
|
||||
// Only used when freq_xlating_bands was non-empty at construction:
|
||||
struct Band
|
||||
{
|
||||
volk_gnsssdr::vector<gr_complex> composite_fir; // frequency-shifted, reversed to match fir_filter's convention
|
||||
std::size_t history_offset; // centers this filter on the longest one's window (same group delay)
|
||||
gr_complex phase;
|
||||
gr_complex phase_incr;
|
||||
int samples_since_phase_renorm;
|
||||
};
|
||||
std::vector<Band> bands_;
|
||||
std::size_t history_len_; // max(taps.size()) - 1 across bands_, 0 if bands_ is empty
|
||||
std::vector<int8_t> decoded_; // history_len_ history samples, then newly decoded ones
|
||||
volk_gnsssdr::vector<float> converted_; // decoded_ converted to float (x256, matching char_to_short)
|
||||
};
|
||||
|
||||
|
||||
|
||||
Reference in new issue
Block a user