Add timestamp offset to management server for EDI input

4 files changed, 24 insertions, 6 deletions

diff --git a/doc/show_dabmux_stats.py b/doc/show_dabmux_stats.py
index 5bb6605..12cad1e 100755
--- a/doc/show_dabmux_stats.py
+++ b/doc/show_dabmux_stats.py
@@ -46,7 +46,7 @@ if len(sys.argv) == 1:
             data = sock.recv().decode("utf-8")
             values = json.loads(data)['values']
 
-            tmpl = "{ident:20}{maxfill:>8}{minfill:>8}{under:>8}{over:>8}{audioleft:>8}{audioright:>8}{peakleft:>8}{peakright:>8}{state:>16}{version:>48}{uptime:>8}"
+            tmpl = "{ident:20}{maxfill:>8}{minfill:>8}{under:>8}{over:>8}{audioleft:>8}{audioright:>8}{peakleft:>8}{peakright:>8}{state:>16}{version:>48}{uptime:>8}{offset:>8}"
             print(tmpl.format(
                 ident="id",
                 maxfill="max",
@@ -59,7 +59,8 @@ if len(sys.argv) == 1:
                 peakright="peak R",
                 state="state",
                 version="version",
-                uptime="uptime"))
+                uptime="uptime",
+                offset="offset"))
 
             for ident in values:
                 v = values[ident]['inputstat']
@@ -85,7 +86,8 @@ if len(sys.argv) == 1:
                     peakright=v['peak_right_slow'],
                     state=v['state'],
                     version=v['version'],
-                    uptime=v['uptime']))
+                    uptime=v['uptime'],
+                    offset=v['last_tist_offset']))
 
 
 elif len(sys.argv) == 2 and sys.argv[1] == "config":
diff --git a/src/ManagementServer.cpp b/src/ManagementServer.cpp
index f3b72d9..599d744 100644
--- a/src/ManagementServer.cpp
+++ b/src/ManagementServer.cpp
@@ -336,7 +336,8 @@ void ManagementServer::handle_message(zmq::message_t& zmq_message)
         }
 
         std::string answerstr(answer.str());
-        m_zmq_sock.send(answerstr.c_str(), answerstr.size());
+        zmq::const_buffer message(answerstr.data(), answerstr.size());
+        m_zmq_sock.send(message, zmq::send_flags::none);
     }
     catch (const std::exception& e) {
         etiLog.level(error) <<
@@ -382,6 +383,13 @@ void InputStat::notifyBuffer(long bufsize)
     prune_statistics(time_now);
 }
 
+void InputStat::notifyTimestampOffset(double offset)
+{
+    unique_lock<mutex> lock(m_mutex);
+
+    m_last_tist_offset = offset;
+}
+
 void InputStat::notifyPeakLevels(int peak_left, int peak_right)
 {
     unique_lock<mutex> lock(m_mutex);
@@ -558,6 +566,7 @@ std::string InputStat::encodeValuesJSON()
         "\"peak_right_slow\": " << to_dB(peak_right) << ", "
         "\"num_underruns\": " << m_num_underruns << ", "
         "\"num_overruns\": " << m_num_overruns << ", "
+        "\"last_tist_offset\": " << m_last_tist_offset << ", "
         "\"version\": \"" << version << "\", "
         "\"uptime\": " << m_uptime_s << ", "
         ;
diff --git a/src/ManagementServer.h b/src/ManagementServer.h
index e88d233..12f5ad6 100644
--- a/src/ManagementServer.h
+++ b/src/ManagementServer.h
@@ -97,6 +97,7 @@ class InputStat
         /* This function is called for every frame read by
          * the multiplexer */
         void notifyBuffer(long bufsize);
+        void notifyTimestampOffset(double offset);
         void notifyPeakLevels(int peak_left, int peak_right);
         void notifyUnderrun(void);
         void notifyOverrun(void);
@@ -123,6 +124,9 @@ class InputStat
         uint32_t m_num_underruns = 0;
         uint32_t m_num_overruns = 0;
 
+        // last measured timestamp offset
+        double m_last_tist_offset = 0;
+
         // Peak audio levels (linear 16-bit PCM) for the two channels.
         // Keep a FIFO of values from the last minutes, apply
         // a short window to also see short-term fluctuations.
diff --git a/src/input/Edi.cpp b/src/input/Edi.cpp
index e6a7e3e..2aa776f 100644
--- a/src/input/Edi.cpp
+++ b/src/input/Edi.cpp
@@ -117,6 +117,7 @@ size_t Edi::readFrame(uint8_t *buffer, size_t size)
 {
     // Save stats data in bytes, not in frames
     m_stats.notifyBuffer(m_frames.size() * size);
+    m_stats.notifyTimestampOffset(0);
 
     EdiDecoder::sti_frame_t sti;
     if (m_is_prebuffering) {
@@ -221,6 +222,7 @@ size_t Edi::readFrame(uint8_t *buffer, size_t size, std::time_t seconds, int utc
                     auto ts_req = EdiDecoder::frame_timestamp_t::from_unix_epoch(seconds, utco, tsta);
                     ts_req += m_tist_delay;
                     const double offset = ts_req.diff_s(m_pending_sti_frame.timestamp);
+                    m_stats.notifyTimestampOffset(offset);
 
                     if (offset < 0) {
                         // Too far in the future
@@ -265,12 +267,12 @@ size_t Edi::readFrame(uint8_t *buffer, size_t size, std::time_t seconds, int utc
 
         if (num_discarded_wrong_size > 0) {
             etiLog.level(warn) << "EDI input " << m_name << ": " <<
-                num_discarded_wrong_size << "packets with wrong size.";
+                num_discarded_wrong_size << " packets with wrong size.";
         }
 
         if (num_discarded_invalid_ts > 0) {
             etiLog.level(warn) << "EDI input " << m_name << ": " <<
-                num_discarded_wrong_size << "packets with invalid timestamp.";
+                num_discarded_wrong_size << " packets with invalid timestamp.";
         }
 
         memset(buffer, 0, size);
@@ -298,6 +300,7 @@ size_t Edi::readFrame(uint8_t *buffer, size_t size, std::time_t seconds, int utc
             auto ts_req = EdiDecoder::frame_timestamp_t::from_unix_epoch(seconds, utco, tsta);
             ts_req += m_tist_delay;
             const double offset = m_pending_sti_frame.timestamp.diff_s(ts_req);
+            m_stats.notifyTimestampOffset(offset);
 
             if (-24e-3 < offset and offset <= 0) {
                 if (not m_pending_sti_frame.version_data.version.empty()) {