Add events

20 files changed, 287 insertions, 62 deletions

diff --git a/Makefile.am b/Makefile.am
index 6e7c9ce..5c75c62 100644
--- a/Makefile.am
+++ b/Makefile.am
@@ -54,6 +54,8 @@ odr_dabmod_SOURCES  = src/DabMod.cpp \
 					  src/EtiReader.h \
 					  src/Eti.cpp \
 					  src/Eti.h \
+					  src/Events.cpp \
+					  src/Events.h \
 					  src/FicSource.cpp \
 					  src/FicSource.h \
 					  src/PuncturingRule.cpp \
diff --git a/doc/receive_events.py b/doc/receive_events.py
new file mode 100755
index 0000000..dca27cd
--- /dev/null
+++ b/doc/receive_events.py
@@ -0,0 +1,59 @@
+#!/usr/bin/env python
+#
+# This is an example program that shows
+# how to receive runtime events from ODR-DabMod
+#
+# LICENSE: see bottom of file
+
+import sys
+import zmq
+import json
+from pprint import pprint
+
+context = zmq.Context()
+sock = context.socket(zmq.SUB)
+
+ep = "tcp://127.0.0.1:5557"
+print(f"Receive from {ep}")
+sock.connect(ep)
+
+# subscribe to all events
+sock.setsockopt(zmq.SUBSCRIBE, bytes([]))
+
+while True:
+    parts = sock.recv_multipart()
+    if len(parts) == 2:
+        print("Received event '{}'".format(parts[0].decode()))
+        pprint(json.loads(parts[1].decode()))
+
+    else:
+        print("Received strange event:")
+        pprint(parts)
+
+    print()
+
+
+# This is free and unencumbered software released into the public domain.
+#
+# Anyone is free to copy, modify, publish, use, compile, sell, or
+# distribute this software, either in source code form or as a compiled
+# binary, for any purpose, commercial or non-commercial, and by any
+# means.
+#
+# In jurisdictions that recognize copyright laws, the author or authors
+# of this software dedicate any and all copyright interest in the
+# software to the public domain. We make this dedication for the benefit
+# of the public at large and to the detriment of our heirs and
+# successors. We intend this dedication to be an overt act of
+# relinquishment in perpetuity of all present and future rights to this
+# software under copyright law.
+#
+# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
+# EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
+# MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT.
+# IN NO EVENT SHALL THE AUTHORS BE LIABLE FOR ANY CLAIM, DAMAGES OR
+# OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE,
+# ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR
+# OTHER DEALINGS IN THE SOFTWARE.
+#
+# For more information, please refer to <http://unlicense.org>
diff --git a/lib/Json.cpp b/lib/Json.cpp
index 9bda8c3..da5b078 100644
--- a/lib/Json.cpp
+++ b/lib/Json.cpp
@@ -67,7 +67,7 @@ namespace json {
 
             ss << "\"" << escape_json(element.first) << "\": ";
 
-            const auto& value = element.second.data;
+            const auto& value = element.second.v;
             if (std::holds_alternative<std::string>(value)) {
                 ss << "\"" << escape_json(std::get<std::string>(value)) << "\"";
             }
diff --git a/lib/Json.h b/lib/Json.h
index 26da9a8..ee67f35 100644
--- a/lib/Json.h
+++ b/lib/Json.h
@@ -51,15 +51,7 @@ namespace json {
             size_t,
             ssize_t,
             bool,
-            std::nullopt_t> data;
-
-        template<class T>
-        value_t operator=(const T& map) {
-            value_t v;
-            v.data = map;
-            return v;
-        }
-
+            std::nullopt_t> v;
     };
 
     std::string map_to_json(const map_t& values);
diff --git a/lib/RemoteControl.cpp b/lib/RemoteControl.cpp
index b544461..fbe0662 100644
--- a/lib/RemoteControl.cpp
+++ b/lib/RemoteControl.cpp
@@ -109,7 +109,7 @@ std::list< std::vector<std::string> > RemoteControllers::get_param_list_values(c
 std::string RemoteControllers::get_showjson() {
     json::map_t root;
     for (auto &controllable : rcs.controllables) {
-        root[controllable->get_rc_name()].data = controllable->get_all_values();
+        root[controllable->get_rc_name()].v = controllable->get_all_values();
     }
 
     return json::map_to_json(root);
diff --git a/src/ConfigParser.cpp b/src/ConfigParser.cpp
index cb4dc24..68ee74b 100644
--- a/src/ConfigParser.cpp
+++ b/src/ConfigParser.cpp
@@ -37,6 +37,7 @@
 #include "ConfigParser.h"
 #include "Utils.h"
 #include "Log.h"
+#include "Events.h"
 #include "DabModulator.h"
 #include "output/SDR.h"
 
@@ -114,11 +115,16 @@ static void parse_configfile(
 
     mod_settings.inputTransport = pt.Get("input.transport", "file");
 
-    mod_settings.edi_max_delay_ms = pt.GetReal("input.edi_max_delay", 0.0f);
+    mod_settings.edi_max_delay_ms = pt.GetReal("input.edi_max_delay", 0.0);
 
     mod_settings.inputName = pt.Get("input.source", "/dev/stdin");
 
     // log parameters:
+    const string events_endpoint = pt.Get("log.events_endpoint", "");
+    if (not events_endpoint.empty()) {
+        events.bind(events_endpoint);
+    }
+
     if (pt.GetInteger("log.syslog", 0) == 1) {
         etiLog.register_backend(make_shared<LogToSyslog>());
     }
diff --git a/src/DabMod.cpp b/src/DabMod.cpp
index 805fab5..fdd9e93 100644
--- a/src/DabMod.cpp
+++ b/src/DabMod.cpp
@@ -47,6 +47,7 @@
 #   include <netinet/in.h>
 #endif
 
+#include "Events.h"
 #include "Utils.h"
 #include "Log.h"
 #include "DabModulator.h"
@@ -324,6 +325,8 @@ int launch_modulator(int argc, char* argv[])
     mod_settings_t mod_settings;
     parse_args(argc, argv, mod_settings);
 
+    etiLog.register_backend(make_shared<LogToEventSender>());
+
     etiLog.level(info) << "Configuration parsed. Starting up version " <<
 #if defined(GITVERSION)
             GITVERSION;
diff --git a/src/DabModulator.cpp b/src/DabModulator.cpp
index 0fe9c6d..4a29132 100644
--- a/src/DabModulator.cpp
+++ b/src/DabModulator.cpp
@@ -438,7 +438,7 @@ const string DabModulator::get_parameter(const string& parameter) const
 const json::map_t DabModulator::get_all_values() const
 {
     json::map_t map;
-    map["rate"] = m_settings.outputRate;
-    map["num_clipped_samples"] = m_formatConverter ? m_formatConverter->get_num_clipped_samples() : 0;
+    map["rate"].v = m_settings.outputRate;
+    map["num_clipped_samples"].v = m_formatConverter ? m_formatConverter->get_num_clipped_samples() : 0;
     return map;
 }
diff --git a/src/Events.cpp b/src/Events.cpp
new file mode 100644
index 0000000..d65b73a
--- /dev/null
+++ b/src/Events.cpp
@@ -0,0 +1,87 @@
+/*
+   Copyright (C) 2007, 2008, 2009, 2010, 2011, 2012
+   Her Majesty the Queen in Right of Canada (Communications Research
+   Center Canada)
+
+   Copyright (C) 2023
+   Matthias P. Braendli, matthias.braendli@mpb.li
+
+    http://www.opendigitalradio.org
+ */
+/*
+   This program is free software: you can redistribute it and/or modify
+   it under the terms of the GNU General Public License as published by
+   the Free Software Foundation, either version 3 of the License, or
+   (at your option) any later version.
+
+   This program is distributed in the hope that it will be useful,
+   but WITHOUT ANY WARRANTY; without even the implied warranty of
+   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+   GNU General Public License for more details.
+
+   You should have received a copy of the GNU General Public License
+   along with this program.  If not, see <https://www.gnu.org/licenses/>.
+ */
+#include <list>
+#include <string>
+#include <iostream>
+#include <sstream>
+#include <iomanip>
+#include <string>
+#include <algorithm>
+
+#include "Events.h"
+
+EventSender events;
+
+EventSender::EventSender() :
+    m_zmq_context(1),
+    m_socket(m_zmq_context, zmq::socket_type::pub)
+{
+    int linger = 2000;
+    m_socket.setsockopt(ZMQ_LINGER, &linger, sizeof(linger));
+}
+
+EventSender::~EventSender()
+{ }
+
+void EventSender::bind(const std::string& bind_endpoint)
+{
+    m_socket.bind(bind_endpoint);
+}
+
+void EventSender::send(const std::string& event_name, const json::map_t& detail)
+{
+    zmq::message_t zmsg1(event_name.data(), event_name.size());
+    const auto detail_json = json::map_to_json(detail);
+    zmq::message_t zmsg2(detail_json.data(), detail_json.size());
+
+    try {
+        m_socket.send(zmsg1, zmq::send_flags::sndmore);
+        m_socket.send(zmsg2, zmq::send_flags::none);
+    }
+    catch (const zmq::error_t& err) {
+        fprintf(stderr, "Cannot send event %s: %s", event_name.c_str(), err.what());
+    }
+}
+
+
+void LogToEventSender::log(log_level_t level, const std::string& message)
+{
+    std::string event_name;
+    if (level == log_level_t::warn) { event_name = "warn"; }
+    else if (level == log_level_t::error) { event_name = "error"; }
+    else if (level == log_level_t::alert) { event_name = "alert"; }
+    else if (level == log_level_t::emerg) { event_name = "emerg"; }
+
+    if (not event_name.empty()) {
+        json::map_t detail;
+        detail["message"].v = message;
+        events.send(event_name, detail);
+    }
+}
+
+std::string LogToEventSender::get_name() const
+{
+    return "EventSender";
+}
diff --git a/src/Events.h b/src/Events.h
new file mode 100644
index 0000000..215c5a8
--- /dev/null
+++ b/src/Events.h
@@ -0,0 +1,76 @@
+/*
+   Copyright (C) 2007, 2008, 2009, 2010, 2011, 2012
+   Her Majesty the Queen in Right of Canada (Communications Research
+   Center Canada)
+
+   Copyright (C) 2023
+   Matthias P. Braendli, matthias.braendli@mpb.li
+
+    http://www.opendigitalradio.org
+
+   This module adds remote-control capability to some of the dabmux/dabmod modules.
+ */
+/*
+   This program is free software: you can redistribute it and/or modify
+   it under the terms of the GNU General Public License as published by
+   the Free Software Foundation, either version 3 of the License, or
+   (at your option) any later version.
+
+   This program is distributed in the hope that it will be useful,
+   but WITHOUT ANY WARRANTY; without even the implied warranty of
+   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+   GNU General Public License for more details.
+
+   You should have received a copy of the GNU General Public License
+   along with this program.  If not, see <https://www.gnu.org/licenses/>.
+ */
+
+#pragma once
+
+#ifdef HAVE_CONFIG_H
+#  include "config.h"
+#endif
+
+#if defined(HAVE_ZEROMQ)
+#  include "zmq.hpp"
+#endif
+
+#include <list>
+#include <unordered_map>
+#include <variant>
+#include <map>
+#include <memory>
+#include <string>
+#include <stdexcept>
+
+#include "Log.h"
+#include "Json.h"
+
+class EventSender {
+    public:
+        EventSender();
+        EventSender(const EventSender& other) = delete;
+        const EventSender& operator=(const EventSender& other) = delete;
+        EventSender(EventSender&& other) = delete;
+        EventSender& operator=(EventSender&& other) = delete;
+        ~EventSender();
+
+        void bind(const std::string& bind_endpoint);
+
+        void send(const std::string& event_name, const json::map_t& detail);
+    private:
+        zmq::context_t m_zmq_context;
+        zmq::socket_t m_socket;
+};
+
+class LogToEventSender: public LogBackend {
+    public:
+        virtual ~LogToEventSender() {};
+        virtual void log(log_level_t level, const std::string& message);
+        virtual std::string get_name() const;
+};
+
+/* events is a singleton used in all parts of the program to output log messages.
+ * It is constructed in Events.cpp */
+extern EventSender events;
+
diff --git a/src/FIRFilter.cpp b/src/FIRFilter.cpp
index d2a6121..57e7127 100644
--- a/src/FIRFilter.cpp
+++ b/src/FIRFilter.cpp
@@ -350,7 +350,7 @@ const string FIRFilter::get_parameter(const string& parameter) const
 const json::map_t FIRFilter::get_all_values() const
 {
     json::map_t map;
-    map["ntaps"] = m_taps.size();
-    map["tapsfile"] = m_taps_file;
+    map["ntaps"].v = m_taps.size();
+    map["tapsfile"].v = m_taps_file;
     return map;
 }
diff --git a/src/GainControl.cpp b/src/GainControl.cpp
index beb93f6..84cf065 100644
--- a/src/GainControl.cpp
+++ b/src/GainControl.cpp
@@ -586,18 +586,18 @@ const string GainControl::get_parameter(const string& parameter) const
 const json::map_t GainControl::get_all_values() const
 {
     json::map_t map;
-    map["digital"] = m_digGain;
+    map["digital"].v = m_digGain;
     switch (m_gainmode) {
         case GainMode::GAIN_FIX:
-            map["mode"] = "fix";
+            map["mode"].v = "fix";
             break;
         case GainMode::GAIN_MAX:
-            map["mode"] = "max";
+            map["mode"].v = "max";
             break;
         case GainMode::GAIN_VAR:
-            map["mode"] = "var";
+            map["mode"].v = "var";
             break;
     }
-    map["var"] = m_var_variance_rc;
+    map["var"].v = m_var_variance_rc;
     return map;
 }
diff --git a/src/GuardIntervalInserter.cpp b/src/GuardIntervalInserter.cpp
index 80394b7..3c2db14 100644
--- a/src/GuardIntervalInserter.cpp
+++ b/src/GuardIntervalInserter.cpp
@@ -306,6 +306,6 @@ const std::string GuardIntervalInserter::get_parameter(const std::string& parame
 const json::map_t GuardIntervalInserter::get_all_values() const
 {
     json::map_t map;
-    map["windowlen"] = d_windowOverlap;
+    map["windowlen"].v = d_windowOverlap;
     return map;
 }
diff --git a/src/MemlessPoly.cpp b/src/MemlessPoly.cpp
index 30d4ce9..184b5bd 100644
--- a/src/MemlessPoly.cpp
+++ b/src/MemlessPoly.cpp
@@ -470,8 +470,8 @@ const string MemlessPoly::get_parameter(const string& parameter) const
 const json::map_t MemlessPoly::get_all_values() const
 {
     json::map_t map;
-    map["ncoefs"] = m_coefs_am.size();
-    map["coefs"] = serialise_coefficients();
-    map["coeffile"] = m_coefs_file;
+    map["ncoefs"].v = m_coefs_am.size();
+    map["coefs"].v = serialise_coefficients();
+    map["coeffile"].v = m_coefs_file;
     return map;
 }
diff --git a/src/TII.cpp b/src/TII.cpp
index 9068630..2656cbf 100644
--- a/src/TII.cpp
+++ b/src/TII.cpp
@@ -388,9 +388,9 @@ const std::string TII::get_parameter(const std::string& parameter) const
 const json::map_t TII::get_all_values() const
 {
     json::map_t map;
-    map["enable"] = m_conf.enable;
-    map["pattern"] = m_conf.pattern;
-    map["comb"] = m_conf.comb;
-    map["old_variant"] = m_conf.old_variant;
+    map["enable"].v = m_conf.enable;
+    map["pattern"].v = m_conf.pattern;
+    map["comb"].v = m_conf.comb;
+    map["old_variant"].v = m_conf.old_variant;
     return map;
 }
diff --git a/src/TimestampDecoder.cpp b/src/TimestampDecoder.cpp
index 4277e55..a7972c9 100644
--- a/src/TimestampDecoder.cpp
+++ b/src/TimestampDecoder.cpp
@@ -304,19 +304,19 @@ const std::string TimestampDecoder::get_parameter(
 const json::map_t TimestampDecoder::get_all_values() const
 {
     json::map_t map;
-    map["offset"] = timestamp_offset;
+    map["offset"].v = timestamp_offset;
     if (full_timestamp_received) {
-        map["timestamp"] = time_secs + ((double)time_pps / 16384000.0);
+        map["timestamp"].v = time_secs + ((double)time_pps / 16384000.0);
     }
     else {
-        map["timestamp"] = std::nullopt;
+        map["timestamp"].v = std::nullopt;
     }
 
     if (full_timestamp_received) {
-        map["timestamp0"] = time_secs_of_frame0 + ((double)time_pps_of_frame0 / 16384000.0);
+        map["timestamp0"].v = time_secs_of_frame0 + ((double)time_pps_of_frame0 / 16384000.0);
     }
     else {
-        map["timestamp0"] = std::nullopt;
+        map["timestamp0"].v = std::nullopt;
     }
     return map;
 }
diff --git a/src/output/Dexter.cpp b/src/output/Dexter.cpp
index 132636c..e52f774 100644
--- a/src/output/Dexter.cpp
+++ b/src/output/Dexter.cpp
@@ -470,20 +470,20 @@ SDRDevice::run_statistics_t Dexter::get_run_statistics(void) const
     run_statistics_t rs;
     {
         std::unique_lock<std::mutex> lock(m_attr_thread_mutex);
-        rs["underruns"] = underflows;
+        rs["underruns"].v = underflows;
     }
-    rs["latepackets"] = num_late;
-    rs["frames"] = num_frames_modulated;
+    rs["latepackets"].v = num_late;
+    rs["frames"].v = num_frames_modulated;
 
-    rs["in_holdover_since"] = 0;
+    rs["in_holdover_since"].v = 0;
     switch (m_clock_state) {
         case DexterClockState::Startup:
-            rs["clock_state"] = "startup"; break;
+            rs["clock_state"].v = "startup"; break;
         case DexterClockState::Normal:
-            rs["clock_state"] = "normal"; break;
+            rs["clock_state"].v = "normal"; break;
         case DexterClockState::Holdover:
-            rs["clock_state"] = "holdover";
-            rs["in_holdover_since"] = m_holdover_since_t;
+            rs["clock_state"].v = "holdover";
+            rs["in_holdover_since"].v = m_holdover_since_t;
             break;
     }
 
diff --git a/src/output/SDR.cpp b/src/output/SDR.cpp
index 4fc3277..6c03b53 100644
--- a/src/output/SDR.cpp
+++ b/src/output/SDR.cpp
@@ -450,7 +450,7 @@ const string SDR::get_parameter(const string& parameter) const
         if (m_device) {
             const auto stat = m_device->get_run_statistics();
             try {
-                const auto& value = stat.at(parameter).data;
+                const auto& value = stat.at(parameter).v;
                 if (std::holds_alternative<string>(value)) {
                     ss << std::get<string>(value);
                 }
@@ -489,19 +489,19 @@ const json::map_t SDR::get_all_values() const
 {
     json::map_t stat = m_device->get_run_statistics();
 
-    stat["txgain"] = m_config.txgain;
-    stat["rxgain"] = m_config.rxgain;
-    stat["freq"] = m_config.frequency;
-    stat["muting"] = m_config.muting;
-    stat["temp"] = std::nullopt;
+    stat["txgain"].v = m_config.txgain;
+    stat["rxgain"].v = m_config.rxgain;
+    stat["freq"].v = m_config.frequency;
+    stat["muting"].v = m_config.muting;
+    stat["temp"].v = std::nullopt;
 
     if (m_device) {
         const std::optional<double> temp = m_device->get_temperature();
         if (temp) {
-            stat["temp"] = *temp;
+            stat["temp"].v = *temp;
         }
     }
-    stat["queued_frames_ms"] = m_queue.size() *
+    stat["queued_frames_ms"].v = m_queue.size() *
             (size_t)chrono::duration_cast<chrono::milliseconds>(transmission_frame_duration(m_config.dabMode))
             .count();
 
diff --git a/src/output/Soapy.cpp b/src/output/Soapy.cpp
index 4d33e39..7931860 100644
--- a/src/output/Soapy.cpp
+++ b/src/output/Soapy.cpp
@@ -183,10 +183,10 @@ double Soapy::get_bandwidth(void) const
 SDRDevice::run_statistics_t Soapy::get_run_statistics(void) const
 {
     run_statistics_t rs;
-    rs["underruns"] = underflows;
-    rs["overruns"] = overflows;
-    rs["timeouts"] = timeouts;
-    rs["frames"] = num_frames_modulated;
+    rs["underruns"].v = underflows;
+    rs["overruns"].v = overflows;
+    rs["timeouts"].v = timeouts;
+    rs["frames"].v = num_frames_modulated;
     return rs;
 }
 
diff --git a/src/output/UHD.cpp b/src/output/UHD.cpp
index 6638b6c..094e021 100644
--- a/src/output/UHD.cpp
+++ b/src/output/UHD.cpp
@@ -380,19 +380,19 @@ void UHD::transmit_frame(struct FrameData&& frame)
 SDRDevice::run_statistics_t UHD::get_run_statistics(void) const
 {
     run_statistics_t rs;
-    rs["underruns"] = num_underflows;
-    rs["overruns"] = num_overflows;
-    rs["late_packets"] = num_late_packets;
-    rs["frames"] = num_frames_modulated;
+    rs["underruns"].v = num_underflows;
+    rs["overruns"].v = num_overflows;
+    rs["late_packets"].v = num_late_packets;
+    rs["frames"].v = num_frames_modulated;
 
     if (m_device_time) {
         const auto gpsdo_stat = m_device_time->get_gnss_stats();
-        rs["gpsdo_holdover"] = gpsdo_stat.holdover;
-        rs["gpsdo_num_sv"] = gpsdo_stat.num_sv;
+        rs["gpsdo_holdover"].v = gpsdo_stat.holdover;
+        rs["gpsdo_num_sv"].v = gpsdo_stat.num_sv;
     }
     else {
-        rs["gpsdo_holdover"] = true;
-        rs["gpsdo_num_sv"] = 0;
+        rs["gpsdo_holdover"].v = true;
+        rs["gpsdo_num_sv"].v = 0;
     }
     return rs;
 }