Replace EDI-to-ETI converter with a dedicated EDI source

16 files changed, 577 insertions, 190 deletions

diff --git a/src/Buffer.cpp b/src/Buffer.cpp
index fa7f52f..8631c42 100644
--- a/src/Buffer.cpp
+++ b/src/Buffer.cpp
@@ -2,6 +2,11 @@
    Copyright (C) 2011
    Her Majesty the Queen in Right of Canada (Communications Research
    Center Canada)
+
+   Copyright (C) 2016
+   Matthias P. Braendli, matthias.braendli@mpb.li
+
+    http://opendigitalradio.org
  */
 /*
    This file is part of ODR-DabMod.
@@ -44,6 +49,16 @@ Buffer::Buffer(size_t len, const void *data)
     setData(data, len);
 }
 
+Buffer::Buffer(const std::vector<uint8_t> &vec)
+{
+    PDEBUG("Buffer::Buffer(vector [%zu])\n", vec.size());
+
+    this->len = 0;
+    this->size = 0;
+    this->data = NULL;
+    setData(vec.data(), vec.size());
+}
+
 
 Buffer::~Buffer()
 {
@@ -58,6 +73,11 @@ Buffer &Buffer::operator=(const Buffer &copy)
     return *this;
 }
 
+Buffer &Buffer::operator=(const std::vector<uint8_t> &copy)
+{
+    setData(copy.data(), copy.size());
+    return *this;
+}
 
 Buffer &Buffer::operator+=(const Buffer &copy)
 {
diff --git a/src/Buffer.h b/src/Buffer.h
index c4c73ce..8c5c768 100644
--- a/src/Buffer.h
+++ b/src/Buffer.h
@@ -2,6 +2,11 @@
    Copyright (C) 2011
    Her Majesty the Queen in Right of Canada (Communications Research
    Center Canada)
+
+   Copyright (C) 2016
+   Matthias P. Braendli, matthias.braendli@mpb.li
+
+    http://opendigitalradio.org
  */
 /*
    This file is part of ODR-DabMod.
@@ -27,6 +32,7 @@
 #endif
 
 #include <unistd.h>
+#include <vector>
 #include <memory>
 
 /* Buffer is a container for a byte array, that is memcpy'ed
@@ -50,7 +56,8 @@ class Buffer {
     public:
         using sptr = std::shared_ptr<Buffer>;
 
-        Buffer(const Buffer& copy);
+        Buffer(const Buffer& copy) = default;
+        Buffer(const std::vector<uint8_t> &vec);
         Buffer(size_t len = 0, const void *data = NULL);
         ~Buffer();
 
@@ -62,6 +69,7 @@ class Buffer {
          */
         void setData(const void *data, size_t len);
         Buffer &operator=(const Buffer &copy);
+        Buffer &operator=(const std::vector<uint8_t> &copy);
 
         /* Concatenate the current data with the new data given.
          * Reallocates memory if needed.
diff --git a/src/DabMod.cpp b/src/DabMod.cpp
index 2dea12e..e48e748 100644
--- a/src/DabMod.cpp
+++ b/src/DabMod.cpp
@@ -187,7 +187,6 @@ int launch_modulator(int argc, char* argv[])
     auto inputZeroMQReader = make_shared<InputZeroMQReader>();
 #endif
 
-    auto inputEdiReader = make_shared<InputEdiReader>();
     auto inputTcpReader = make_shared<InputTcpReader>();
 
     struct sigaction sa;
@@ -691,6 +690,12 @@ int launch_modulator(int argc, char* argv[])
         fprintf(stderr, "%zu Hz\n", outputRate);
     }
 
+
+    EtiReader etiReader(tist_offset_s, tist_delay_stages);
+    EdiReader ediReader;
+    EdiDecoder::ETIDecoder ediInput(ediReader);
+    EdiUdpInput ediUdpInput(ediInput);
+
     if (inputTransport == "file") {
         // Opening ETI input file
         if (inputFileReader.Open(inputName, loop) == -1) {
@@ -717,8 +722,7 @@ int launch_modulator(int argc, char* argv[])
         m.inputReader = inputTcpReader.get();
     }
     else if (inputTransport == "edi") {
-        inputEdiReader->Open(inputName);
-        m.inputReader = inputEdiReader.get();
+        ediUdpInput.Open(inputName);
     }
     else
     {
@@ -772,94 +776,139 @@ int launch_modulator(int argc, char* argv[])
     }
     set_thread_name("modulator");
 
-    while (run_again) {
-        Flowgraph flowgraph;
+    if (ediUdpInput.isEnabled()) {
+        while (run_again) {
+            Flowgraph flowgraph;
 
-        m.flowgraph = &flowgraph;
-        m.data.setLength(6144);
+            etiLog.level(debug) << "Build mod";
+            auto modulator = make_shared<DabModulator>(
+                    ediReader, tiiConfig, outputRate, clockRate,
+                    dabMode, gainMode, digitalgain, normalise,
+                    filterTapsFilename);
 
-        auto input = make_shared<InputMemory>(&m.data);
-        auto modulator = make_shared<DabModulator>(
-                tist_offset_s, tist_delay_stages,
-                tiiConfig, outputRate, clockRate, dabMode, gainMode,
-                digitalgain, normalise, filterTapsFilename);
+            etiLog.level(debug) << "Connect";
+            if (format_converter) {
+                flowgraph.connect(modulator, format_converter);
+                flowgraph.connect(format_converter, output);
+            }
+            else {
+                flowgraph.connect(modulator, output);
+            }
 
-        flowgraph.connect(input, modulator);
-        if (format_converter) {
-            flowgraph.connect(modulator, format_converter);
-            flowgraph.connect(format_converter, output);
-        }
-        else {
-            flowgraph.connect(modulator, output);
-        }
+            etiLog.level(debug) << "SetETISource";
 
 #if defined(HAVE_OUTPUT_UHD)
-        if (useUHDOutput) {
-            ((OutputUHD*)output.get())->setETIReader(modulator->getEtiReader());
+            if (useUHDOutput) {
+                ((OutputUHD*)output.get())->setETISource(modulator->getEtiSource());
+            }
+#endif
+
+            etiLog.level(debug) << "Loop";
+            size_t framecount = 0;
+            while (true) {
+                while (not ediReader.isFrameReady()) {
+                    ediUdpInput.rxPacket();
+                }
+                etiLog.level(debug) << "Frame Ready";
+                framecount++;
+                flowgraph.run();
+                etiLog.level(debug) << "now clear";
+                ediReader.clearFrame();
+
+                /* Check every once in a while if the remote control
+                 * is still working */
+                if ((framecount % 250) == 0) {
+                    rcs.check_faults();
+                }
+            }
+
         }
+    }
+    else {
+        while (run_again) {
+            Flowgraph flowgraph;
+
+            m.flowgraph = &flowgraph;
+            m.data.setLength(6144);
+
+            auto input = make_shared<InputMemory>(&m.data);
+            auto modulator = make_shared<DabModulator>(
+                    etiReader, tiiConfig, outputRate, clockRate,
+                    dabMode, gainMode, digitalgain, normalise,
+                    filterTapsFilename);
+
+            if (format_converter) {
+                flowgraph.connect(modulator, format_converter);
+                flowgraph.connect(format_converter, output);
+            }
+            else {
+                flowgraph.connect(modulator, output);
+            }
+
+#if defined(HAVE_OUTPUT_UHD)
+            if (useUHDOutput) {
+                ((OutputUHD*)output.get())->setETISource(modulator->getEtiSource());
+            }
 #endif
 
-        m.inputReader->PrintInfo();
-
-        run_modulator_state_t st = run_modulator(m);
-        etiLog.log(trace, "DABMOD,run_modulator() = %d", st);
-
-        switch (st) {
-            case run_modulator_state_t::failure:
-                etiLog.level(error) << "Modulator failure.";
-                run_again = false;
-                ret = 1;
-                break;
-            case run_modulator_state_t::again:
-                etiLog.level(warn) << "Restart modulator.";
-                run_again = false;
-                if (inputTransport == "file") {
-                    if (inputFileReader.Open(inputName, loop) == -1) {
-                        etiLog.level(error) << "Unable to open input file!";
-                        ret = 1;
+            m.inputReader->PrintInfo();
+
+            run_modulator_state_t st = run_modulator(m);
+            etiLog.log(trace, "DABMOD,run_modulator() = %d", st);
+
+            switch (st) {
+                case run_modulator_state_t::failure:
+                    etiLog.level(error) << "Modulator failure.";
+                    run_again = false;
+                    ret = 1;
+                    break;
+                case run_modulator_state_t::again:
+                    etiLog.level(warn) << "Restart modulator.";
+                    run_again = false;
+                    if (inputTransport == "file") {
+                        if (inputFileReader.Open(inputName, loop) == -1) {
+                            etiLog.level(error) << "Unable to open input file!";
+                            ret = 1;
+                        }
+                        else {
+                            run_again = true;
+                        }
                     }
-                    else {
+                    else if (inputTransport == "zeromq") {
+#if defined(HAVE_ZEROMQ)
                         run_again = true;
+                        // Create a new input reader
+                        inputZeroMQReader = make_shared<InputZeroMQReader>();
+                        inputZeroMQReader->Open(inputName, inputMaxFramesQueued);
+                        m.inputReader = inputZeroMQReader.get();
+#endif
                     }
-                }
-                else if (inputTransport == "zeromq") {
-#if defined(HAVE_ZEROMQ)
+                    else if (inputTransport == "tcp") {
+                        inputTcpReader = make_shared<InputTcpReader>();
+                        inputTcpReader->Open(inputName);
+                        m.inputReader = inputTcpReader.get();
+                    }
+                    break;
+                case run_modulator_state_t::reconfigure:
+                    etiLog.level(warn) << "Detected change in ensemble configuration.";
+                    /* We can keep the input in this care */
                     run_again = true;
-                    // Create a new input reader
-                    inputZeroMQReader = make_shared<InputZeroMQReader>();
-                    inputZeroMQReader->Open(inputName, inputMaxFramesQueued);
-                    m.inputReader = inputZeroMQReader.get();
-#endif
-                }
-                else if (inputTransport == "tcp") {
-                    inputTcpReader = make_shared<InputTcpReader>();
-                    inputTcpReader->Open(inputName);
-                    m.inputReader = inputTcpReader.get();
-                }
-                break;
-            case run_modulator_state_t::reconfigure:
-                etiLog.level(warn) << "Detected change in ensemble configuration.";
-                /* We can keep the input in this care */
-                run_again = true;
-                break;
-            case run_modulator_state_t::normal_end:
-            default:
-                etiLog.level(info) << "modulator stopped.";
-                ret = 0;
-                run_again = false;
-                break;
-        }
-
-        fprintf(stderr, "\n\n");
-        etiLog.level(info) << m.framecount << " DAB frames encoded";
-        etiLog.level(info) << ((float)m.framecount * 0.024f) << " seconds encoded";
-
-        m.data.setLength(0);
-    }
+                    break;
+                case run_modulator_state_t::normal_end:
+                default:
+                    etiLog.level(info) << "modulator stopped.";
+                    ret = 0;
+                    run_again = false;
+                    break;
+            }
+
+            fprintf(stderr, "\n\n");
+            etiLog.level(info) << m.framecount << " DAB frames encoded";
+            etiLog.level(info) << ((float)m.framecount * 0.024f) << " seconds encoded";
 
-    ////////////////////////////////////////////////////////////////////////
-    // Cleaning things
-    ////////////////////////////////////////////////////////////////////////
+            m.data.setLength(0);
+        }
+    }
 
     etiLog.level(info) << "Terminating";
     return ret;
@@ -867,6 +916,7 @@ int launch_modulator(int argc, char* argv[])
 
 run_modulator_state_t run_modulator(modulator_data& m)
 {
+
     auto ret = run_modulator_state_t::failure;
     try {
         while (running) {
@@ -887,9 +937,6 @@ run_modulator_state_t run_modulator(modulator_data& m)
                 PDEBUG("* Read frame %lu\n", m.framecount);
                 PDEBUG("*****************************************\n");
 
-                ////////////////////////////////////////////////////////////////
-                // Processing data
-                ////////////////////////////////////////////////////////////////
                 m.flowgraph->run();
 
                 /* Check every once in a while if the remote control
diff --git a/src/DabModulator.cpp b/src/DabModulator.cpp
index 6c28a2e..cbe8f07 100644
--- a/src/DabModulator.cpp
+++ b/src/DabModulator.cpp
@@ -3,7 +3,7 @@
    Her Majesty the Queen in Right of Canada (Communications Research
    Center Canada)
 
-   Copyright (C) 2015
+   Copyright (C) 2016
    Matthias P. Braendli, matthias.braendli@mpb.li
 
     http://opendigitalradio.org
@@ -55,21 +55,21 @@
 #include "Log.h"
 
 DabModulator::DabModulator(
-        double& tist_offset_s, unsigned tist_delay_stages,
+        EtiSource& etiSource,
         tii_config_t& tiiConfig,
         unsigned outputRate, unsigned clockRate,
         unsigned dabMode, GainMode gainMode,
         float& digGain, float normalise,
         const std::string& filterTapsFilename
         ) :
-    ModCodec(),
+    ModInput(),
     myOutputRate(outputRate),
     myClockRate(clockRate),
     myDabMode(dabMode),
     myGainMode(gainMode),
     myDigGain(digGain),
     myNormalise(normalise),
-    myEtiReader(tist_offset_s, tist_delay_stages),
+    myEtiSource(etiSource),
     myFlowgraph(NULL),
     myFilterTapsFilename(filterTapsFilename),
     myTiiConfig(tiiConfig)
@@ -134,16 +134,14 @@ void DabModulator::setMode(unsigned mode)
 }
 
 
-int DabModulator::process(Buffer* const dataIn, Buffer* dataOut)
+int DabModulator::process(Buffer* dataOut)
 {
     using namespace std;
 
-    PDEBUG("DabModulator::process(dataIn: %p, dataOut: %p)\n",
-            dataIn, dataOut);
+    PDEBUG("DabModulator::process(dataOut: %p)\n", dataOut);
 
-    myEtiReader.process(dataIn);
     if (myFlowgraph == NULL) {
-        unsigned mode = myEtiReader.getMode();
+        unsigned mode = myEtiSource.getMode();
         if (myDabMode != 0) {
             mode = myDabMode;
         } else if (mode == 0) {
@@ -156,10 +154,8 @@ int DabModulator::process(Buffer* const dataIn, Buffer* dataOut)
         // CIF data initialisation
         ////////////////////////////////////////////////////////////////
         auto cifPrbs = make_shared<PrbsGenerator>(864 * 8, 0x110);
-        auto cifMux = make_shared<FrameMultiplexer>(
-                myFicSizeOut + 864 * 8, myEtiReader.getSubchannels());
-
-        auto cifPart = make_shared<BlockPartitioner>(mode, myEtiReader.getFp());
+        auto cifMux = make_shared<FrameMultiplexer>(myEtiSource);
+        auto cifPart = make_shared<BlockPartitioner>(mode, myEtiSource.getFp());
 
         auto cifMap = make_shared<QpskSymbolMapper>(myNbCarriers);
         auto cifRef = make_shared<PhaseReference>(mode);
@@ -229,7 +225,7 @@ int DabModulator::process(Buffer* const dataIn, Buffer* dataOut)
         ////////////////////////////////////////////////////////////////
         // Processing FIC
         ////////////////////////////////////////////////////////////////
-        shared_ptr<FicSource> fic(myEtiReader.getFic());
+        shared_ptr<FicSource> fic(myEtiSource.getFic());
         ////////////////////////////////////////////////////////////////
         // Data initialisation
         ////////////////////////////////////////////////////////////////
@@ -269,7 +265,7 @@ int DabModulator::process(Buffer* const dataIn, Buffer* dataOut)
         ////////////////////////////////////////////////////////////////
         // Configuring subchannels
         ////////////////////////////////////////////////////////////////
-        for (const auto& subchannel : myEtiReader.getSubchannels()) {
+        for (const auto& subchannel : myEtiSource.getSubchannels()) {
 
             ////////////////////////////////////////////////////////////
             // Data initialisation
diff --git a/src/DabModulator.h b/src/DabModulator.h
index 77c0457..d768875 100644
--- a/src/DabModulator.h
+++ b/src/DabModulator.h
@@ -3,7 +3,7 @@
    Her Majesty the Queen in Right of Canada (Communications Research
    Center Canada)
 
-   Copyright (C) 2015
+   Copyright (C) 2016
    Matthias P. Braendli, matthias.braendli@mpb.li
 
     http://opendigitalradio.org
@@ -45,11 +45,11 @@
 #include "TII.h"
 
 
-class DabModulator : public ModCodec
+class DabModulator : public ModInput
 {
 public:
     DabModulator(
-            double& tist_offset_s, unsigned tist_delay_stages,
+            EtiSource& etiSource,
             tii_config_t& tiiConfig,
             unsigned outputRate, unsigned clockRate,
             unsigned dabMode, GainMode gainMode,
@@ -59,11 +59,11 @@ public:
     DabModulator& operator=(const DabModulator& other) = delete;
     virtual ~DabModulator();
 
-    int process(Buffer* const dataIn, Buffer* dataOut);
+    int process(Buffer* dataOut);
     const char* name() { return "DabModulator"; }
 
     /* Required to get the timestamp */
-    EtiReader* getEtiReader() { return &myEtiReader; }
+    EtiSource* getEtiSource() { return &myEtiSource; }
 
 protected:
     void setMode(unsigned mode);
@@ -74,7 +74,7 @@ protected:
     GainMode myGainMode;
     float& myDigGain;
     float myNormalise;
-    EtiReader myEtiReader;
+    EtiSource& myEtiSource;
     Flowgraph* myFlowgraph;
     OutputMemory* myOutput;
     std::string myFilterTapsFilename;
diff --git a/src/EtiReader.cpp b/src/EtiReader.cpp
index 010124c..3207a1f 100644
--- a/src/EtiReader.cpp
+++ b/src/EtiReader.cpp
@@ -34,6 +34,7 @@
 #include <sys/types.h>
 #include <string.h>
 #include <arpa/inet.h>
+#include <regex>
 
 using namespace std;
 
@@ -56,17 +57,14 @@ EtiReader::EtiReader(
         double& tist_offset_s,
         unsigned tist_delay_stages) :
     state(EtiReaderStateSync),
-    myTimestampDecoder(tist_offset_s, tist_delay_stages)
+    myTimestampDecoder(tist_offset_s, tist_delay_stages),
+    myCurrentFrame(0),
+    eti_fc_valid(false)
 {
-    PDEBUG("EtiReader::EtiReader()\n");
-
     rcs.enrol(&myTimestampDecoder);
-
-    myCurrentFrame = 0;
-    eti_fc_valid = false;
 }
 
-std::shared_ptr<FicSource>& EtiReader::getFic()
+std::shared_ptr<FicSource>& EtiSource::getFic()
 {
     return myFicSource;
 }
@@ -90,7 +88,7 @@ unsigned EtiReader::getFp()
 }
 
 
-const std::vector<std::shared_ptr<SubchannelSource> >& EtiReader::getSubchannels()
+const std::vector<std::shared_ptr<SubchannelSource> > EtiReader::getSubchannels() const
 {
     return mySources;
 }
@@ -158,7 +156,9 @@ int EtiReader::process(const Buffer* dataIn)
                 throw std::runtime_error("FIC must be present to modulate!");
             }
             if (not myFicSource) {
-                myFicSource = make_shared<FicSource>(eti_fc);
+                unsigned ficf = eti_fc.FICF;
+                unsigned mid = eti_fc.MID;
+                myFicSource = make_shared<FicSource>(ficf, mid);
             }
             break;
         case EtiReaderStateNst:
@@ -173,8 +173,12 @@ int EtiReader::process(const Buffer* dataIn)
 
                 mySources.clear();
                 for (unsigned i = 0; i < eti_fc.NST; ++i) {
+                    const auto tpl = eti_stc[i].TPL;
                     mySources.push_back(
-                            make_shared<SubchannelSource>(eti_stc[i]));
+                            make_shared<SubchannelSource>(
+                                eti_stc[i].getStartAddress(),
+                                eti_stc[i].getSTL(),
+                                tpl));
                     PDEBUG("Sstc %u:\n", i);
                     PDEBUG(" Stc%i.scid: %i\n", i, eti_stc[i].SCID);
                     PDEBUG(" Stc%i.sad: %u\n", i, eti_stc[i].getStartAddress());
@@ -300,3 +304,247 @@ uint32_t EtiReader::getPPSOffset()
     return timestamp;
 }
 
+
+unsigned EdiReader::getMode()
+{
+    if (not m_fc_valid) {
+        assert(false);
+        throw std::runtime_error("Trying to access Mode before it is ready!");
+    }
+    return m_fc.mid;
+}
+
+
+unsigned EdiReader::getFp()
+{
+    if (not m_fc_valid) {
+        throw std::runtime_error("Trying to access FP before it is ready!");
+    }
+    return m_fc.fp;
+}
+
+const std::vector<std::shared_ptr<SubchannelSource> > EdiReader::getSubchannels() const
+{
+    std::vector<std::shared_ptr<SubchannelSource> > sources;
+
+    sources.resize(m_sources.size());
+    for (const auto s : m_sources) {
+        if (s.first < sources.size()) {
+            sources.at(s.first) = s.second;
+        }
+        else {
+            throw std::runtime_error("Missing subchannel data in EDI source");
+        }
+    }
+
+    return sources;
+}
+
+bool EdiReader::isFrameReady()
+{
+    return m_frameReady;
+}
+
+void EdiReader::clearFrame()
+{
+    m_frameReady = false;
+    m_proto_valid = false;
+    m_fc_valid = false;
+    m_fic.clear();
+}
+
+void EdiReader::update_protocol(
+        const std::string& proto,
+        uint16_t major,
+        uint16_t minor)
+{
+    m_proto_valid = (proto == "DETI" and major == 0 and minor == 0);
+
+    if (not m_proto_valid) {
+        throw std::invalid_argument("Wrong EDI protocol");
+    }
+}
+
+void EdiReader::update_err(uint8_t err)
+{
+    if (not m_proto_valid) {
+        throw std::logic_error("Cannot update ERR before protocol");
+    }
+    m_err = err;
+}
+
+void EdiReader::update_fc_data(const EdiDecoder::eti_fc_data& fc_data)
+{
+    if (not m_proto_valid) {
+        throw std::logic_error("Cannot update FC before protocol");
+    }
+
+    m_fc_valid = false;
+    m_fc = fc_data;
+
+    if (not m_fc.ficf) {
+        throw std::invalid_argument("FIC must be present");
+    }
+
+    if (m_fc.mid > 4) {
+        throw std::invalid_argument("Invalid MID");
+    }
+
+    if (m_fc.fp > 7) {
+        throw std::invalid_argument("Invalid FP");
+    }
+
+    m_fc_valid = true;
+}
+
+void EdiReader::update_fic(const std::vector<uint8_t>& fic)
+{
+    if (not m_proto_valid) {
+        throw std::logic_error("Cannot update FIC before protocol");
+    }
+    m_fic = fic;
+}
+
+void EdiReader::update_edi_time(
+        uint32_t utco,
+        uint32_t seconds)
+{
+    if (not m_proto_valid) {
+        throw std::logic_error("Cannot update time before protocol");
+    }
+
+    m_utco = utco;
+    m_seconds = seconds;
+
+    // TODO check validity
+    m_time_valid = true;
+}
+
+void EdiReader::update_mnsc(uint16_t mnsc)
+{
+    if (not m_proto_valid) {
+        throw std::logic_error("Cannot update MNSC before protocol");
+    }
+
+    m_mnsc = mnsc;
+}
+
+void EdiReader::update_rfu(uint16_t rfu)
+{
+    if (not m_proto_valid) {
+        throw std::logic_error("Cannot update RFU before protocol");
+    }
+
+    m_rfu = rfu;
+}
+
+void EdiReader::add_subchannel(const EdiDecoder::eti_stc_data& stc)
+{
+    if (not m_proto_valid) {
+        throw std::logic_error("Cannot add subchannel before protocol");
+    }
+
+    if (m_sources.count(stc.stream_index) == 0) {
+        m_sources[stc.stream_index] = make_shared<SubchannelSource>(stc.sad, stc.stl(), stc.tpl);
+    }
+
+    auto& source = m_sources[stc.stream_index];
+
+    if (source->framesize() != stc.mst.size()) {
+        throw std::invalid_argument(
+                "EDI: MST data length inconsistent with FIC");
+    }
+    source->loadSubchannelData(stc.mst);
+
+    if (m_sources.size() > 64) {
+        throw std::invalid_argument("Too many subchannels");
+    }
+}
+
+void EdiReader::assemble()
+{
+    etiLog.level(debug) << "Calling assemble";
+    if (not m_proto_valid) {
+        throw std::logic_error("Cannot assemble EDI data before protocol");
+    }
+
+    if (not m_fc_valid) {
+        throw std::logic_error("Cannot assemble EDI data without FC");
+    }
+
+    if (m_fic.empty()) {
+        throw std::logic_error("Cannot assemble EDI data without FIC");
+    }
+
+    // ETS 300 799 Clause 5.3.2, but we don't support not having
+    // a FIC
+    if (    (m_fc.mid == 3 and m_fic.size() != 32 * 4) or
+            (m_fc.mid != 3 and m_fic.size() != 24 * 4) ) {
+        stringstream ss;
+        ss << "Invalid FIC length " << m_fic.size() <<
+            " for MID " << m_fc.mid;
+        throw std::invalid_argument(ss.str());
+    }
+
+    if (not myFicSource) {
+        myFicSource = make_shared<FicSource>(m_fc.ficf, m_fc.mid);
+    }
+
+    myFicSource->loadFicData(m_fic);
+
+    // Accept zero subchannels, because of an edge-case that can happen
+    // during reconfiguration. See ETS 300 799 Clause 5.3.3
+
+    // TODO check time validity
+
+    /* TODO timestamp
+    myTimestampDecoder.updateTimestampEti(
+            eti_fc.FP & 0x3,
+            eti_eoh.MNSC, getPPSOffset(), eti_fc.FCT);
+            */
+    m_frameReady = true;
+}
+
+EdiUdpInput::EdiUdpInput(EdiDecoder::ETIDecoder& decoder) :
+    m_enabled(false),
+    m_port(0),
+    m_decoder(decoder) { }
+
+int EdiUdpInput::Open(const std::string& uri)
+{
+    etiLog.level(info) << "Opening EDI :" << uri;
+
+    int ret = 1;
+
+    const std::regex re_udp("udp://:([0-9]+)");
+    std::smatch m;
+    if (std::regex_match(uri, m, re_udp)) {
+        m_port = std::stoi(m[1].str());
+
+        etiLog.level(info) << "EDI port :" << m_port;
+        ret = m_sock.reinit(m_port, "0.0.0.0");
+        m_enabled = (ret == 0);
+    }
+
+    return ret;
+}
+
+void EdiUdpInput::rxPacket()
+{
+    const size_t packsize = 8192;
+    UdpPacket packet(packsize);
+
+    int ret = m_sock.receive(packet);
+    if (ret == 0) {
+        const auto &buf = packet.getBuffer();
+        if (packet.getSize() == packsize) {
+            fprintf(stderr, "Warning, possible UDP truncation\n");
+        }
+
+        m_decoder.push_packet(buf);
+    }
+    else {
+        fprintf(stderr, "Socket error: %s\n", inetErrMsg);
+    }
+}
+
diff --git a/src/EtiReader.h b/src/EtiReader.h
index cfc03af..04d627d 100644
--- a/src/EtiReader.h
+++ b/src/EtiReader.h
@@ -36,6 +36,8 @@
 #include "FicSource.h"
 #include "SubchannelSource.h"
 #include "TimestampDecoder.h"
+#include "lib/edi/ETIDecoder.hpp"
+#include "lib/UdpSocket.h"
 
 #include <vector>
 #include <memory>
@@ -43,26 +45,42 @@
 #include <sys/types.h>
 
 
-class EtiReader
+class EtiSource
+{
+public:
+    virtual unsigned getMode() = 0;
+    virtual unsigned getFp() = 0;
+
+    virtual bool sourceContainsTimestamp() = 0;
+    virtual void calculateTimestamp(struct frame_timestamp& ts) = 0;
+
+    virtual std::shared_ptr<FicSource>& getFic(void);
+    virtual const std::vector<std::shared_ptr<SubchannelSource> > getSubchannels() const = 0;
+
+protected:
+    std::shared_ptr<FicSource> myFicSource;
+};
+
+class EtiReader : public EtiSource
 {
 public:
     EtiReader(
             double& tist_offset_s,
             unsigned tist_delay_stages);
 
-    std::shared_ptr<FicSource>& getFic();
-    unsigned getMode();
-    unsigned getFp();
-    const std::vector<std::shared_ptr<SubchannelSource> >& getSubchannels();
+    virtual unsigned getMode();
+    virtual unsigned getFp();
     int process(const Buffer* dataIn);
 
-    void calculateTimestamp(struct frame_timestamp& ts)
+    virtual void calculateTimestamp(struct frame_timestamp& ts)
     {
         myTimestampDecoder.calculateTimestamp(ts);
     }
 
     /* Returns true if we have valid time stamps in the ETI*/
-    bool sourceContainsTimestamp();
+    virtual bool sourceContainsTimestamp();
+
+    virtual const std::vector<std::shared_ptr<SubchannelSource> > getSubchannels() const;
 
 private:
     /* Transform the ETI TIST to a PPS offset in units of 1/16384000 s */
@@ -78,13 +96,93 @@ private:
     eti_EOH eti_eoh;
     eti_EOF eti_eof;
     eti_TIST eti_tist;
-    std::shared_ptr<FicSource> myFicSource;
-    std::vector<std::shared_ptr<SubchannelSource> > mySources;
     TimestampDecoder myTimestampDecoder;
 
     size_t myCurrentFrame;
     bool eti_fc_valid;
+
+    std::vector<std::shared_ptr<SubchannelSource> > mySources;
+};
+
+class EdiReader : public EtiSource, public EdiDecoder::DataCollector
+{
+public:
+    virtual unsigned getMode();
+    virtual unsigned getFp();
+    virtual bool sourceContainsTimestamp() { return false; }
+    virtual void calculateTimestamp(struct frame_timestamp& ts)
+    { /* TODO */ }
+    virtual const std::vector<std::shared_ptr<SubchannelSource> > getSubchannels() const;
+
+    virtual bool isFrameReady(void);
+    virtual void clearFrame(void);
+
+    // Tell the ETIWriter what EDI protocol we receive in *ptr.
+    // This is not part of the ETI data, but is used as check
+    virtual void update_protocol(
+            const std::string& proto,
+            uint16_t major,
+            uint16_t minor);
+
+    // Update the data for the frame characterisation
+    virtual void update_fc_data(const EdiDecoder::eti_fc_data& fc_data);
+
+    virtual void update_fic(const std::vector<uint8_t>& fic);
+
+    virtual void update_err(uint8_t err);
+
+    // In addition to TSTA in ETI, EDI also transports more time
+    // stamp information.
+    virtual void update_edi_time(
+            uint32_t utco,
+            uint32_t seconds);
+
+    virtual void update_mnsc(uint16_t mnsc);
+
+    virtual void update_rfu(uint16_t rfu);
+
+    virtual void add_subchannel(const EdiDecoder::eti_stc_data& stc);
+
+    // Tell the ETIWriter that the AFPacket is complete
+    virtual void assemble(void);
+private:
+    bool m_proto_valid = false;
+    bool m_frameReady = false;
+
+    uint8_t m_err;
+
+    bool m_fc_valid = false;
+    EdiDecoder::eti_fc_data m_fc;
+
+    std::vector<uint8_t> m_fic;
+
+    bool m_time_valid = false;
+    uint32_t m_utco;
+    uint32_t m_seconds;
+
+    uint16_t m_mnsc = 0xffff;
+
+    // 16 bits: RFU field in EOH
+    uint16_t m_rfu = 0xffff;
+
+    std::map<uint8_t, std::shared_ptr<SubchannelSource> > m_sources;
 };
 
+class EdiUdpInput {
+    public:
+        EdiUdpInput(EdiDecoder::ETIDecoder& decoder);
 
+        int Open(const std::string& uri);
+
+        bool isEnabled(void) const { return m_enabled; }
+
+        void rxPacket(void);
+
+    private:
+        bool m_enabled;
+        int m_port;
+
+        UdpSocket m_sock;
+        EdiDecoder::ETIDecoder& m_decoder;
+};
 
diff --git a/src/FicSource.cpp b/src/FicSource.cpp
index 85614f6..92932ec 100644
--- a/src/FicSource.cpp
+++ b/src/FicSource.cpp
@@ -39,7 +39,7 @@ const std::vector<PuncturingRule>& FicSource::get_rules()
 }
 
 
-FicSource::FicSource(eti_FC &fc) :
+FicSource::FicSource(unsigned ficf, unsigned mid) :
     ModInput()
 {
 //    PDEBUG("FicSource::FicSource(...)\n");
@@ -47,13 +47,13 @@ FicSource::FicSource(eti_FC &fc) :
 //    PDEBUG("  Framesize: %i\n", d_framesize);
 //    PDEBUG("  Protection: %i\n", d_protection);
 
-    if (fc.FICF == 0) {
+    if (ficf == 0) {
         d_framesize = 0;
         d_buffer.setLength(0);
         return;
     }
 
-    if (fc.MID == 3) {
+    if (mid == 3) {
         d_framesize = 32 * 4;
         d_puncturing_rules.emplace_back(29 * 16, 0xeeeeeeee);
         d_puncturing_rules.emplace_back(3 * 16, 0xeeeeeeec);
diff --git a/src/FicSource.h b/src/FicSource.h
index cb1d3a8..77ac741 100644
--- a/src/FicSource.h
+++ b/src/FicSource.h
@@ -39,7 +39,7 @@
 class FicSource : public ModInput
 {
 public:
-    FicSource(eti_FC &fc);
+    FicSource(unsigned ficf, unsigned mid);
 
     size_t getFramesize();
     const std::vector<PuncturingRule>& get_rules();
diff --git a/src/FrameMultiplexer.cpp b/src/FrameMultiplexer.cpp
index c8ee299..d16ee44 100644
--- a/src/FrameMultiplexer.cpp
+++ b/src/FrameMultiplexer.cpp
@@ -2,7 +2,7 @@
    Copyright (C) 2005, 2006, 2007, 2008, 2009, 2010, 2011 Her Majesty
    the Queen in Right of Canada (Communications Research Center Canada)
 
-   Copyright (C) 2016
+   Copyright (C) 2017
    Matthias P. Braendli, matthias.braendli@mpb.li
 
     http://opendigitalradio.org
@@ -37,24 +37,12 @@
 typedef std::complex<float> complexf;
 
 FrameMultiplexer::FrameMultiplexer(
-        size_t framesize,
-        const std::vector<std::shared_ptr<SubchannelSource> >& subchannels) :
+        const EtiSource& etiSource) :
     ModMux(),
-    d_frameSize(framesize),
-    mySubchannels(subchannels)
+    m_etiSource(etiSource)
 {
-    PDEBUG("FrameMultiplexer::FrameMultiplexer(%zu) @ %p\n", framesize, this);
-
-}
-
-
-FrameMultiplexer::~FrameMultiplexer()
-{
-    PDEBUG("FrameMultiplexer::~FrameMultiplexer() @ %p\n", this);
-
 }
 
-
 // dataIn[0] -> PRBS
 // dataIn[1+] -> subchannels
 int FrameMultiplexer::process(std::vector<Buffer*> dataIn, Buffer* dataOut)
@@ -81,12 +69,14 @@ int FrameMultiplexer::process(std::vector<Buffer*> dataIn, Buffer* dataOut)
     // Write PRBS
     memcpy(out, (*in)->getData(), (*in)->getLength());
     ++in;
+
     // Write subchannel
-    if (mySubchannels.size() != dataIn.size() - 1) {
+    const auto subchannels = m_etiSource.getSubchannels();
+    if (subchannels.size() != dataIn.size() - 1) {
         throw std::out_of_range(
                 "FrameMultiplexer detected subchannel size change!");
     }
-    auto subchannel = mySubchannels.begin();
+    auto subchannel = subchannels.begin();
     while (in != dataIn.end()) {
         if ((*subchannel)->framesizeCu() * 8 != (*in)->getLength()) {
             throw std::out_of_range(
diff --git a/src/FrameMultiplexer.h b/src/FrameMultiplexer.h
index e01f4bf..680cdc7 100644
--- a/src/FrameMultiplexer.h
+++ b/src/FrameMultiplexer.h
@@ -2,7 +2,7 @@
    Copyright (C) 2005, 2006, 2007, 2008, 2009, 2010, 2011 Her Majesty
    the Queen in Right of Canada (Communications Research Center Canada)
 
-   Copyright (C) 2016
+   Copyright (C) 2017
    Matthias P. Braendli, matthias.braendli@mpb.li
 
     http://opendigitalradio.org
@@ -33,6 +33,7 @@
 
 #include "ModPlugin.h"
 #include "SubchannelSource.h"
+#include "EtiReader.h"
 #include <memory>
 
 #include <sys/types.h>
@@ -42,19 +43,13 @@ class FrameMultiplexer : public ModMux
 {
 public:
     FrameMultiplexer(
-            size_t frameSize,
-            const std::vector<std::shared_ptr<SubchannelSource> >& subchannels);
-    virtual ~FrameMultiplexer();
-    FrameMultiplexer(const FrameMultiplexer&) = delete;
-    FrameMultiplexer& operator=(const FrameMultiplexer&) = delete;
-
+            const EtiSource& etiSource);
 
     int process(std::vector<Buffer*> dataIn, Buffer* dataOut);
     const char* name() { return "FrameMultiplexer"; }
 
 protected:
-    size_t d_frameSize;
-    const std::vector<std::shared_ptr<SubchannelSource> >& mySubchannels;
+    const EtiSource& m_etiSource;
 };
 
 
diff --git a/src/InputReader.h b/src/InputReader.h
index fddfccf..f0e7197 100644
--- a/src/InputReader.h
+++ b/src/InputReader.h
@@ -245,28 +245,5 @@ class InputZeroMQReader : public InputReader
         struct InputZeroMQThreadData workerdata_;
 };
 
-class InputEdiReader : public InputReader
-{
-public:
-    InputEdiReader();
-
-    int Open(const std::string& uri);
-
-    int GetNextFrame(void* buffer);
-
-    void PrintInfo(void);
-
-private:
-    void rx_packet(void);
-
-    std::vector<uint8_t> getEtiFrame(void);
-
-    EdiDecoder::ETIWriter m_writer;
-    EdiDecoder::ETIDecoder m_decoder;
-
-    int m_port;
-    UdpSocket m_sock;
-};
-
 #endif
 
diff --git a/src/OutputUHD.cpp b/src/OutputUHD.cpp
index 55a24b2..26283a8 100644
--- a/src/OutputUHD.cpp
+++ b/src/OutputUHD.cpp
@@ -273,9 +273,9 @@ OutputUHD::~OutputUHD()
 }
 
 
-void OutputUHD::setETIReader(EtiReader *etiReader)
+void OutputUHD::setETISource(EtiSource *etiSource)
 {
-    myEtiReader = etiReader;
+    myEtiSource = etiSource;
 }
 
 int transmission_frame_duration_ms(unsigned int dabMode)
@@ -331,7 +331,7 @@ int OutputUHD::process(Buffer* dataIn)
             // we only set the delay buffer from the dab mode signaled in ETI if the
             // dab mode was not set in contructor
             if (myTFDurationMs == 0) {
-                SetDelayBuffer(myEtiReader->getMode());
+                SetDelayBuffer(myEtiSource->getMode());
             }
 
             worker.start(&uwd);
@@ -350,7 +350,7 @@ int OutputUHD::process(Buffer* dataIn)
         }
 
         uwd.sourceContainsTimestamp = myConf.enableSync &&
-            myEtiReader->sourceContainsTimestamp();
+            myEtiSource->sourceContainsTimestamp();
 
 
         if (uwd.check_gpsfix) {
@@ -387,7 +387,7 @@ int OutputUHD::process(Buffer* dataIn)
                     frame.buf.begin());
         }
 
-        myEtiReader->calculateTimestamp(frame.ts);
+        myEtiSource->calculateTimestamp(frame.ts);
 
         if (!uwd.running) {
             worker.stop();
diff --git a/src/OutputUHD.h b/src/OutputUHD.h
index 9987c79..ed68aaa 100644
--- a/src/OutputUHD.h
+++ b/src/OutputUHD.h
@@ -225,7 +225,7 @@ class OutputUHD: public ModOutput, public RemoteControllable {
 
         const char* name() { return "OutputUHD"; }
 
-        void setETIReader(EtiReader *etiReader);
+        void setETISource(EtiSource *etiSource);
 
         /*********** REMOTE CONTROL ***************/
 
@@ -242,7 +242,7 @@ class OutputUHD: public ModOutput, public RemoteControllable {
         OutputUHD(const OutputUHD& other) = delete;
         OutputUHD& operator=(const OutputUHD& other) = delete;
 
-        EtiReader *myEtiReader;
+        EtiSource *myEtiSource;
         OutputUHDConfig& myConf;
         uhd::usrp::multi_usrp::sptr myUsrp;
         std::shared_ptr<boost::barrier> mySyncBarrier;
diff --git a/src/SubchannelSource.cpp b/src/SubchannelSource.cpp
index 7632de5..82636f9 100644
--- a/src/SubchannelSource.cpp
+++ b/src/SubchannelSource.cpp
@@ -65,11 +65,15 @@ const std::vector<PuncturingRule>& SubchannelSource::get_rules() const
 }
 
 
-SubchannelSource::SubchannelSource(eti_STC &stc) :
+SubchannelSource::SubchannelSource(
+            uint8_t sad,
+            uint16_t stl,
+            uint8_t tpl
+            ) :
     ModInput(),
-    d_start_address(stc.getStartAddress()),
-    d_framesize(stc.getSTL() * 8),
-    d_protection(stc.TPL)
+    d_start_address(sad),
+    d_framesize(stl * 8),
+    d_protection(tpl)
 {
     PDEBUG("SubchannelSource::SubchannelSource(...) @ %p\n", this);
     PDEBUG("  Start address: %zu\n", d_start_address);
@@ -109,7 +113,7 @@ SubchannelSource::SubchannelSource(eti_STC &stc) :
                 fprintf(stderr,
                         "Protection form(%zu), option(%zu) and level(%zu)\n",
                         protectionForm(), protectionOption(), protectionLevel());
-                fprintf(stderr, "Subchannel TPL: 0x%x (%u)\n", stc.TPL, stc.TPL);
+                fprintf(stderr, "Subchannel TPL: 0x%x (%u)\n", tpl, tpl);
                 throw std::runtime_error("SubchannelSource::SubchannelSource "
                         "unknown protection level!");
             }
@@ -143,7 +147,7 @@ SubchannelSource::SubchannelSource(eti_STC &stc) :
                 fprintf(stderr,
                         "Protection form(%zu), option(%zu) and level(%zu)\n",
                         protectionForm(), protectionOption(), protectionLevel());
-                fprintf(stderr, "Subchannel TPL: 0x%x (%u)\n", stc.TPL, stc.TPL);
+                fprintf(stderr, "Subchannel TPL: 0x%x (%u)\n", tpl, tpl);
                 throw std::runtime_error("SubchannelSource::SubchannelSource "
                         "unknown protection level!");
             }
@@ -151,7 +155,7 @@ SubchannelSource::SubchannelSource(eti_STC &stc) :
             fprintf(stderr,
                     "Protection form(%zu), option(%zu) and level(%zu)\n",
                     protectionForm(), protectionOption(), protectionLevel());
-            fprintf(stderr, "Subchannel TPL: 0x%x (%u)\n", stc.TPL, stc.TPL);
+            fprintf(stderr, "Subchannel TPL: 0x%x (%u)\n", tpl, tpl);
             throw std::runtime_error("SubchannelSource::SubchannelSource "
                     "unknown protection option!");
         }
diff --git a/src/SubchannelSource.h b/src/SubchannelSource.h
index a38c281..2dab941 100644
--- a/src/SubchannelSource.h
+++ b/src/SubchannelSource.h
@@ -41,7 +41,11 @@
 class SubchannelSource : public ModInput
 {
 public:
-    SubchannelSource(eti_STC &stc);
+    SubchannelSource(
+            uint8_t sad,
+            uint16_t stl,
+            uint8_t tpl
+            );
 
     size_t startAddress() const;
     size_t framesize() const;