add ZMQ signal source

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 gr::basic_block_sptr();
+}
+
 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,79 @@
+/*!
+ * \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"
+
+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))
+{
+    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);
+
+    // Each .endpointN must be specified
+    for (auto n = 0u; n < getRfChannels(); ++n)
+        {
+            auto property = role + ".endpoint"s + std::to_string(n);
+            auto endpoint = configuration->property(property, ""s);
+            if (not endpoint.empty())
+                {
+                    LOG(INFO) << "Connecting to ZMQ pub at " << endpoint;
+                    d_source_blocks.push_back(gr::zeromq::sub_source::make(d_item_size, vlen, endpoint.data(), timeout_ms, pass_tags, hwm));
+                }
+            else
+                {
+                    std::cerr
+                        << "For ZMQ_Signal_Source " << role << ", the .endpointN property must be defined\n"
+                        << "for all values of N from 0 to " << getRfChannels() - 1 << 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 [[maybe_unused]] */) -> void
+{
+    // for now, nothing to connect
+}
+
+auto ZmqSignalSource::disconnect(gr::top_block_sptr /* top_block [[maybe_unused]] */) -> void
+{
+    // for now, nothing to disconnect
+}
+
+auto ZmqSignalSource::get_right_block() -> gr::basic_block_sptr
+{
+    return d_source_blocks.front();
+}
+
+auto ZmqSignalSource::get_right_block(int RF_channel) -> gr::basic_block_sptr
+{
+    return d_source_blocks.at(RF_channel);  // throws std::out_of_range
+}

src/algorithms/signal_source/adapters/zmq_signal_source.h

@@ -0,0 +1,74 @@
+/*!
+ * \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/zeromq/sub_source.h>
+#include <pmt/pmt.h>
+
+/** \addtogroup Signal_Source
+ * \{ */
+/** \addtogroup Signal_Source_adapters
+ * \{ */
+
+//! This class supports the following properties:
+//!
+//!   .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)
+//!   .vlen       - vector length of the input items (default 1, one item)
+//!   .endpointN  - the ZMQ endpoint to be connected to (repeat for each channel)
+//!
+//!   .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;
+    auto get_right_block(int RF_channel) -> gr::basic_block_sptr override;
+
+private:
+    std::vector<gr::zeromq::sub_source::sptr> d_source_blocks;
+
+    size_t d_item_size;
+};
+
+/** \} */
+/** \} */
+#endif