Add work in progress odr-zmq2edi

8 files changed, 660 insertions, 6 deletions

diff --git a/src/DabMultiplexer.cpp b/src/DabMultiplexer.cpp
index c109919..8265b34 100644
--- a/src/DabMultiplexer.cpp
+++ b/src/DabMultiplexer.cpp
@@ -417,7 +417,7 @@ void DabMultiplexer::mux_frame(std::vector<std::shared_ptr<DabOutput> >& outputs
     //****** FCT ******//
     // Incremente for each frame, overflows at 249
     fc->FCT = currentFrame % 250;
-    edi_tagDETI.dflc = currentFrame % 5000;
+    edi_tagDETI.dlfc = currentFrame % 5000;
 
     //****** FICF ******//
     // Fast Information Channel Flag, 1 bit, =1 if FIC present
@@ -669,6 +669,10 @@ void DabMultiplexer::mux_frame(std::vector<std::shared_ptr<DabOutput> >& outputs
                 shared_ptr<OutputMetadata> md_edi_time =
                     make_shared<OutputMetadataEDITime>(edi_tagDETI.seconds);
                 output->setMetadata(md_edi_time);
+
+                shared_ptr<OutputMetadata> md_dlfc =
+                    make_shared<OutputMetadataDLFC>(currentFrame % 5000);
+                output->setMetadata(md_dlfc);
             }
         }
     }
@@ -708,6 +712,18 @@ void DabMultiplexer::mux_frame(std::vector<std::shared_ptr<DabOutput> >& outputs
 
     int frame_size = (FLtmp + 1 + 1 + 1 + 1) * 4;
 
+    for (auto output : outputs) {
+        auto out_zmq = std::dynamic_pointer_cast<DabOutputZMQ>(output);
+        if (out_zmq) {
+            // The separator allows the receiver to associate the right
+            // metadata with the right ETI frame, since the output gathers
+            // four ETI frames into one message
+            shared_ptr<OutputMetadata> md_sep =
+                make_shared<OutputMetadataSeparation>();
+            out_zmq->setMetadata(md_sep);
+        }
+    }
+
     // Give the data to the outputs
     for (auto output : outputs) {
         if (output->Write(etiFrame, frame_size) == -1) {
diff --git a/src/Makefile.am b/src/Makefile.am
index 557621c..c3c764f 100644
--- a/src/Makefile.am
+++ b/src/Makefile.am
@@ -26,7 +26,7 @@ else
 GITVERSION_FLAGS =
 endif
 
-bin_PROGRAMS=odr-dabmux zmqinput-keygen
+bin_PROGRAMS=odr-dabmux zmqinput-keygen odr-zmq2edi
 
 if HAVE_OUTPUT_RAW_TEST
 bin_PROGRAMS+=odr-zmq2farsync
@@ -124,5 +124,28 @@ odr_zmq2farsync_LDADD    = $(ZMQ_LIBS)
 odr_zmq2farsync_CFLAGS   = -Wall $(GITVERSION_FLAGS) -I$(FARSYNC_DIR)
 odr_zmq2farsync_CXXFLAGS = -Wall -std=c++11 $(GITVERSION_FLAGS) -I$(FARSYNC_DIR)
 
+odr_zmq2edi_SOURCES  = zmq2edi/zmq2edi.cpp \
+					   dabOutput/dabOutput.h \
+					   dabOutput/metadata.h dabOutput/metadata.cpp \
+					   dabOutput/edi/AFPacket.cpp dabOutput/edi/AFPacket.h \
+					   dabOutput/edi/TagItems.cpp dabOutput/edi/TagItems.h \
+					   dabOutput/edi/TagPacket.cpp dabOutput/edi/TagPacket.h \
+					   dabOutput/edi/PFT.cpp dabOutput/edi/PFT.h \
+					   dabOutput/edi/Interleaver.cpp dabOutput/edi/Interleaver.h \
+					   InetAddress.h InetAddress.cpp \
+					   UdpSocket.h UdpSocket.cpp \
+					   ReedSolomon.h ReedSolomon.cpp \
+					   fec/char.h fec/rs-common.h \
+					   fec/decode_rs_char.c fec/decode_rs.h \
+					   fec/encode_rs_char.c fec/encode_rs.h \
+					   fec/fec.h \
+					   fec/init_rs_char.c fec/init_rs.h \
+					   Log.h Log.cpp \
+					   crc.h crc.c \
+					   zmq.hpp
+odr_zmq2edi_LDADD    = $(ZMQ_LIBS)
+odr_zmq2edi_CFLAGS   = -Wall $(GITVERSION_FLAGS)
+odr_zmq2edi_CXXFLAGS = -Wall -std=c++11 $(GITVERSION_FLAGS)
+
 EXTRA_DIST = fec/README.md \
 			 fec/LICENSE
diff --git a/src/dabOutput/edi/TagItems.cpp b/src/dabOutput/edi/TagItems.cpp
index 75a4ecf..ed8517d 100644
--- a/src/dabOutput/edi/TagItems.cpp
+++ b/src/dabOutput/edi/TagItems.cpp
@@ -72,8 +72,8 @@ std::vector<uint8_t> TagDETI::Assemble()
     packet.push_back(0);
     packet.push_back(0);
 
-    uint8_t fct  = dflc % 250;
-    uint8_t fcth = dflc / 250;
+    uint8_t fct  = dlfc % 250;
+    uint8_t fcth = dlfc / 250;
 
 
     uint16_t detiHeader = fct | (fcth << 8) | (rfudf << 13) | (ficf << 14) | (atstf << 15);
@@ -122,7 +122,7 @@ std::vector<uint8_t> TagDETI::Assemble()
     packet[6] = (taglength >> 8) & 0xFF;
     packet[7] = taglength & 0xFF;
 
-    dflc = (dflc+1) % 5000;
+    dlfc = (dlfc+1) % 5000;
 
     /*
     std::cerr << "TagItem deti, packet.size " << packet.size() << std::endl;
diff --git a/src/dabOutput/edi/TagItems.h b/src/dabOutput/edi/TagItems.h
index 856d25b..19284d2 100644
--- a/src/dabOutput/edi/TagItems.h
+++ b/src/dabOutput/edi/TagItems.h
@@ -64,7 +64,7 @@ class TagDETI : public TagItem
         uint8_t rfa = 0;
         uint8_t rfu = 0; // MNSC is valid
         uint16_t mnsc = 0;
-        uint16_t dflc = 0; // modulo 5000 frame counter
+        uint16_t dlfc = 0; // modulo 5000 frame counter
 
         // ATST (optional)
         bool atstf = false; // presence of atst data
diff --git a/src/dabOutput/metadata.cpp b/src/dabOutput/metadata.cpp
index 6dfd65c..cb355cc 100644
--- a/src/dabOutput/metadata.cpp
+++ b/src/dabOutput/metadata.cpp
@@ -59,6 +59,16 @@ size_t write_meta(output_metadata_id_e md, uint8_t *buf, T value)
     return 3 + len_value;
 }
 
+size_t OutputMetadataSeparation::write(uint8_t *buf)
+{
+    buf[0] = static_cast<uint8_t>(getId());
+
+    // Length 0
+    buf[1] = 0;
+    buf[2] = 0;
+    return 3;
+}
+
 size_t OutputMetadataUTCO::write(uint8_t *buf)
 {
     return write_meta(getId(), buf, utco);
@@ -68,3 +78,9 @@ size_t OutputMetadataEDITime::write(uint8_t *buf)
 {
     return write_meta(getId(), buf, seconds);
 }
+
+size_t OutputMetadataDLFC::write(uint8_t *buf)
+{
+    return write_meta(getId(), buf, dlfc);
+}
+
diff --git a/src/dabOutput/metadata.h b/src/dabOutput/metadata.h
index e250938..ed16de5 100644
--- a/src/dabOutput/metadata.h
+++ b/src/dabOutput/metadata.h
@@ -39,12 +39,18 @@
  */
 
 enum class output_metadata_id_e {
+    // Contains no value, can be used to group fields
+    separation_marker = 0,
+
     // TAI-UTC offset, value is int16_t.
     utc_offset = 1,
 
     /* EDI Time is the number of SI seconds since 2000-01-01 T 00:00:00 UTC.
      * value is an uint32_t */
     edi_time = 2,
+
+    /* The DLFC field from the EDI TAG deti. value is uint16_t */
+    dlfc = 3,
 };
 
 struct OutputMetadata {
@@ -58,6 +64,14 @@ struct OutputMetadata {
     virtual size_t write(uint8_t *buf) = 0;
 };
 
+struct OutputMetadataSeparation : public OutputMetadata {
+    explicit OutputMetadataSeparation() {}
+    output_metadata_id_e getId(void) const { return output_metadata_id_e::separation_marker; }
+    virtual size_t getLength(void) const { return 3; }
+    virtual size_t write(uint8_t *buf);
+};
+
+
 struct OutputMetadataUTCO : public OutputMetadata {
     explicit OutputMetadataUTCO(int16_t utco) : utco(utco) {}
     output_metadata_id_e getId(void) const { return output_metadata_id_e::utc_offset; }
@@ -77,3 +91,13 @@ struct OutputMetadataEDITime : public OutputMetadata {
 
 };
 
+struct OutputMetadataDLFC : public OutputMetadata {
+    explicit OutputMetadataDLFC(uint16_t dlfc) : dlfc(dlfc) {}
+    output_metadata_id_e getId(void) const { return output_metadata_id_e::dlfc; }
+    virtual size_t getLength(void) const { return 5; }
+    virtual size_t write(uint8_t *buf);
+
+    uint16_t dlfc;
+
+};
+
diff --git a/src/zmq2edi/README.md b/src/zmq2edi/README.md
new file mode 100644
index 0000000..e509479
--- /dev/null
+++ b/src/zmq2edi/README.md
@@ -0,0 +1,8 @@
+Convert an ZeroMQ stream to EDI
+===============================
+
+This *zmq2edi* tool can receive a ZMQ ETI stream from
+ODR-DabMux and generate and EDI stream.
+
+Quite useful if your modulator wants EDI input, and your network is not good
+enough making you want to use something based on TCP.
diff --git a/src/zmq2edi/zmq2edi.cpp b/src/zmq2edi/zmq2edi.cpp
new file mode 100644
index 0000000..7a2eed1
--- /dev/null
+++ b/src/zmq2edi/zmq2edi.cpp
@@ -0,0 +1,567 @@
+/*
+   Copyright (C) 2002, 2003, 2004, 2005, 2006, 2007, 2008, 2009, 2010,
+   2011, 2012 Her Majesty the Queen in Right of Canada (Communications
+   Research Center Canada)
+
+   Copyright (C) 2017
+   Matthias P. Braendli, matthias.braendli@mpb.li
+
+    http://www.opendigitalradio.org
+   */
+/*
+   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 "dabOutput/dabOutput.h"
+#include "dabOutput/edi/TagItems.h"
+#include "dabOutput/edi/TagPacket.h"
+#include "dabOutput/edi/AFPacket.h"
+#include "dabOutput/edi/PFT.h"
+#include "dabOutput/edi/Interleaver.h"
+#include "Log.h"
+#include "zmq.hpp"
+#include "math.h"
+#include <getopt.h>
+#include <string.h>
+#include <iostream>
+#include <iterator>
+#include <vector>
+
+static edi_configuration_t edi_conf;
+
+static std::ofstream edi_debug_file;
+
+// The TagPacket will then be placed into an AFPacket
+static edi::AFPacketiser edi_afPacketiser;
+
+// The AF Packet will be protected with reed-solomon and split in fragments
+static edi::PFT edi_pft;
+
+// To mitigate for burst packet loss, PFT fragments can be sent out-of-order
+static edi::Interleaver edi_interleaver;
+
+// This metadata gets transmitted in the zmq stream
+struct metadata_t {
+    uint32_t edi_time;
+    int16_t utc_offset;
+    uint16_t dlfc;
+};
+
+void usage(void)
+{
+    using namespace std;
+
+    cerr << "Usage:" << endl;
+    cerr << "odr-zmq2edi [options] <source>" << endl << endl;
+
+    cerr << "Where the options are:" << endl;
+    cerr << " <source> is a ZMQ URL that points to a ODR-DabMux ZMQ output." << endl;
+    cerr << " -d <destination ip> sets the destination ip." << endl;
+    cerr << " -p <destination port> sets the destination port." << endl;
+    cerr << " -s <source port> sets the source port." << endl;
+    cerr << " -S <source ip> select the source IP in case we want to use multicast." << endl;
+    cerr << " -t <ttl> set the packet's TTL." << endl;
+    cerr << " -P Disable PFT and send AFPackets." << endl;
+    cerr << " -f <fec> sets the FEC." << endl;
+    cerr << " -i <interleave> enables the interleaved with this latency." << endl;
+    cerr << " -D dumps the EDI to edi.debug file." << endl;
+    cerr << " -v Enables verbose mode." << endl;
+    cerr << " -a <tagpacket alignement> sets the alignment of the TAG Packet (default 8)." << endl;
+}
+
+static void edi_setup(void) {
+    if (edi_conf.verbose) {
+        etiLog.log(info, "Setup EDI");
+    }
+
+    if (edi_conf.dump) {
+        edi_debug_file.open("./edi.debug");
+    }
+
+    if (edi_conf.enabled()) {
+        for (auto& edi_destination : edi_conf.destinations) {
+            auto edi_output = std::make_shared<UdpSocket>(edi_destination.source_port);
+
+            if (not edi_destination.source_addr.empty()) {
+                int err = edi_output->setMulticastSource(edi_destination.source_addr.c_str());
+                if (err) {
+                    throw std::runtime_error("EDI socket set source failed!");
+                }
+                err = edi_output->setMulticastTTL(edi_destination.ttl);
+                if (err) {
+                    throw std::runtime_error("EDI socket set TTL failed!");
+                }
+            }
+
+            edi_destination.socket = edi_output;
+        }
+    }
+
+    if (edi_conf.verbose) {
+        etiLog.log(info, "EDI set up");
+    }
+
+    // The AF Packet will be protected with reed-solomon and split in fragments
+    edi::PFT pft(edi_conf);
+    edi_pft = pft;
+
+    if (edi_conf.interleaver_enabled()) {
+        edi_interleaver.SetLatency(edi_conf.latency_frames);
+    }
+}
+
+static void print_edi_conf(void)
+{
+    if (edi_conf.enabled()) {
+        etiLog.level(info) << "EDI";
+        etiLog.level(info) << " verbose     " << edi_conf.verbose;
+        for (auto& edi_dest : edi_conf.destinations) {
+            etiLog.level(info) << " to " << edi_dest.dest_addr << ":" << edi_conf.dest_port;
+            if (not edi_dest.source_addr.empty()) {
+                etiLog.level(info) << "  source      " << edi_dest.source_addr;
+                etiLog.level(info) << "  ttl         " << edi_dest.ttl;
+            }
+            etiLog.level(info) << "  source port " << edi_dest.source_port;
+        }
+        if (edi_conf.interleaver_enabled()) {
+            etiLog.level(info) << " interleave     " << edi_conf.latency_frames * 24 << " ms";
+        }
+    }
+    else {
+        etiLog.level(info) << "EDI disabled";
+    }
+}
+
+static metadata_t get_md_one_frame(uint8_t *buf, size_t size, size_t *consumed_bytes)
+{
+    size_t remaining = size;
+    if (remaining < 3) {
+        etiLog.level(warn) << "Insufficient data to parse metadata";
+        throw std::runtime_error("Insufficient data");
+    }
+
+    metadata_t md;
+    bool utc_offset_received = false;
+    bool edi_time_received = false;
+    bool dlfc_received = false;
+
+    while (remaining) {
+        uint8_t id = buf[0];
+        uint16_t len = (((uint16_t)buf[0]) << 8) + buf[1];
+
+        if (id == static_cast<uint8_t>(output_metadata_id_e::separation_marker)) {
+            if (len != 0) {
+                etiLog.level(warn) << "Invalid length for metadata: separation_marker";
+            }
+
+            if (not utc_offset_received or not edi_time_received or not dlfc_received) {
+                throw std::runtime_error("Incomplete metadata received");
+            }
+
+            remaining -= 3;
+            *consumed_bytes = size - remaining;
+            return md;
+        }
+        else if (id == static_cast<uint8_t>(output_metadata_id_e::utc_offset)) {
+            if (len != 2) {
+                etiLog.level(warn) << "Invalid length for metadata: utc_offset";
+            }
+            if (remaining < 2) {
+                throw std::runtime_error("Insufficient data for utc_offset");
+            }
+            uint16_t utco;
+            std::memcpy(&utco, buf + 2, sizeof(utco));
+            md.utc_offset = ntohl(utco);
+            utc_offset_received = true;
+            remaining -= 5;
+            buf += 5;
+        }
+        else if (id == static_cast<uint8_t>(output_metadata_id_e::edi_time)) {
+            if (len != 4) {
+                etiLog.level(warn) << "Invalid length for metadata: edi_time";
+            }
+            if (remaining < 4) {
+                throw std::runtime_error("Insufficient data for edi_time");
+            }
+            uint32_t edi_time;
+            std::memcpy(&edi_time, buf + 2, sizeof(edi_time));
+            md.edi_time = ntohl(edi_time);
+            edi_time_received = true;
+            remaining -= 7;
+            buf += 7;
+        }
+        else if (id == static_cast<uint8_t>(output_metadata_id_e::dlfc)) {
+            if (len != 3) {
+                etiLog.level(warn) << "Invalid length for metadata: dlfc";
+            }
+            if (remaining < 2) {
+                throw std::runtime_error("Insufficient data for dlfc");
+            }
+            uint16_t dlfc;
+            std::memcpy(&dlfc, buf + 2, sizeof(dlfc));
+            md.dlfc = ntohl(dlfc);
+            dlfc_received = true;
+            remaining -= 5;
+            buf += 5;
+        }
+    }
+
+    throw std::runtime_error("Insufficient data");
+}
+
+static void send_eti_frame(uint8_t* p, metadata_t metadata)
+{
+    edi::TagDETI edi_tagDETI;
+    edi::TagStarPTR edi_tagStarPtr;
+    std::map<int, edi::TagESTn> edi_subchannelToTag;
+    // The above Tag Items will be assembled into a TAG Packet
+    edi::TagPacket edi_tagpacket(edi_conf.tagpacket_alignment);
+
+    // SYNC
+    edi_tagDETI.stat = p[0];
+
+    // LIDATA FCT
+    //const int fct = p[4];
+    edi_tagDETI.dlfc = metadata.dlfc;
+
+    bool ficf = (p[5] & 0x80) >> 7;
+
+    const int nst = p[5] & 0x7F;
+
+    edi_tagDETI.fp = (p[6] & 0xE0) >> 5;
+    const int mid = (p[6] & 0x18) >> 3;
+    edi_tagDETI.mid = mid;
+    //const int fl = (p[6] & 0x07) * 256 + p[7];
+
+    int ficl = 0;
+    if (ficf == 0) {
+        etiLog.level(warn) << "Not FIC in data stream!";
+        return;
+    }
+    else if (mid == 3) {
+        ficl = 32;
+    }
+    else {
+        ficl = 24;
+    }
+
+    std::vector<uint32_t> sad(nst);
+    std::vector<uint32_t> stl(nst);
+    // Loop over STC subchannels:
+    for (int i=0; i < nst; i++) {
+        uint32_t scid = (p[8 + 4*i] & 0xFC) >> 2;
+        sad[i] = (p[8+4*i] & 0x03) * 256 + p[9+4*i];
+        uint32_t tpl = (p[10+4*i] & 0xFC) >> 2;
+        stl[i] = (p[10+4*i] & 0x03) * 256 + \
+                 p[11+4*i];
+
+        edi::TagESTn tag_ESTn;
+        tag_ESTn.id = i;
+        tag_ESTn.scid = scid;
+        tag_ESTn.sad = sad[i];
+        tag_ESTn.tpl = tpl;
+        tag_ESTn.rfa = 0; // two bits
+        tag_ESTn.mst_length = stl[i];
+        tag_ESTn.mst_data = nullptr;
+
+        edi_subchannelToTag[i] = tag_ESTn;
+    }
+
+    const uint16_t mnsc = p[8 + 4*nst] * 256 + \
+                          p[8 + 4*nst + 1];
+    edi_tagDETI.mnsc = mnsc;
+
+    /*const uint16_t crc1 = p[8 + 4*nst + 2]*256 + \
+                          p[8 + 4*nst + 3]; */
+
+    edi_tagDETI.fic_data = p + 12 + 4*nst;
+    edi_tagDETI.fic_length = ficl * 4;
+
+    // loop over MSC subchannels
+    int offset = 0;
+    for (int i=0; i < nst; i++) {
+        edi::TagESTn& tag = edi_subchannelToTag[i];
+        tag.mst_data = (p + 12 + 4*nst + ficf*ficl*4 + offset);
+
+        offset += stl[i] * 8;
+    }
+
+    /*
+    const uint16_t crc2 = p[12 + 4*nst + ficf*ficl*4 + offset] * 256 + \
+                          p[12 + 4*nst + ficf*ficl*4 + offset + 1]; */
+
+    // TIST
+    const size_t tist_ix = 12 + 4*nst + ficf*ficl*4 + offset + 4;
+    uint32_t tist = (uint32_t)(p[tist_ix]) << 24 |
+                    (uint32_t)(p[tist_ix+1]) << 16 |
+                    (uint32_t)(p[tist_ix+2]) << 8 |
+                    (uint32_t)(p[tist_ix+3]);
+
+    edi_tagDETI.tsta = tist;
+    edi_tagDETI.atstf = 1;
+    edi_tagDETI.utco = metadata.utc_offset;
+    edi_tagDETI.seconds = metadata.edi_time;
+
+    if (edi_conf.enabled()) {
+        // put tags *ptr, DETI and all subchannels into one TagPacket
+        edi_tagpacket.tag_items.push_back(&edi_tagStarPtr);
+        edi_tagpacket.tag_items.push_back(&edi_tagDETI);
+
+        for (auto& tag : edi_subchannelToTag) {
+            edi_tagpacket.tag_items.push_back(&tag.second);
+        }
+
+        // Assemble into one AF Packet
+        edi::AFPacket edi_afpacket = edi_afPacketiser.Assemble(edi_tagpacket);
+
+        if (edi_conf.enable_pft) {
+            // Apply PFT layer to AF Packet (Reed Solomon FEC and Fragmentation)
+            std::vector<edi::PFTFragment> edi_fragments = edi_pft.Assemble(edi_afpacket);
+
+            if (edi_conf.verbose) {
+                fprintf(stderr, "EDI number of PFT fragment before interleaver %zu\n",
+                        edi_fragments.size());
+            }
+
+            if (edi_conf.interleaver_enabled()) {
+                edi_fragments = edi_interleaver.Interleave(edi_fragments);
+            }
+
+            // Send over ethernet
+            for (const auto& edi_frag : edi_fragments) {
+                for (auto& dest : edi_conf.destinations) {
+                    InetAddress addr;
+                    addr.setAddress(dest.dest_addr.c_str());
+                    addr.setPort(edi_conf.dest_port);
+
+                    dest.socket->send(edi_frag, addr);
+                }
+
+                if (edi_conf.dump) {
+                    std::ostream_iterator<uint8_t> debug_iterator(edi_debug_file);
+                    std::copy(edi_frag.begin(), edi_frag.end(), debug_iterator);
+                }
+            }
+
+            if (edi_conf.verbose) {
+                fprintf(stderr, "EDI number of PFT fragments %zu\n",
+                        edi_fragments.size());
+            }
+        }
+        else {
+            // Send over ethernet
+            for (auto& dest : edi_conf.destinations) {
+                InetAddress addr;
+                addr.setAddress(dest.dest_addr.c_str());
+                addr.setPort(edi_conf.dest_port);
+
+                dest.socket->send(edi_afpacket, addr);
+            }
+
+            if (edi_conf.dump) {
+                std::ostream_iterator<uint8_t> debug_iterator(edi_debug_file);
+                std::copy(edi_afpacket.begin(), edi_afpacket.end(), debug_iterator);
+            }
+        }
+    }
+}
+
+int start(int argc, char **argv)
+{
+    edi_destination_t edi_destination;
+
+    edi_conf.enable_pft = true;
+
+    if (argc == 0) {
+        usage();
+        return 1;
+    }
+
+    char ch = 0;
+    while (ch != -1) {
+        ch = getopt(argc, argv, "d:p:s:S:t:Pf:i:Dva:");
+        switch (ch) {
+            case -1:
+                break;
+            case 'd':
+                edi_destination.dest_addr = optarg;
+                break;
+            case 'p':
+                edi_conf.dest_port = std::stoi(optarg);
+                break;
+            case 's':
+                edi_destination.source_port = std::stoi(optarg);
+                break;
+            case 'S':
+                edi_destination.source_addr = optarg;
+                break;
+            case 't':
+                edi_destination.ttl = std::stoi(optarg);
+                break;
+            case 'P':
+                edi_conf.enable_pft = false;
+                break;
+            case 'f':
+                edi_conf.fec = std::stoi(optarg);
+                break;
+            case 'i':
+                {
+                double interleave_ms = std::stod(optarg);
+                if (interleave_ms != 0.0) {
+                    if (interleave_ms < 0) {
+                        throw std::runtime_error("EDI output: negative interleave value is invalid.");
+                    }
+
+                    auto latency_rounded = lround(interleave_ms / 24.0);
+                    if (latency_rounded * 24 > 30000) {
+                        throw std::runtime_error("EDI output: interleaving set for more than 30 seconds!");
+                    }
+
+                    edi_conf.latency_frames = latency_rounded;
+                }
+                }
+                break;
+            case 'D':
+                edi_conf.dump = true;
+                break;
+            case 'v':
+                edi_conf.verbose = true;
+                break;
+            case 'a':
+                edi_conf.tagpacket_alignment = std::stoi(optarg);
+                break;
+            case 'h':
+                usage();
+                return 1;
+            default:
+                etiLog.log(error, "Option '%c' not understood", ch);
+                usage();
+                return 1;
+        }
+    }
+
+    if (optind >= argc) {
+        etiLog.level(error) << "source option is missing";
+        return 1;
+    }
+
+    print_edi_conf();
+    edi_setup();
+
+    const char* source_url = argv[optind];
+
+    etiLog.level(info) << "Opening ZMQ input: " << source_url;
+    zmq::context_t zmq_ctx(1);
+    zmq::socket_t zmq_sock(zmq_ctx, ZMQ_SUB);
+
+    zmq_sock.connect(source_url);
+    zmq_sock.setsockopt(ZMQ_SUBSCRIBE, NULL, 0); // subscribe to all messages
+
+    etiLog.level(info) << "Entering main loop";
+    size_t frame_count = 0;
+    size_t loop_counter = 0;
+    size_t error_count = 0;
+    while (error_count < 10)
+    {
+        zmq::message_t incoming;
+        zmq_sock.recv(&incoming);
+
+        zmq_dab_message_t* dab_msg = (zmq_dab_message_t*)incoming.data();
+
+        if (dab_msg->version != 1) {
+            etiLog.level(error) << "ZeroMQ wrong packet version " << dab_msg->version;
+            error_count++;
+        }
+
+        int offset = sizeof(dab_msg->version) +
+            NUM_FRAMES_PER_ZMQ_MESSAGE * sizeof(*dab_msg->buflen);
+
+        std::list<std::pair<std::vector<uint8_t>, metadata_t> > all_frames;
+
+        for (int i = 0; i < NUM_FRAMES_PER_ZMQ_MESSAGE; i++) {
+            if (dab_msg->buflen[i] <= 0 ||
+                dab_msg->buflen[i] > 6144)
+            {
+                etiLog.level(error) << "ZeroMQ buffer " << i << " has invalid length " <<
+                    dab_msg->buflen[i];
+                error_count++;
+            }
+            else {
+                std::vector<uint8_t> buf(6144, 0x55);
+
+                const int framesize = dab_msg->buflen[i];
+
+                memcpy(&buf.front(),
+                        ((uint8_t*)incoming.data()) + offset,
+                        framesize);
+
+                all_frames.emplace_back(
+                        std::piecewise_construct,
+                        std::make_tuple(std::move(buf)),
+                        std::make_tuple());
+
+                offset += framesize;
+            }
+        }
+
+        for (auto &f : all_frames) {
+            size_t consumed_bytes = 0;
+
+            std::cerr << "MD: " << incoming.size() - offset << std::endl;
+
+            f.second = get_md_one_frame(
+                    static_cast<uint8_t*>(incoming.data()) + offset,
+                    incoming.size() - offset,
+                    &consumed_bytes);
+
+            offset += consumed_bytes;
+        }
+
+        for (auto &f : all_frames) {
+            send_eti_frame(f.first.data(), f.second);
+            frame_count++;
+        }
+
+        loop_counter++;
+        if (loop_counter > 250) {
+            etiLog.level(info) << "Transmitted " << frame_count << " ETI frames";
+            loop_counter = 0;
+        }
+    }
+
+    return error_count > 0 ? 2 : 0;
+}
+
+int main(int argc, char **argv)
+{
+    etiLog.level(info) << "ZMQ2EDI converter from " <<
+        PACKAGE_NAME << " " <<
+#if defined(GITVERSION)
+        GITVERSION <<
+#else
+        PACKAGE_VERSION <<
+#endif
+        " starting up";
+
+    try {
+        return start(argc, argv);
+    }
+    catch (std::runtime_error &e) {
+        etiLog.level(error) << "Error: " << e.what();
+    }
+
+    return 1;
+}