Common: update common code with EDI input changes

8 files changed, 122 insertions, 58 deletions

diff --git a/lib/edi/STIDecoder.cpp b/lib/edi/STIDecoder.cpp
index d55cc12..b6b9878 100644
--- a/lib/edi/STIDecoder.cpp
+++ b/lib/edi/STIDecoder.cpp
@@ -1,5 +1,5 @@
 /*
-   Copyright (C) 2019
+   Copyright (C) 2020
    Matthias P. Braendli, matthias.braendli@mpb.li
 
    http://opendigitalradio.org
@@ -30,9 +30,9 @@ namespace EdiDecoder {
 
 using namespace std;
 
-STIDecoder::STIDecoder(STIDataCollector& data_collector, bool verbose) :
+STIDecoder::STIDecoder(STIDataCollector& data_collector) :
     m_data_collector(data_collector),
-    m_dispatcher(std::bind(&STIDecoder::packet_completed, this), verbose)
+    m_dispatcher(std::bind(&STIDecoder::packet_completed, this))
 {
     using std::placeholders::_1;
     using std::placeholders::_2;
@@ -50,6 +50,11 @@ STIDecoder::STIDecoder(STIDataCollector& data_collector, bool verbose) :
             std::bind(&STIDecoder::decode_odrversion, this, _1, _2));
 }
 
+void STIDecoder::set_verbose(bool verbose)
+{
+    m_dispatcher.set_verbose(verbose);
+}
+
 void STIDecoder::push_bytes(const vector<uint8_t> &buf)
 {
     m_dispatcher.push_bytes(buf);
@@ -67,7 +72,7 @@ void STIDecoder::setMaxDelay(int num_af_packets)
 
 #define AFPACKET_HEADER_LEN 10 // includes SYNC
 
-bool STIDecoder::decode_starptr(const vector<uint8_t> &value, uint16_t)
+bool STIDecoder::decode_starptr(const std::vector<uint8_t>& value, const tag_name_t& n)
 {
     if (value.size() != 0x40 / 8) {
         etiLog.log(warn, "Incorrect length %02lx for *PTR", value.size());
@@ -87,7 +92,7 @@ bool STIDecoder::decode_starptr(const vector<uint8_t> &value, uint16_t)
     return true;
 }
 
-bool STIDecoder::decode_dsti(const vector<uint8_t> &value, uint16_t)
+bool STIDecoder::decode_dsti(const std::vector<uint8_t>& value, const tag_name_t& n)
 {
     size_t offset = 0;
 
@@ -156,10 +161,14 @@ bool STIDecoder::decode_dsti(const vector<uint8_t> &value, uint16_t)
     return true;
 }
 
-bool STIDecoder::decode_ssn(const vector<uint8_t> &value, uint16_t n)
+bool STIDecoder::decode_ssn(const std::vector<uint8_t>& value, const tag_name_t& name)
 {
     sti_payload_data sti;
 
+    uint16_t n = 0;
+    n = (uint16_t)(name[2]) << 8;
+    n |= (uint16_t)(name[3]);
+
     sti.stream_index = n - 1; // n is 1-indexed
     sti.rfa = value[0] >> 3;
     sti.tid = value[0] & 0x07;
@@ -182,12 +191,12 @@ bool STIDecoder::decode_ssn(const vector<uint8_t> &value, uint16_t n)
     return true;
 }
 
-bool STIDecoder::decode_stardmy(const vector<uint8_t>& /*value*/, uint16_t)
+bool STIDecoder::decode_stardmy(const std::vector<uint8_t>&, const tag_name_t&)
 {
     return true;
 }
 
-bool STIDecoder::decode_odraudiolevel(const vector<uint8_t>& value, uint16_t)
+bool STIDecoder::decode_odraudiolevel(const std::vector<uint8_t>& value, const tag_name_t& n)
 {
     constexpr size_t expected_length = 2 * sizeof(int16_t);
 
@@ -210,7 +219,7 @@ bool STIDecoder::decode_odraudiolevel(const vector<uint8_t>& value, uint16_t)
     return true;
 }
 
-bool STIDecoder::decode_odrversion(const vector<uint8_t>& value, uint16_t)
+bool STIDecoder::decode_odrversion(const std::vector<uint8_t>& value, const tag_name_t& n)
 {
     const auto vd = parse_odr_version_data(value);
     m_data_collector.update_odr_version(vd);
diff --git a/lib/edi/STIDecoder.hpp b/lib/edi/STIDecoder.hpp
index e2aa850..9d55728 100644
--- a/lib/edi/STIDecoder.hpp
+++ b/lib/edi/STIDecoder.hpp
@@ -1,5 +1,5 @@
 /*
-   Copyright (C) 2019
+   Copyright (C) 2020
    Matthias P. Braendli, matthias.braendli@mpb.li
 
    http://opendigitalradio.org
@@ -98,7 +98,9 @@ class STIDataCollector {
  */
 class STIDecoder {
     public:
-        STIDecoder(STIDataCollector& data_collector, bool verbose);
+        STIDecoder(STIDataCollector& data_collector);
+
+        void set_verbose(bool verbose);
 
         /* Push bytes into the decoder. The buf can contain more
          * than a single packet. This is useful when reading from streams
@@ -117,13 +119,13 @@ class STIDecoder {
         void setMaxDelay(int num_af_packets);
 
     private:
-        bool decode_starptr(const std::vector<uint8_t> &value, uint16_t);
-        bool decode_dsti(const std::vector<uint8_t> &value, uint16_t);
-        bool decode_ssn(const std::vector<uint8_t> &value, uint16_t n);
-        bool decode_stardmy(const std::vector<uint8_t> &value, uint16_t);
+        bool decode_starptr(const std::vector<uint8_t>& value, const tag_name_t& n);
+        bool decode_dsti(const std::vector<uint8_t>& value, const tag_name_t& n);
+        bool decode_ssn(const std::vector<uint8_t>& value, const tag_name_t& n);
+        bool decode_stardmy(const std::vector<uint8_t>& value, const tag_name_t& n);
 
-        bool decode_odraudiolevel(const std::vector<uint8_t> &value, uint16_t);
-        bool decode_odrversion(const std::vector<uint8_t> &value, uint16_t);
+        bool decode_odraudiolevel(const std::vector<uint8_t>& value, const tag_name_t& n);
+        bool decode_odrversion(const std::vector<uint8_t>& value, const tag_name_t& n);
 
         void packet_completed();
 
diff --git a/lib/edi/common.cpp b/lib/edi/common.cpp
index 470b3ba..3005802 100644
--- a/lib/edi/common.cpp
+++ b/lib/edi/common.cpp
@@ -1,5 +1,5 @@
 /*
-   Copyright (C) 2019
+   Copyright (C) 2020
    Matthias P. Braendli, matthias.braendli@mpb.li
 
    http://opendigitalradio.org
@@ -26,6 +26,7 @@
 #include <cassert>
 #include <cmath>
 #include <cstdio>
+#include <cctype>
 
 namespace EdiDecoder {
 
@@ -66,7 +67,7 @@ time_t frame_timestamp_t::to_unix_epoch() const
     return 946684800 + seconds - utco;
 }
 
-double frame_timestamp_t::diff_ms(const frame_timestamp_t& other) const
+double frame_timestamp_t::diff_s(const frame_timestamp_t& other) const
 {
     const double lhs = (double)seconds + (tsta / 16384000.0);
     const double rhs = (double)other.seconds + (other.tsta / 16384000.0);
@@ -111,10 +112,30 @@ std::chrono::system_clock::time_point frame_timestamp_t::to_system_clock() const
     return ts;
 }
 
+std::string tag_name_to_human_readable(const tag_name_t& name)
+{
+    std::string s;
+    for (const uint8_t c : name) {
+        if (isprint(c)) {
+            s += (char)c;
+        }
+        else {
+            char escaped[5];
+            snprintf(escaped, 5, "\\x%02x", c);
+            s += escaped;
+        }
+    }
+    return s;
+}
 
 TagDispatcher::TagDispatcher(
-        std::function<void()>&& af_packet_completed, bool verbose) :
-    m_af_packet_completed(move(af_packet_completed))
+        std::function<void()>&& af_packet_completed) :
+    m_af_packet_completed(move(af_packet_completed)),
+    m_tagpacket_handler([](const std::vector<uint8_t>& ignore){})
+{
+}
+
+void TagDispatcher::set_verbose(bool verbose)
 {
     m_pft.setVerbose(verbose);
 }
@@ -295,6 +316,11 @@ void TagDispatcher::register_tag(const std::string& tag, tag_handler&& h)
     m_handlers[tag] = move(h);
 }
 
+void TagDispatcher::register_tagpacket_handler(tagpacket_handler&& h)
+{
+    m_tagpacket_handler = move(h);
+}
+
 
 bool TagDispatcher::decode_tagpacket(const vector<uint8_t> &payload)
 {
@@ -326,31 +352,23 @@ bool TagDispatcher::decode_tagpacket(const vector<uint8_t> &payload)
             break;
         }
 
+        const array<uint8_t, 4> tag_name({
+               (uint8_t)tag_sz[0], (uint8_t)tag_sz[1], (uint8_t)tag_sz[2], (uint8_t)tag_sz[3]
+               });
         vector<uint8_t> tag_value(taglength);
         copy(   payload.begin() + i+8,
                 payload.begin() + i+8+taglength,
                 tag_value.begin());
 
-        bool tagsuccess = false;
+        bool tagsuccess = true;
         bool found = false;
         for (auto tag_handler : m_handlers) {
-            if (tag_handler.first.size() == 4 and tag_handler.first == tag) {
-                found = true;
-                tagsuccess = tag_handler.second(tag_value, 0);
-            }
-            else if (tag_handler.first.size() == 3 and
-                    tag.substr(0, 3) == tag_handler.first) {
-                found = true;
-                uint8_t n = tag_sz[3];
-                tagsuccess = tag_handler.second(tag_value, n);
-            }
-            else if (tag_handler.first.size() == 2 and
-                    tag.substr(0, 2) == tag_handler.first) {
+            if (    (tag_handler.first.size() == 4 and tag == tag_handler.first) or
+                    (tag_handler.first.size() == 3 and tag.substr(0, 3) == tag_handler.first) or
+                    (tag_handler.first.size() == 2 and tag.substr(0, 2) == tag_handler.first) or
+                    (tag_handler.first.size() == 1 and tag.substr(0, 1) == tag_handler.first)) {
                 found = true;
-                uint16_t n = 0;
-                n = (uint16_t)(tag_sz[2]) << 8;
-                n |= (uint16_t)(tag_sz[3]);
-                tagsuccess = tag_handler.second(tag_value, n);
+                tagsuccess &= tag_handler.second(tag_value, tag_name);
             }
         }
 
@@ -366,6 +384,8 @@ bool TagDispatcher::decode_tagpacket(const vector<uint8_t> &payload)
         }
     }
 
+    m_tagpacket_handler(payload);
+
     return success;
 }
 
diff --git a/lib/edi/common.hpp b/lib/edi/common.hpp
index 2a9c683..498b28a 100644
--- a/lib/edi/common.hpp
+++ b/lib/edi/common.hpp
@@ -1,5 +1,5 @@
 /*
-   Copyright (C) 2019
+   Copyright (C) 2020
    Matthias P. Braendli, matthias.braendli@mpb.li
 
    http://opendigitalradio.org
@@ -41,7 +41,7 @@ struct frame_timestamp_t {
     std::time_t to_unix_epoch() const;
     std::chrono::system_clock::time_point to_system_clock() const;
 
-    double diff_ms(const frame_timestamp_t& other) const;
+    double diff_s(const frame_timestamp_t& other) const;
 
     frame_timestamp_t& operator+=(const std::chrono::milliseconds& ms);
 
@@ -55,6 +55,10 @@ struct decode_state_t {
     size_t num_bytes_consumed;
 };
 
+using tag_name_t = std::array<uint8_t, 4>;
+
+std::string tag_name_to_human_readable(const tag_name_t& name);
+
 /* The TagDispatcher takes care of decoding EDI, with or without PFT, and
  * will call functions when TAGs are encountered.
  *
@@ -63,7 +67,10 @@ struct decode_state_t {
  */
 class TagDispatcher {
     public:
-        TagDispatcher(std::function<void()>&& af_packet_completed, bool verbose);
+        TagDispatcher(std::function<void()>&& af_packet_completed);
+
+        void set_verbose(bool verbose);
+
 
         /* Push bytes into the decoder. The buf can contain more
          * than a single packet. This is useful when reading from streams
@@ -81,9 +88,19 @@ class TagDispatcher {
          */
         void setMaxDelay(int num_af_packets);
 
-        using tag_handler = std::function<bool(std::vector<uint8_t>, uint16_t)>;
+        /* Handler function for a tag. The first argument contains the tag value,
+         * the second argument contains the tag name */
+        using tag_handler = std::function<bool(const std::vector<uint8_t>&, const tag_name_t&)>;
+
+        /* Register a handler for a tag. If the tag string can be length 0, 1, 2, 3 or 4.
+         * If is shorter than 4, it will perform a longest match on the tag name.
+         */
         void register_tag(const std::string& tag, tag_handler&& h);
 
+        /* The complete tagpacket can also be retrieved */
+        using tagpacket_handler = std::function<void(const std::vector<uint8_t>&)>;
+        void register_tagpacket_handler(tagpacket_handler&& h);
+
     private:
         decode_state_t decode_afpacket(const std::vector<uint8_t> &input_data);
         bool decode_tagpacket(const std::vector<uint8_t> &payload);
@@ -93,6 +110,7 @@ class TagDispatcher {
         std::vector<uint8_t> m_input_data;
         std::map<std::string, tag_handler> m_handlers;
         std::function<void()> m_af_packet_completed;
+        tagpacket_handler m_tagpacket_handler;
 };
 
 // Data carried inside the ODRv EDI TAG
diff --git a/lib/edioutput/TagPacket.cpp b/lib/edioutput/TagPacket.cpp
index b0bf9a1..ec52ad7 100644
--- a/lib/edioutput/TagPacket.cpp
+++ b/lib/edioutput/TagPacket.cpp
@@ -1,5 +1,5 @@
 /*
-   Copyright (C) 2014
+   Copyright (C) 2020
    Matthias P. Braendli, matthias.braendli@mpb.li
 
     http://www.opendigitalradio.org
@@ -41,17 +41,19 @@ TagPacket::TagPacket(unsigned int alignment) : m_alignment(alignment)
 
 std::vector<uint8_t> TagPacket::Assemble()
 {
-    std::list<TagItem*>::iterator tag;
+    if (raw_tagpacket.size() > 0 and tag_items.size() > 0) {
+        throw std::logic_error("TagPacket: both raw and items used!");
+    }
 
-    std::vector<uint8_t> packet;
+    if (raw_tagpacket.size() > 0) {
+        return raw_tagpacket;
+    }
 
-    //std::cerr << "Assemble TAGPacket" << std::endl;
+    std::vector<uint8_t> packet;
 
-    for (tag = tag_items.begin(); tag != tag_items.end(); ++tag) {
-        std::vector<uint8_t> tag_data = (*tag)->Assemble();
+    for (auto tag : tag_items) {
+        std::vector<uint8_t> tag_data = tag->Assemble();
         packet.insert(packet.end(), tag_data.begin(), tag_data.end());
-
-        //std::cerr << "     Add TAGItem of length " << tag_data.size() << std::endl;
     }
 
     if (m_alignment == 0) { /* no padding */ }
diff --git a/lib/edioutput/TagPacket.h b/lib/edioutput/TagPacket.h
index 1e40ce7..b53b718 100644
--- a/lib/edioutput/TagPacket.h
+++ b/lib/edioutput/TagPacket.h
@@ -1,5 +1,5 @@
 /*
-   Copyright (C) 2014
+   Copyright (C) 2020
    Matthias P. Braendli, matthias.braendli@mpb.li
 
     http://www.opendigitalradio.org
@@ -39,6 +39,9 @@ namespace edi {
 // Assemble function that puts the bytestream together and adds
 // padding such that the total length is a multiple of 8 Bytes.
 //
+// Alternatively, a raw tagpacket can be used instead of the
+// items list
+//
 // ETSI TS 102 821, 5.1 Tag Packet
 class TagPacket
 {
@@ -48,6 +51,8 @@ class TagPacket
 
         std::list<TagItem*> tag_items;
 
+        std::vector<uint8_t> raw_tagpacket;
+
     private:
         unsigned int m_alignment;
 };
diff --git a/lib/edioutput/Transport.cpp b/lib/edioutput/Transport.cpp
index 4c91483..d8627fd 100644
--- a/lib/edioutput/Transport.cpp
+++ b/lib/edioutput/Transport.cpp
@@ -1,5 +1,5 @@
 /*
-   Copyright (C) 2019
+   Copyright (C) 2020
    Matthias P. Braendli, matthias.braendli@mpb.li
 
     http://www.opendigitalradio.org
@@ -33,7 +33,7 @@ namespace edi {
 
 void configuration_t::print() const
 {
-    etiLog.level(info) << "EDI";
+    etiLog.level(info) << "EDI Output";
     etiLog.level(info) << " verbose     " << verbose;
     for (auto edi_dest : destinations) {
         if (auto udp_dest = dynamic_pointer_cast<edi::udp_destination_t>(edi_dest)) {
@@ -67,7 +67,7 @@ Sender::Sender(const configuration_t& conf) :
     edi_pft(m_conf)
 {
     if (m_conf.verbose) {
-        etiLog.log(info, "Setup EDI");
+        etiLog.log(info, "Setup EDI Output");
     }
 
     for (const auto& edi_dest : m_conf.destinations) {
@@ -104,7 +104,7 @@ Sender::Sender(const configuration_t& conf) :
     }
 
     if (m_conf.verbose) {
-        etiLog.log(info, "EDI set up");
+        etiLog.log(info, "EDI output set up");
     }
 }
 
@@ -118,7 +118,7 @@ void Sender::write(const TagPacket& tagpacket)
         vector<edi::PFTFragment> edi_fragments = edi_pft.Assemble(af_packet);
 
         if (m_conf.verbose) {
-            fprintf(stderr, "EDI number of PFT fragment before interleaver %zu\n",
+            fprintf(stderr, "EDI Output: Number of PFT fragment before interleaver %zu\n",
                     edi_fragments.size());
         }
 
@@ -127,7 +127,7 @@ void Sender::write(const TagPacket& tagpacket)
         }
 
         if (m_conf.verbose) {
-            fprintf(stderr, "EDI number of PFT fragments %zu\n",
+            fprintf(stderr, "EDI Output: Number of PFT fragments %zu\n",
                     edi_fragments.size());
         }
 
@@ -169,6 +169,12 @@ void Sender::write(const TagPacket& tagpacket)
                 Socket::InetAddress addr;
                 addr.resolveUdpDestination(udp_dest->dest_addr, m_conf.dest_port);
 
+                if (af_packet.size() > 1400 and not m_udp_fragmentation_warning_printed) {
+                    fprintf(stderr, "EDI Output: AF packet larger than 1400,"
+                            " consider using PFT to avoid UP fragmentation.\n");
+                    m_udp_fragmentation_warning_printed = true;
+                }
+
                 udp_sockets.at(udp_dest.get())->send(af_packet, addr);
             }
             else if (auto tcp_dest = dynamic_pointer_cast<edi::tcp_server_t>(dest)) {
diff --git a/lib/edioutput/Transport.h b/lib/edioutput/Transport.h
index 73b2ab6..56ded3b 100644
--- a/lib/edioutput/Transport.h
+++ b/lib/edioutput/Transport.h
@@ -1,5 +1,5 @@
 /*
-   Copyright (C) 2019
+   Copyright (C) 2020
    Matthias P. Braendli, matthias.braendli@mpb.li
 
     http://www.opendigitalradio.org
@@ -50,6 +50,8 @@ class Sender {
         void write(const TagPacket& tagpacket);
 
     private:
+        bool m_udp_fragmentation_warning_printed = false;
+
         configuration_t m_conf;
         std::ofstream edi_debug_file;