Merge branch 'jwmelto-zmq-signal-source' into next

CMakeLists.txt

@@ -48,6 +48,8 @@ option(ENABLE_FLEXIBAND "Enable the use of the signal source adater for the Tele
 
 option(ENABLE_ARRAY "Enable the use of CTTC's antenna array front-end as signal source (experimental)" OFF)
 
+option(ENABLE_ZMQ "Enable GNU Radio ZeroMQ Messaging, requires gr-zeromq" OFF)
+
 # Performance analysis tools
 option(ENABLE_GPERFTOOLS "Enable linking to Gperftools libraries (tcmalloc and profiler)" OFF)
 
@@ -596,7 +598,7 @@ set_package_properties(VOLK PROPERTIES
 ################################################################################
 # GNU Radio - https://www.gnuradio.org
 ################################################################################
-set(GR_REQUIRED_COMPONENTS RUNTIME PMT BLOCKS FFT FILTER ANALOG)
+list(APPEND GR_REQUIRED_COMPONENTS RUNTIME PMT BLOCKS FFT FILTER ANALOG)
 find_package(UHD)
 set_package_properties(UHD PROPERTIES
     PURPOSE "Used for communication with front-ends of the USRP family."
@@ -606,10 +608,14 @@ if(ENABLE_UHD)
     if(NOT UHD_FOUND)
         set(ENABLE_UHD OFF)
     else()
-        set(GR_REQUIRED_COMPONENTS ${GR_REQUIRED_COMPONENTS} UHD)
+        list(APPEND GR_REQUIRED_COMPONENTS UHD)
     endif()
 endif()
 
+if(ENABLE_ZMQ)
+    list(APPEND GR_REQUIRED_COMPONENTS ZEROMQ)
+endif()
+
 find_package(GNURADIO)
 set_package_properties(GNURADIO PROPERTIES
     PURPOSE "Implements flowgraph scheduler, provides some processing blocks and classes to create new ones."
@@ -3401,6 +3407,7 @@ add_feature_info(ENABLE_AD9361 ENABLE_AD9361 "Enables Ad9361_Fpga_Signal_Source
 add_feature_info(ENABLE_RAW_UDP ENABLE_RAW_UDP "Enables Custom_UDP_Signal_Source for custom UDP packet sample source. Requires libpcap.")
 add_feature_info(ENABLE_FLEXIBAND ENABLE_FLEXIBAND "Enables Flexiband_Signal_Source for using Teleorbit's Flexiband RF front-end. Requires gr-teleorbit.")
 add_feature_info(ENABLE_ARRAY ENABLE_ARRAY "Enables Raw_Array_Signal_Source and Array_Signal_Conditioner for using CTTC's antenna array. Requires gr-dbfcttc.")
+add_feature_info(ENABLE_ZMQ ENABLE_ZMQ "Enables ZMQ_Signal_Source for GNU Radio ZeroMQ messages. Requires gr-zeromq.")
 add_feature_info(ENABLE_GPERFTOOLS ENABLE_GPERFTOOLS "Enables performance analysis. Requires Gperftools.")
 add_feature_info(ENABLE_GPROF ENABLE_GPROF "Enables performance analysis with 'gprof'.")
 add_feature_info(ENABLE_CLANG_TIDY ENABLE_CLANG_TIDY "Runs clang-tidy along with the compiler. Requires Clang.")

cmake/Modules/FindGNURADIO.cmake

@@ -1,7 +1,7 @@
 # GNSS-SDR is a Global Navigation Satellite System software-defined receiver.
 # This file is part of GNSS-SDR.
 #
-# SPDX-FileCopyrightText: 2011-2020 C. Fernandez-Prades cfernandez(at)cttc.es
+# SPDX-FileCopyrightText: 2011-2022 C. Fernandez-Prades cfernandez(at)cttc.es
 # SPDX-License-Identifier: BSD-3-Clause
 
 ########################################################################
@@ -193,6 +193,7 @@ gr_module(TRELLIS gnuradio-trellis gnuradio/trellis/api.h gnuradio-trellis)
 gr_module(UHD gnuradio-uhd gnuradio/uhd/api.h gnuradio-uhd)
 gr_module(VOCODER gnuradio-vocoder gnuradio/vocoder/api.h gnuradio-vocoder)
 gr_module(WAVELET gnuradio-wavelet gnuradio/wavelet/api.h gnuradio-wavelet)
+gr_module(ZEROMQ gnuradio-zeromq gnuradio/zeromq/api.h gnuradio-zeromq)
 
 
 list(REMOVE_DUPLICATES GNURADIO_ALL_INCLUDE_DIRS)

docs/CHANGELOG.md

@@ -78,6 +78,8 @@ All notable changes to GNSS-SDR will be documented in this file.
 - The new configuration parameter `Channels_XX.RF_channel_ID` allows to specify
   the signal source per channel group.
 - Allowed the CMake project to be a sub-project.
+- Added a `ZMQ_Signal_Source` for working with streams of samples published via
+  [ZeroMQ](https://zeromq.org/).
 
 See the definitions of concepts and metrics at
 https://gnss-sdr.org/design-forces/

src/algorithms/signal_source/adapters/CMakeLists.txt

@@ -1,7 +1,7 @@
 # GNSS-SDR is a Global Navigation Satellite System software-defined receiver.
 # This file is part of GNSS-SDR.
 #
-# SPDX-FileCopyrightText: 2010-2020 C. Fernandez-Prades cfernandez(at)cttc.es
+# SPDX-FileCopyrightText: 2010-2022 C. Fernandez-Prades cfernandez(at)cttc.es
 # SPDX-License-Identifier: BSD-3-Clause
 
 
@@ -84,6 +84,11 @@ if(ENABLE_UHD)
     set(OPT_DRIVER_HEADERS ${OPT_DRIVER_HEADERS} uhd_signal_source.h)
 endif()
 
+if(ENABLE_ZMQ)
+    list(APPEND OPT_DRIVER_SOURCES zmq_signal_source.cc)
+    list(APPEND OPT_DRIVER_HEADERS zmq_signal_source.h)
+endif()
+
 
 set(SIGNAL_SOURCE_ADAPTER_SOURCES
     signal_source_base.cc
@@ -192,6 +197,13 @@ if(ENABLE_UHD)
     )
 endif()
 
+if(ENABLE_ZMQ)
+    target_link_libraries(signal_source_adapters
+        PUBLIC
+            Gnuradio::zeromq
+    )
+endif()
+
 if(ENABLE_OSMOSDR AND GROSMOSDR_FOUND)
     target_link_libraries(signal_source_adapters
         PUBLIC

src/algorithms/signal_source/adapters/signal_source_base.cc

@@ -9,7 +9,7 @@
  * GNSS-SDR is a Global Navigation Satellite System software-defined receiver.
  * This file is part of GNSS-SDR.
  *
- * Copyright (C) 2010-2021  (see AUTHORS file for a list of contributors)
+ * Copyright (C) 2010-2022  (see AUTHORS file for a list of contributors)
  * SPDX-License-Identifier: GPL-3.0-or-later
  *
  * -----------------------------------------------------------------------------
@@ -37,6 +37,12 @@ size_t SignalSourceBase::getRfChannels() const
     return rfChannels_;
 }
 
+gr::basic_block_sptr SignalSourceBase::get_left_block()
+{
+    LOG(WARNING) << "Trying to get signal source left block.";
+    return {};
+}
+
 SignalSourceBase::SignalSourceBase(ConfigurationInterface const* configuration, std::string role, std::string impl)
     : role_(std::move(role)), implementation_(std::move(impl))
 {
@@ -44,3 +50,52 @@ SignalSourceBase::SignalSourceBase(ConfigurationInterface const* configuration,
     // depending on the order of initialization, assign rfChannels_ in the body
     rfChannels_ = configuration->property(role_ + ".RF_channels"s, uint64_t(1U));
 }
+
+size_t SignalSourceBase::decode_item_type(std::string const& item_type, bool* is_interleaved, bool throw_on_error)
+{
+    size_t item_size = 0;
+
+    // The default is for samples not to be interleaved
+    if (is_interleaved) *is_interleaved = false;  // NOLINT
+
+    if (item_type == "gr_complex"s)
+        {
+            item_size = sizeof(gr_complex);
+        }
+    else if (item_type == "float"s)
+        {
+            item_size = sizeof(float);
+        }
+    else if (item_type == "short"s)
+        {
+            item_size = sizeof(int16_t);
+        }
+    else if (item_type == "ishort"s)
+        {
+            item_size = sizeof(int16_t);
+            if (is_interleaved) *is_interleaved = true;  // NOLINT
+        }
+    else if (item_type == "byte"s)
+        {
+            item_size = sizeof(int8_t);
+        }
+    else if (item_type == "ibyte"s)
+        {
+            item_size = sizeof(int8_t);
+            if (is_interleaved) *is_interleaved = true;  // NOLINT
+        }
+    else
+        {
+            if (throw_on_error)
+                {
+                    throw std::invalid_argument(item_type + " is not a recognized item type"s);
+                }
+            else
+                {
+                    LOG(WARNING) << item_type
+                                 << " unrecognized item type. Using gr_complex.";
+                    item_size = sizeof(gr_complex);
+                }
+        }
+    return item_size;
+}

src/algorithms/signal_source/adapters/signal_source_base.h

@@ -9,7 +9,7 @@
  * GNSS-SDR is a Global Navigation Satellite System software-defined receiver.
  * This file is part of GNSS-SDR.
  *
- * Copyright (C) 2010-2021  (see AUTHORS file for a list of contributors)
+ * Copyright (C) 2010-2022  (see AUTHORS file for a list of contributors)
  * SPDX-License-Identifier: GPL-3.0-or-later
  *
  * -----------------------------------------------------------------------------
@@ -32,11 +32,21 @@ public:
     std::string implementation() final;
 
     size_t getRfChannels() const override;
+    gr::basic_block_sptr get_left_block() override;  // non-sensical; implement once
 
 protected:
     //! Constructor
     SignalSourceBase(ConfigurationInterface const* configuration, std::string role, std::string impl);
 
+    //! utility for decoding passed ".item_type" values
+    //!  @param[in]  item_type - user provided string, should be one of the known types
+    //!  @param[out] is_interleaved - if non-null, the pointed to memory is updated with
+    //!                               whether the data is interleaved I/Q (e.g., ishort)
+    //!  @param[in] throw_on_error  - if true, throw an exception if the string does not
+    //!                               represent a known type
+    //!  @return the size in bytes of the passed type
+    size_t decode_item_type(std::string const& item_type, bool* is_interleaved = nullptr, bool throw_on_error = false);
+
 private:
     std::string const role_;
     std::string const implementation_;

src/algorithms/signal_source/adapters/zmq_signal_source.cc

@@ -0,0 +1,77 @@
+/*!
+ * \file zmq_signal_source.cc
+ * \brief Signal source which reads from ZeroMQ.
+ * \author Jim Melton, 2022. jim.melton(at)sncorp.com
+ *
+ * -----------------------------------------------------------------------------
+ *
+ * GNSS-SDR is a Global Navigation Satellite System software-defined receiver.
+ * This file is part of GNSS-SDR.
+ *
+ * Copyright (C) 2010-2022  (see AUTHORS file for a list of contributors)
+ * SPDX-License-Identifier: GPL-3.0-or-later
+ *
+ * -----------------------------------------------------------------------------
+ */
+
+#include "zmq_signal_source.h"
+#include "configuration_interface.h"
+#include "gnss_sdr_string_literals.h"
+#include <glog/logging.h>
+
+using namespace std::string_literals;
+
+ZmqSignalSource::ZmqSignalSource(const ConfigurationInterface* configuration,
+    const std::string& role,
+    unsigned int /* in_stream [[maybe_unused]] */,
+    unsigned int /* out_stream [[maybe_unused]] */,
+    Concurrent_Queue<pmt::pmt_t>* /* queue [[maybe_unused]] */)
+    : SignalSourceBase(configuration, role, "ZMQ_Signal_Source"s),
+      d_item_size(decode_item_type(configuration->property(role + ".item_type"s, "gr_complex"s), nullptr, true)),
+      d_dump_filename(configuration->property(role + ".dump_filename"s, "data/zmq_dump.dat"s)),
+      d_dump(configuration->property(role + ".dump"s, false))
+{
+    auto vlen = configuration->property(role + ".vlen"s, 1);
+    auto pass_tags = configuration->property(role + ".pass_tags"s, false);
+    auto timeout_ms = configuration->property(role + ".timeout_ms"s, 100);
+    auto hwm = configuration->property(role + ".hwm"s, -1);
+
+    auto property = role + ".endpoint"s;
+    auto endpoint = configuration->property(property, ""s);
+
+    if (not endpoint.empty())
+        {
+            LOG(INFO) << "Connecting to ZMQ pub at " << endpoint;
+            d_source_block = gr::zeromq::sub_source::make(d_item_size, vlen, endpoint.data(), timeout_ms, pass_tags, hwm);
+        }
+    else
+        {
+            std::cerr << "For ZMQ_Signal_Source " << property << " must be defined" << std::endl;
+            throw std::invalid_argument(property + ": undefined");
+        }
+}
+
+
+auto ZmqSignalSource::item_size() -> size_t { return d_item_size; }
+
+auto ZmqSignalSource::connect(gr::top_block_sptr top_block) -> void
+{
+    if (d_dump)
+        {
+            d_dump_sink = gr::blocks::file_sink::make(item_size(), d_dump_filename.data());
+            top_block->connect(d_source_block, 0, d_dump_sink, 0);
+        }
+}
+
+auto ZmqSignalSource::disconnect(gr::top_block_sptr top_block) -> void
+{
+    if (d_dump)
+        {
+            top_block->disconnect(d_dump_sink);
+        }
+}
+
+auto ZmqSignalSource::get_right_block() -> gr::basic_block_sptr
+{
+    return d_source_block;
+}

src/algorithms/signal_source/adapters/zmq_signal_source.h

@@ -0,0 +1,78 @@
+/*!
+ * \file zmq_signal_source.h
+ * \brief Signal source which reads from ZeroMQ.
+ * \author Jim Melton, 2022. jim.melton(at)sncorp.com
+ *
+ * -----------------------------------------------------------------------------
+ *
+ * GNSS-SDR is a Global Navigation Satellite System software-defined receiver.
+ * This file is part of GNSS-SDR.
+ *
+ * Copyright (C) 2010-2022  (see AUTHORS file for a list of contributors)
+ * SPDX-License-Identifier: GPL-3.0-or-later
+ *
+ * -----------------------------------------------------------------------------
+ */
+
+#ifndef GNSS_SDR_ZMQ_SIGNAL_SOURCE_H
+#define GNSS_SDR_ZMQ_SIGNAL_SOURCE_H
+
+#include "signal_source_base.h"
+//
+#include "concurrent_queue.h"
+#include <gnuradio/blocks/file_sink.h>  // for dump
+#include <gnuradio/zeromq/sub_source.h>
+#include <pmt/pmt.h>
+#include <string>
+
+/** \addtogroup Signal_Source
+ * \{ */
+/** \addtogroup Signal_Source_adapters
+ * \{ */
+
+//! This class supports the following properties:
+//!
+//!   .endpoint   - the ZMQ endpoint to be connected to
+//!   .vlen       - vector length of the input items (default 1, one item)
+//!                 this must match the size of the publisher!
+//!   .pass_tags  - boolean flag if tags should be propagated (default false)
+//!   .timeout_ms - receive timeout, in milliseconds (default 100)
+//!   .hwm        - ZMQ high water mark (default -1, ZMQ default)
+//!
+//!   .item_type - data type of the samples (default "gr_complex")
+//!
+//! (probably should be abstracted to the base class)
+//!
+//!   .dump          - whether to archive input data
+//!   .dump_filename - if dumping, path to file for output
+//!
+
+class ZmqSignalSource : public SignalSourceBase
+{
+public:
+    ZmqSignalSource(const ConfigurationInterface* configuration,
+        const std::string& role,
+        unsigned int in_stream,
+        unsigned int out_stream,
+        Concurrent_Queue<pmt::pmt_t>* queue);
+
+    ~ZmqSignalSource() = default;
+
+    size_t item_size() override;
+
+    auto connect(gr::top_block_sptr top_block) -> void override;
+    auto disconnect(gr::top_block_sptr top_block) -> void override;
+    auto get_right_block() -> gr::basic_block_sptr override;
+
+private:
+    gr::zeromq::sub_source::sptr d_source_block;
+    gr::blocks::file_sink::sptr d_dump_sink;
+
+    size_t d_item_size;
+    std::string d_dump_filename;
+    bool d_dump;
+};
+
+/** \} */
+/** \} */
+#endif

src/core/receiver/CMakeLists.txt

@@ -1,7 +1,7 @@
 # GNSS-SDR is a Global Navigation Satellite System software-defined receiver.
 # This file is part of GNSS-SDR.
 #
-# SPDX-FileCopyrightText: 2010-2020 C. Fernandez-Prades cfernandez(at)cttc.es
+# SPDX-FileCopyrightText: 2010-2022 C. Fernandez-Prades cfernandez(at)cttc.es
 # SPDX-License-Identifier: BSD-3-Clause
 
 
@@ -112,6 +112,10 @@ if(ENABLE_ARRAY)
     target_compile_definitions(core_receiver PRIVATE -DRAW_ARRAY_DRIVER=1)
 endif()
 
+if(ENABLE_ZMQ)
+    target_compile_definitions(core_receiver PRIVATE -DZEROMQ_DRIVER=1)
+endif()
+
 if(ENABLE_FLEXIBAND)
     target_compile_definitions(core_receiver PRIVATE -DFLEXIBAND_DRIVER=1)
 endif()

src/core/receiver/gnss_block_factory.cc

@@ -16,7 +16,7 @@
  * GNSS-SDR is a Global Navigation Satellite System software-defined receiver.
  * This file is part of GNSS-SDR.
  *
- * Copyright (C) 2010-2020  (see AUTHORS file for a list of contributors)
+ * Copyright (C) 2010-2022  (see AUTHORS file for a list of contributors)
  * SPDX-License-Identifier: GPL-3.0-or-later
  *
  * -----------------------------------------------------------------------------
@@ -171,6 +171,10 @@
 #include "flexiband_signal_source.h"
 #endif
 
+#if ZEROMQ_DRIVER
+#include "zmq_signal_source.h"
+#endif
+
 #if CUDA_GPU_ACCEL
 #include "gps_l1_ca_dll_pll_tracking_gpu.h"
 #endif
@@ -799,6 +803,16 @@ std::unique_ptr<GNSSBlockInterface> GNSSBlockFactory::GetBlock(
                 }
 #endif
 
+#if ZEROMQ_DRIVER
+            else if (implementation == "ZMQ_Signal_Source")
+                {
+                    std::unique_ptr<GNSSBlockInterface> block_ = std::make_unique<ZmqSignalSource>(configuration, role, in_streams,
+                        out_streams, queue);
+                    block = std::move(block_);
+                }
+#endif
+
+
             // DATA TYPE ADAPTER -----------------------------------------------------------
             else if (implementation == "Byte_To_Short")
                 {