EDI: put more code in common between DabMux and ZMQ2EDI

6 files changed, 343 insertions, 48 deletions

diff --git a/src/dabOutput/edi/Config.h b/src/dabOutput/edi/Config.h
new file mode 100644
index 0000000..d3678d9
--- /dev/null
+++ b/src/dabOutput/edi/Config.h
@@ -0,0 +1,71 @@
+/*
+   Copyright (C) 2019
+   Matthias P. Braendli, matthias.braendli@mpb.li
+
+    http://www.opendigitalradio.org
+
+   EDI output,
+   UDP and TCP transports and their configuration
+
+   */
+/*
+   This file is part of ODR-DabMux.
+
+   ODR-DabMux 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.
+
+   ODR-DabMux 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 ODR-DabMux.  If not, see <http://www.gnu.org/licenses/>.
+   */
+
+#pragma once
+
+#include "config.h"
+#include <vector>
+#include <string>
+#include <memory>
+#include <cstdint>
+
+namespace edi {
+
+/** Configuration for EDI output */
+
+struct destination_t {
+    virtual ~destination_t() {};
+};
+
+// Can represent both unicast and multicast destinations
+struct udp_destination_t : public destination_t {
+    std::string dest_addr;
+    std::string source_addr;
+    unsigned int source_port = 0;
+    unsigned int ttl = 10;
+};
+
+struct configuration_t {
+    unsigned chunk_len = 207;        // RSk, data length of each chunk
+    unsigned fec       = 0;          // number of fragments that can be recovered
+    bool dump          = false;      // dump a file with the EDI packets
+    bool verbose       = false;
+    bool enable_pft    = false;      // Enable protection and fragmentation
+    unsigned int tagpacket_alignment = 0;
+    std::vector<std::shared_ptr<destination_t> > destinations;
+    unsigned int dest_port = 0;      // common destination port, because it's encoded in the transport layer
+    unsigned int latency_frames = 0; // if nonzero, enable interleaver with a latency of latency_frames * 24ms
+
+    bool enabled() const { return destinations.size() > 0; }
+    bool interleaver_enabled() const { return latency_frames > 0; }
+
+    void print() const;
+};
+
+}
+
+
diff --git a/src/dabOutput/edi/Interleaver.cpp b/src/dabOutput/edi/Interleaver.cpp
index e31326b..f26a50e 100644
--- a/src/dabOutput/edi/Interleaver.cpp
+++ b/src/dabOutput/edi/Interleaver.cpp
@@ -30,6 +30,7 @@
    */
 
 #include "Interleaver.h"
+#include <cassert>
 
 namespace edi {
 
diff --git a/src/dabOutput/edi/PFT.cpp b/src/dabOutput/edi/PFT.cpp
index 1c885f9..5b93016 100644
--- a/src/dabOutput/edi/PFT.cpp
+++ b/src/dabOutput/edi/PFT.cpp
@@ -1,5 +1,5 @@
 /*
-   Copyright (C) 2014
+   Copyright (C) 2019
    Matthias P. Braendli, matthias.braendli@mpb.li
 
     http://www.opendigitalradio.org
@@ -50,6 +50,30 @@ using namespace std;
 // An integer division that rounds up, i.e. ceil(a/b)
 #define CEIL_DIV(a, b) (a % b == 0  ? a / b : a / b + 1)
 
+PFT::PFT() { }
+
+PFT::PFT(const configuration_t &conf) :
+    m_k(conf.chunk_len),
+    m_m(conf.fec),
+    m_dest_port(conf.dest_port),
+    m_pseq(0),
+    m_num_chunks(0),
+    m_verbose(conf.verbose)
+    {
+        if (m_k > 207) {
+            etiLog.level(warn) <<
+                "EDI PFT: maximum chunk size is 207.";
+            throw std::out_of_range("EDI PFT Chunk size too large.");
+        }
+
+        if (m_m > 5) {
+            etiLog.level(warn) <<
+                "EDI PFT: high number of recoverable fragments"
+                " may lead to large overhead";
+            // See TS 102 821, 7.2.1 Known values, list entry for 'm'
+        }
+    }
+
 RSBlock PFT::Protect(AFPacket af_packet)
 {
     RSBlock rs_block;
@@ -98,7 +122,7 @@ RSBlock PFT::Protect(AFPacket af_packet)
     // Calculate RS for each chunk and assemble RS block
     for (size_t i = 0; i < af_packet.size(); i+= chunk_len) {
         vector<uint8_t> chunk(207);
-        vector<uint8_t> protection(ParityBytes);
+        vector<uint8_t> protection(PARITYBYTES);
 
         // copy chunk_len bytes into new chunk
         memcpy(&chunk.front(), &af_packet[i], chunk_len);
@@ -139,7 +163,7 @@ vector< vector<uint8_t> > PFT::ProtectAndFragment(AFPacket af_packet)
 #endif
 
         // TS 102 821 7.2.2: s_max = MIN(floor(c*p/(m+1)), MTU - h))
-        const size_t max_payload_size = ( m_num_chunks * ParityBytes ) / (m_m + 1);
+        const size_t max_payload_size = ( m_num_chunks * PARITYBYTES ) / (m_m + 1);
 
         // Calculate fragment count and size
         // TS 102 821 7.2.2: ceil((l + c*p + z) / s_max)
diff --git a/src/dabOutput/edi/PFT.h b/src/dabOutput/edi/PFT.h
index 05afdb1..4076bf3 100644
--- a/src/dabOutput/edi/PFT.h
+++ b/src/dabOutput/edi/PFT.h
@@ -1,5 +1,5 @@
 /*
-   Copyright (C) 2014
+   Copyright (C) 2019
    Matthias P. Braendli, matthias.braendli@mpb.li
 
     http://www.opendigitalradio.org
@@ -40,7 +40,7 @@
 #include "AFPacket.h"
 #include "Log.h"
 #include "ReedSolomon.h"
-#include "dabOutput/dabOutput.h"
+#include "dabOutput/edi/Config.h"
 
 namespace edi {
 
@@ -50,38 +50,10 @@ typedef std::vector<uint8_t> PFTFragment;
 class PFT
 {
     public:
-        static const int ParityBytes = 48;
-
-        PFT() :
-            m_k(207),
-            m_m(3),
-            m_dest_port(12000),
-            m_pseq(0),
-            m_num_chunks(0),
-            m_verbose(false)
-        { }
-
-        PFT(const edi_configuration_t &conf) :
-            m_k(conf.chunk_len),
-            m_m(conf.fec),
-            m_dest_port(conf.dest_port),
-            m_pseq(0),
-            m_num_chunks(0),
-            m_verbose(conf.verbose)
-        {
-            if (m_k > 207) {
-                etiLog.level(warn) <<
-                        "EDI PFT: maximum chunk size is 207.";
-                throw std::out_of_range("EDI PFT Chunk size too large.");
-            }
-
-            if (m_m > 5) {
-                etiLog.level(warn) <<
-                    "EDI PFT: high number of recoverable fragments"
-                    " may lead to large overhead";
-                // See TS 102 821, 7.2.1 Known values, list entry for 'm'
-            }
-        }
+        static constexpr int PARITYBYTES = 48;
+
+        PFT();
+        PFT(const configuration_t& conf);
 
         // return a list of PFT fragments with the correct
         // PFT headers
@@ -94,17 +66,12 @@ class PFT
         std::vector< std::vector<uint8_t> > ProtectAndFragment(AFPacket af_packet);
 
     private:
-        unsigned int m_k; // length of RS data word
-        unsigned int m_m; // number of fragments that can be recovered if lost
-
-        unsigned int m_dest_port; // Destination port for transport header
-
-        uint16_t m_pseq;
-
-        size_t m_num_chunks;
-
-        bool m_verbose;
-
+        unsigned int m_k = 207; // length of RS data word
+        unsigned int m_m = 3; // number of fragments that can be recovered if lost
+        unsigned int m_dest_port = 12000; // Destination port for transport header
+        uint16_t m_pseq = 0;
+        size_t m_num_chunks = 0;
+        bool m_verbose = 0;
 };
 
 }
diff --git a/src/dabOutput/edi/Transport.cpp b/src/dabOutput/edi/Transport.cpp
new file mode 100644
index 0000000..d433239
--- /dev/null
+++ b/src/dabOutput/edi/Transport.cpp
@@ -0,0 +1,164 @@
+/*
+   Copyright (C) 2019
+   Matthias P. Braendli, matthias.braendli@mpb.li
+
+    http://www.opendigitalradio.org
+
+   EDI output,
+   UDP and TCP transports and their configuration
+
+   */
+/*
+   This file is part of ODR-DabMux.
+
+   ODR-DabMux 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.
+
+   ODR-DabMux 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 ODR-DabMux.  If not, see <http://www.gnu.org/licenses/>.
+   */
+
+#include "Transport.h"
+#include <iterator>
+
+using namespace std;
+
+namespace edi {
+
+void configuration_t::print() const
+{
+    etiLog.level(info) << "EDI";
+    etiLog.level(info) << " verbose     " << verbose;
+    for (auto edi_dest : destinations) {
+        if (auto udp_dest = dynamic_pointer_cast<edi::udp_destination_t>(edi_dest)) {
+            etiLog.level(info) << " to " << udp_dest->dest_addr << ":" << dest_port;
+            if (not udp_dest->source_addr.empty()) {
+                etiLog.level(info) << "  source      " << udp_dest->source_addr;
+                etiLog.level(info) << "  ttl         " << udp_dest->ttl;
+            }
+            etiLog.level(info) << "  source port " << udp_dest->source_port;
+        }
+        else {
+            throw std::logic_error("EDI destination not implemented");
+        }
+    }
+    if (interleaver_enabled()) {
+        etiLog.level(info) << " interleave     " << latency_frames * 24 << " ms";
+    }
+}
+
+
+Sender::Sender(const configuration_t& conf) :
+    m_conf(conf),
+    edi_pft(m_conf)
+{
+    if (m_conf.verbose) {
+        etiLog.log(info, "Setup EDI");
+    }
+
+    for (const auto& edi_dest : m_conf.destinations) {
+        if (const auto udp_dest = dynamic_pointer_cast<edi::udp_destination_t>(edi_dest)) {
+            auto udp_socket = std::make_shared<UdpSocket>(udp_dest->source_port);
+
+            if (not udp_dest->source_addr.empty()) {
+                int err = udp_socket->setMulticastSource(udp_dest->source_addr.c_str());
+                if (err) {
+                    throw runtime_error("EDI socket set source failed!");
+                }
+                err = udp_socket->setMulticastTTL(udp_dest->ttl);
+                if (err) {
+                    throw runtime_error("EDI socket set TTL failed!");
+                }
+            }
+
+            udp_sockets.emplace(udp_dest.get(), udp_socket);
+        }
+    }
+
+    if (m_conf.interleaver_enabled()) {
+        edi_interleaver.SetLatency(m_conf.latency_frames);
+    }
+
+    if (m_conf.dump) {
+        edi_debug_file.open("./edi.debug");
+    }
+
+    if (m_conf.verbose) {
+        etiLog.log(info, "EDI set up");
+    }
+}
+
+void Sender::write(const TagPacket& tagpacket)
+{
+    // Assemble into one AF Packet
+    edi::AFPacket af_packet = edi_afPacketiser.Assemble(tagpacket);
+
+    if (m_conf.enable_pft) {
+        // Apply PFT layer to AF Packet (Reed Solomon FEC and Fragmentation)
+        vector<edi::PFTFragment> edi_fragments = edi_pft.Assemble(af_packet);
+
+        if (m_conf.verbose) {
+            fprintf(stderr, "EDI number of PFT fragment before interleaver %zu",
+                    edi_fragments.size());
+        }
+
+        if (m_conf.interleaver_enabled()) {
+            edi_fragments = edi_interleaver.Interleave(edi_fragments);
+        }
+
+        // Send over ethernet
+        for (const auto& edi_frag : edi_fragments) {
+            for (auto& dest : m_conf.destinations) {
+                if (const auto& udp_dest = dynamic_pointer_cast<edi::udp_destination_t>(dest)) {
+                    InetAddress addr;
+                    addr.setAddress(udp_dest->dest_addr.c_str());
+                    addr.setPort(m_conf.dest_port);
+
+                    udp_sockets.at(udp_dest.get())->send(edi_frag, addr);
+                }
+                else {
+                    throw std::logic_error("EDI destination not implemented");
+                }
+            }
+
+            if (m_conf.dump) {
+                std::ostream_iterator<uint8_t> debug_iterator(edi_debug_file);
+                std::copy(edi_frag.begin(), edi_frag.end(), debug_iterator);
+            }
+        }
+
+        if (m_conf.verbose) {
+            fprintf(stderr, "EDI number of PFT fragments %zu",
+                    edi_fragments.size());
+        }
+    }
+    else {
+        // Send over ethernet
+        for (auto& dest : m_conf.destinations) {
+            if (const auto& udp_dest = dynamic_pointer_cast<edi::udp_destination_t>(dest)) {
+                InetAddress addr;
+                addr.setAddress(udp_dest->dest_addr.c_str());
+                addr.setPort(m_conf.dest_port);
+
+                udp_sockets.at(udp_dest.get())->send(af_packet, addr);
+            }
+            else {
+                throw std::logic_error("EDI destination not implemented");
+            }
+        }
+
+        if (m_conf.dump) {
+            std::ostream_iterator<uint8_t> debug_iterator(edi_debug_file);
+            std::copy(af_packet.begin(), af_packet.end(), debug_iterator);
+        }
+    }
+}
+
+}
diff --git a/src/dabOutput/edi/Transport.h b/src/dabOutput/edi/Transport.h
new file mode 100644
index 0000000..3c48c96
--- /dev/null
+++ b/src/dabOutput/edi/Transport.h
@@ -0,0 +1,68 @@
+/*
+   Copyright (C) 2019
+   Matthias P. Braendli, matthias.braendli@mpb.li
+
+    http://www.opendigitalradio.org
+
+   EDI output,
+   UDP and TCP transports and their configuration
+
+   */
+/*
+   This file is part of ODR-DabMux.
+
+   ODR-DabMux 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.
+
+   ODR-DabMux 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 ODR-DabMux.  If not, see <http://www.gnu.org/licenses/>.
+   */
+
+#pragma once
+
+#include "config.h"
+#include "dabOutput/edi/Config.h"
+#include "AFPacket.h"
+#include "PFT.h"
+#include "Interleaver.h"
+#include <vector>
+#include <unordered_map>
+#include <stdexcept>
+#include <cstdint>
+#include "dabOutput/dabOutput.h"
+
+namespace edi {
+
+/** Configuration for EDI output */
+
+class Sender {
+    public:
+        Sender(const configuration_t& conf);
+
+        void write(const TagPacket& tagpacket);
+
+    private:
+        configuration_t m_conf;
+        std::ofstream edi_debug_file;
+
+        // The TagPacket will then be placed into an AFPacket
+        edi::AFPacketiser edi_afPacketiser;
+
+        // The AF Packet will be protected with reed-solomon and split in fragments
+        edi::PFT edi_pft;
+
+        // To mitigate for burst packet loss, PFT fragments can be sent out-of-order
+        edi::Interleaver edi_interleaver;
+
+        std::unordered_map<udp_destination_t*, std::shared_ptr<UdpSocket>> udp_sockets;
+};
+
+}
+