Used zeromq to connect recorders dynamically.

Author: shajen

CMakeLists.txt

@@ -21,6 +21,7 @@ find_package(Gnuradio COMPONENTS
     video-sdl
     network
     soapy
+    zeromq
 )
 find_package(nlohmann_json REQUIRED)
 find_package(PahoMqttCpp REQUIRED)
@@ -48,6 +49,7 @@ target_link_libraries(auto_sdr
     gnuradio::gnuradio-fft
     gnuradio::gnuradio-filter
     gnuradio::gnuradio-soapy
+    gnuradio::gnuradio-zeromq
     nlohmann_json::nlohmann_json
     spdlog::spdlog
     PahoMqttCpp::paho-mqttpp3
@@ -62,6 +64,7 @@ target_link_libraries(auto_sdr_test
     gnuradio::gnuradio-fft
     gnuradio::gnuradio-filter
     gnuradio::gnuradio-soapy
+    gnuradio::gnuradio-zeromq
     nlohmann_json::nlohmann_json
     spdlog::spdlog
     PahoMqttCpp::paho-mqttpp3

sources/application.cpp

@@ -24,7 +24,7 @@ Application::Application(nlohmann::json& tmpJson, const ArgConfig& argConfig)
       if (!device.enabled) {
         Logger::info(LABEL, "device disabled, skipping: {}", colored(GREEN, "{}", device.getName()));
       } else {
-        m_scanners.push_back(std::make_unique<Scanner>(m_config, device, m_remoteController, m_config.recordersCount()));
+        m_scanners.push_back(std::make_unique<Scanner>(m_config, device, m_remoteController));
       }
     } catch (const std::exception& exception) {
       Logger::exception(LABEL, exception, SPDLOG_LOC, "open device failed: {}", device.getName());

sources/arg_config.h

@@ -11,4 +11,5 @@ struct ArgConfig {
   std::string mqttUrl;
   std::string mqttUser;
   std::string mqttPassword;
+  std::string workDir = ".";
 };

sources/config.cpp

@@ -57,4 +57,6 @@ std::string Config::mqttPassword() const { return m_argConfig.mqttPassword; }
 
 std::string Config::latitude() const { return m_fileConfig.position.latitude; }
 std::string Config::longitude() const { return m_fileConfig.position.longitude; }
-int Config::altitude() const { return m_fileConfig.position.altitude; }
+int Config::altitude() const { return m_fileConfig.position.altitude; }
+
+std::string Config::workDir() const { return m_argConfig.workDir; }

sources/config.h

@@ -63,6 +63,8 @@ class Config {
   std::string longitude() const;
   int altitude() const;
 
+  std::string workDir() const;
+
  private:
   const std::string m_id;
   const ArgConfig& m_argConfig;

sources/main.cpp

@@ -29,6 +29,7 @@ int main(int argc, char** argv) {
   app.add_option("--mqtt-url", argConfig.mqttUrl, "mqtt url")->required();
   app.add_option("--mqtt-user", argConfig.mqttUser, "mqtt username")->required();
   app.add_option("--mqtt-password", argConfig.mqttPassword, "mqtt password")->required();
+  app.add_option("--work-dir", argConfig.workDir, "work directory");
   CLI11_PARSE(app, argc, argv);
 
   dup2(fileno(fopen("/dev/null", "w")), fileno(stderr));

sources/radio/recorder.cpp

@@ -4,6 +4,7 @@
 #include <gnuradio/blocks/complex_to_interleaved_char.h>
 #include <gnuradio/blocks/stream_to_vector.h>
 #include <gnuradio/filter/rational_resampler.h>
+#include <gnuradio/zeromq/sub_source.h>
 #include <logger.h>
 #include <network/query.h>
 #include <radio/blocks/file_sink.h>
@@ -12,18 +13,26 @@
 
 constexpr auto LABEL = "recorder";
 
-Recorder::Recorder(const Config& config, std::shared_ptr<gr::top_block> tb, std::shared_ptr<gr::block> source, Frequency sampleRate, std::function<void(const nlohmann::json&)> send)
-    : m_config(config), m_sampleRate(sampleRate), m_send(send), m_connector(tb) {
+Recorder::Recorder(const Config& config, const std::string& zeromq, Frequency sampleRate, const Recording& recording, std::function<void(const nlohmann::json&)> send)
+    : m_config(config), m_sampleRate(sampleRate), m_recording(recording), m_send(send), m_tb(gr::make_top_block("recorder")), m_rawFileSinkBlock(nullptr), m_connector(m_tb) {
+  Logger::info(
+      LABEL,
+      "start recorder, source: {}, name: {}, frequency: {}, bandwidth: {}, modulation: {}",
+      colored(BLUE, "{}", m_recording.source),
+      colored(BLUE, "{}", m_recording.name),
+      formatFrequency(m_recording.recordingFrequency, GREEN),
+      formatFrequency(m_recording.bandwidth, GREEN),
+      colored(BLUE, "{}", m_recording.modulation));
+
+  auto source = gr::zeromq::sub_source::make(sizeof(gr_complex), 1, const_cast<char*>(zeromq.c_str()));
   std::vector<Block> blocks;
-  m_blocker = std::make_shared<Blocker>(gr::io_signature::make(1, 1, sizeof(gr_complex)), true);
   m_shiftBlock = gr::blocks::rotator_cc::make();
   blocks.push_back(source);
-  blocks.push_back(m_blocker);
   blocks.push_back(m_shiftBlock);
 
   Block lastResampler;
   for (const auto& [factor1, factor2] : getResamplersFactors(m_sampleRate, m_recording.bandwidth, RESAMPLER_THRESHOLD)) {
-    Logger::info(LABEL, "rational resampler factors: {}, {}", colored(GREEN, "{}", factor1), colored(GREEN, "{}", factor2));
+    Logger::debug(LABEL, "rational resampler factors: {}, {}", colored(GREEN, "{}", factor1), colored(GREEN, "{}", factor2));
     lastResampler = gr::filter::rational_resampler<gr_complex, gr_complex, gr_complex>::make(factor1, factor2);
     blocks.push_back(lastResampler);
   }
@@ -39,49 +48,30 @@ Recorder::Recorder(const Config& config, std::shared_ptr<gr::top_block> tb, std:
     m_rawFileSinkBlock = std::make_shared<FileSink<gr_complex>>(1, true);
     m_connector.connect(lastResampler, m_rawFileSinkBlock);
   }
-}
-
-Recorder::~Recorder() {
-  Logger::info(LABEL, "stopping");
-  stopRecording();
-  Logger::info(LABEL, "stoped");
-}
-
-Recording Recorder::getRecording() const { return m_recording; }
-
-bool Recorder::isRecording() { return !m_blocker->isBlocking(); }
 
-void Recorder::startRecording(const Recording& recording) {
-  if (!isRecording()) {
-    m_firstDataTime = getTime();
-    m_lastDataTime = m_firstDataTime;
-    m_recording = recording;
-    m_shiftBlock->set_phase_inc(2.0l * M_PIl * (static_cast<double>(-recording.shift()) / static_cast<float>(m_sampleRate)));
-    if (DEBUG_SAVE_RECORDING_RAW_IQ) {
-      m_rawFileSinkBlock->startRecording(getRawFileName("recording", "fc", recording.recordingFrequency, m_recording.bandwidth));
-    }
-    m_blocker->setBlocking(false);
-    m_buffer->clear();
-  } else {
-    Logger::warn(LABEL, "can not start recording, recorder already recording");
+  m_firstDataTime = getTime();
+  m_lastDataTime = m_firstDataTime;
+  m_shiftBlock->set_phase_inc(2.0l * M_PIl * (static_cast<double>(-m_recording.shift()) / static_cast<float>(m_sampleRate)));
+  if (m_rawFileSinkBlock) {
+    m_rawFileSinkBlock->startRecording(getRawFileName("recording", "fc", m_recording.recordingFrequency, m_recording.bandwidth));
   }
+  m_tb->start();
 }
 
-void Recorder::stopRecording() {
-  if (isRecording()) {
-    if (DEBUG_SAVE_RECORDING_RAW_IQ) {
-      m_rawFileSinkBlock->stopRecording();
-    }
-    m_blocker->setBlocking(true);
-    m_buffer->clear();
-  } else {
-    Logger::warn(LABEL, "can not stop recording, recorder do not recording");
+Recorder::~Recorder() {
+  Logger::info(LABEL, "stop recorder, frequency: {}, time: {} ms", formatFrequency(m_recording.recordingFrequency, RED), getDuration().count());
+  if (m_rawFileSinkBlock) {
+    m_rawFileSinkBlock->stopRecording();
   }
+  m_tb->stop();
+  m_tb->wait();
 }
 
+Recording Recorder::getRecording() const { return m_recording; }
+
 void Recorder::flush() {
   m_lastDataTime = getTime();
-  if (DEBUG_SAVE_RECORDING_RAW_IQ) {
+  if (m_rawFileSinkBlock) {
     m_rawFileSinkBlock->flush();
   }
   m_buffer->popSingleSample([this](const SimpleComplex* data, const int size, const std::chrono::milliseconds& time) {

sources/radio/recorder.h

@@ -20,23 +20,20 @@ class Recorder {
   Recorder(const Recorder&) = delete;
   Recorder& operator=(const Recorder&) = delete;
 
-  Recorder(const Config& config, std::shared_ptr<gr::top_block> tb, std::shared_ptr<gr::block> source, Frequency sampleRate, std::function<void(const nlohmann::json&)> send);
+  Recorder(const Config& config, const std::string& zeromq, Frequency sampleRate, const Recording& recording, std::function<void(const nlohmann::json&)> send);
   ~Recorder();
 
   Recording getRecording() const;
-  bool isRecording();
-  void startRecording(const Recording& recording);
-  void stopRecording();
   void flush();
   std::chrono::milliseconds getDuration() const;
 
  private:
   const Config& m_config;
   const Frequency m_sampleRate;
-  Recording m_recording;
+  const Recording m_recording;
   const std::function<void(const nlohmann::json&)> m_send;
 
-  std::shared_ptr<Blocker> m_blocker;
+  std::shared_ptr<gr::top_block> m_tb;
   std::shared_ptr<gr::blocks::rotator_cc> m_shiftBlock;
   std::shared_ptr<FileSink<gr_complex>> m_rawFileSinkBlock;
   std::shared_ptr<Buffer<SimpleComplex>> m_buffer;

sources/radio/sdr_device.cpp

@@ -7,32 +7,35 @@
 #include <gnuradio/fft/fft_v.h>
 #include <gnuradio/fft/window.h>
 #include <gnuradio/soapy/source.h>
+#include <gnuradio/zeromq/pub_sink.h>
 #include <logger.h>
 #include <radio/blocks/decimator.h>
 #include <radio/blocks/psd.h>
 #include <radio/blocks/spectrogram.h>
 
+#include <filesystem>
+
 constexpr auto LABEL = "sdr";
 
-SdrDevice::SdrDevice(const Config& config, const Device& device, RemoteController& remoteController, TransmissionNotification& notification, const int recordersCount)
-    : m_sampleRate(device.sample_rate), m_isInitialized(false), m_frequencyRange({0, 0}), m_tb(gr::make_top_block("sdr")), m_powerFileSink(nullptr), m_rawIqFileSink(nullptr), m_connector(m_tb) {
+SdrDevice::SdrDevice(const Config& config, const Device& device, RemoteController& remoteController, TransmissionNotification& notification)
+    : m_config(config),
+      m_device(device),
+      m_zeromq(fmt::format("ipc://{}/{}_{}_zeromq_stream.sock", std::filesystem::canonical(m_config.workDir()).string(), device.driver, device.serial)),
+      m_remoteController(remoteController),
+      m_sampleRate(device.sample_rate),
+      m_isInitialized(false),
+      m_frequencyRange({0, 0}),
+      m_tb(gr::make_top_block("sdr")),
+      m_powerFileSink(nullptr),
+      m_rawIqFileSink(nullptr),
+      m_connector(m_tb) {
   Logger::info(LABEL, "starting");
-  Logger::info(
-      LABEL,
-      "driver: {}, serial: {}, sample rate: {}, recorders: {}",
-      colored(GREEN, "{}", device.driver),
-      colored(GREEN, "{}", device.serial),
-      formatFrequency(m_sampleRate),
-      colored(GREEN, "{}", recordersCount));
+  Logger::info(LABEL, "driver: {}, serial: {}, sample rate: {}", colored(GREEN, "{}", device.driver), colored(GREEN, "{}", device.serial), formatFrequency(m_sampleRate));
+  Logger::info(LABEL, "zeromq: {}", colored(GREEN, "{}", m_zeromq));
 
   m_source = std::make_shared<SdrSource>(device);
   setupChains(config, device, remoteController, notification);
 
-  Logger::info(LABEL, "recording bandwidth: {}", formatFrequency(config.recordingBandwidth()));
-  for (int i = 0; i < recordersCount; ++i) {
-    m_recorders.push_back(std::make_unique<Recorder>(config, m_tb, m_source, m_sampleRate, std::bind(&RemoteController::sendTransmission, remoteController, device.getName(), std::placeholders::_1)));
-  }
-
   m_tb->start();
   Logger::info(LABEL, "started");
 }
@@ -41,6 +44,7 @@ SdrDevice::~SdrDevice() {
   Logger::info(LABEL, "stopping");
   m_tb->stop();
   m_tb->wait();
+  std::remove(m_zeromq.c_str());
   Logger::info(LABEL, "stopped");
 }
 
@@ -73,66 +77,44 @@ void SdrDevice::setFrequencyRange(FrequencyRange frequencyRange) {
 }
 
 void SdrDevice::updateRecordings(const std::vector<Recording> recordings) {
-  const auto isWaitingForRecording = [&recordings](const Frequency& frequency) {
-    return std::find_if(recordings.begin(), recordings.end(), [frequency](const Recording& recording) {
-             // improve auto formatter
-             return frequency == recording.recordingFrequency;
-           }) != recordings.end();
-  };
-  const auto getRecorder = [this](const Frequency& frequency) {
-    return std::find_if(m_recorders.begin(), m_recorders.end(), [frequency](const std::unique_ptr<Recorder>& recorder) {
+  const auto findRecorder = [this](const Recording& recording) {
+    return std::find_if(m_recorders.begin(), m_recorders.end(), [recording](const std::unique_ptr<Recorder>& recorder) {
       // improve auto formatter
-      return frequency == recorder->getRecording().recordingFrequency;
+      return recording.recordingFrequency == recorder->getRecording().recordingFrequency;
     });
   };
-  const auto getFreeRecorder = [this]() {
-    return std::find_if(m_recorders.begin(), m_recorders.end(), [](const std::unique_ptr<Recorder>& recorder) {
-      // improve auto formatter
-      return !recorder->isRecording();
-    });
+
+  const auto isRecordingActive = [recordings](const Recording& recording1) {
+    return std::find_if(recordings.begin(), recordings.end(), [recording1](const Recording& recording2) {
+             // improve auto formatter
+             return recording1.recordingFrequency == recording2.recordingFrequency;
+           }) != recordings.end();
   };
 
-  for (auto& recorder : m_recorders) {
-    if (recorder->isRecording()) {
-      if (!isWaitingForRecording(recorder->getRecording().recordingFrequency)) {
-        recorder->stopRecording();
-        Logger::info(LABEL, "stop recorder, frequency: {}, time: {} ms", formatFrequency(recorder->getRecording().recordingFrequency, RED), recorder->getDuration().count());
-      }
-    }
+  std::erase_if(m_recorders, [isRecordingActive](const std::unique_ptr<Recorder>& recorder) { return !isRecordingActive(recorder->getRecording()); });
+
+  if (m_recorders.size() < static_cast<size_t>(m_config.recordersCount())) {
+    ignoredTransmissions.clear();
   }
 
   for (const auto& recording : recordings) {
-    const auto itRecorder = getRecorder(recording.recordingFrequency);
-    if (itRecorder != m_recorders.end()) {
-      const auto& recorder = *itRecorder;
-      if (!recorder->isRecording()) {
-        Logger::warn(LABEL, "start recorder that should be already started, frequency: {}", formatFrequency(recording.recordingFrequency), GREEN);
-      }
+    const auto it = findRecorder(recording);
+    if (it != m_recorders.end()) {
       if (recording.flush) {
-        recorder->flush();
+        (*it)->flush();
       }
     } else {
-      const auto itFreeRecorder = getFreeRecorder();
-      if (itFreeRecorder != m_recorders.end()) {
-        const auto& freeRecorder = *itFreeRecorder;
-        freeRecorder->startRecording(recording);
-        Logger::info(LABEL, "start recorder, frequency: {}", formatFrequency(recording.recordingFrequency, GREEN));
+      if (m_recorders.size() < static_cast<size_t>(m_config.recordersCount())) {
+        m_recorders.push_back(
+            std::make_unique<Recorder>(m_config, m_zeromq, m_sampleRate, recording, std::bind(&RemoteController::sendTransmission, m_remoteController, m_device.getName(), std::placeholders::_1)));
       } else {
         if (!ignoredTransmissions.count(recording.recordingFrequency)) {
-          Logger::info(LABEL, "no recorders available, frequency: {}", formatFrequency(recording.recordingFrequency, YELLOW));
+          Logger::info(LABEL, "maximum recorders limit reached, frequency: {}", formatFrequency(recording.recordingFrequency, RED));
           ignoredTransmissions.insert(recording.recordingFrequency);
         }
       }
     }
   }
-
-  for (auto it = ignoredTransmissions.begin(); it != ignoredTransmissions.cend();) {
-    if (isWaitingForRecording(*it)) {
-      it++;
-    } else {
-      ignoredTransmissions.erase(it++);
-    }
-  }
 }
 
 Frequency SdrDevice::getFrequency() const { return m_frequencyRange.center(); }
@@ -169,4 +151,7 @@ void SdrDevice::setupChains(const Config& config, const Device& device, RemoteCo
     m_rawIqFileSink = std::make_shared<FileSink<gr_complex>>(1, false);
     m_connector.connect<Block>(m_source, m_rawIqFileSink);
   }
+
+  auto sink = gr::zeromq::pub_sink::make(sizeof(gr_complex), 1, const_cast<char*>(m_zeromq.c_str()));
+  m_connector.connect<Block>(m_source, sink);
 }

sources/radio/sdr_device.h

@@ -19,7 +19,7 @@
 
 class SdrDevice {
  public:
-  SdrDevice(const Config& config, const Device& device, RemoteController& remoteController, TransmissionNotification& notification, const int recordersCount);
+  SdrDevice(const Config& config, const Device& device, RemoteController& remoteController, TransmissionNotification& notification);
   ~SdrDevice();
 
   void setFrequencyRange(FrequencyRange frequencyRange);
@@ -29,6 +29,10 @@ class SdrDevice {
   Frequency getFrequency() const;
   void setupChains(const Config& config, const Device& device, RemoteController& remoteController, TransmissionNotification& notification);
 
+  const Config& m_config;
+  const Device m_device;
+  const std::string m_zeromq;
+  RemoteController& m_remoteController;
   const Frequency m_sampleRate;
   bool m_isInitialized;
   FrequencyRange m_frequencyRange;