Make DPD Feedback server SDRDevice-agnostic

9 files changed, 740 insertions, 92 deletions

diff --git a/src/output/Feedback.cpp b/src/output/Feedback.cpp
new file mode 100644
index 0000000..f0bbd98
--- /dev/null
+++ b/src/output/Feedback.cpp
@@ -0,0 +1,356 @@
+/*
+   Copyright (C) 2005, 2006, 2007, 2008, 2009, 2010 Her Majesty the
+   Queen in Right of Canada (Communications Research Center Canada)
+
+   Copyright (C) 2017
+   Matthias P. Braendli, matthias.braendli@mpb.li
+
+    http://opendigitalradio.org
+
+DESCRIPTION:
+   This presents a TCP socket to an external tool which calculates
+   a Digital Predistortion model from a short sequence of transmit
+   samples and corresponding receive samples.
+*/
+
+/*
+   This file is part of ODR-DabMod.
+
+   ODR-DabMod 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-DabMod 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-DabMod.  If not, see <http://www.gnu.org/licenses/>.
+ */
+
+#ifdef HAVE_CONFIG_H
+#   include <config.h>
+#endif
+
+#include <vector>
+#include <complex>
+#include <cstring>
+#include <sys/socket.h>
+#include <errno.h>
+#include <poll.h>
+#include <boost/date_time/posix_time/posix_time.hpp>
+#include "output/Feedback.h"
+#include "Utils.h"
+#include "Socket.h"
+
+using namespace std;
+
+namespace Output {
+
+DPDFeedbackServer::DPDFeedbackServer(
+        std::shared_ptr<SDRDevice> device,
+        uint16_t port,
+        uint32_t sampleRate) :
+    m_port(port),
+    m_sampleRate(sampleRate),
+    m_device(device)
+{
+    if (m_port) {
+        m_running.store(true);
+
+        rx_burst_thread = boost::thread(
+                &DPDFeedbackServer::ReceiveBurstThread, this);
+
+        burst_tcp_thread = boost::thread(
+                &DPDFeedbackServer::ServeFeedbackThread, this);
+    }
+}
+
+DPDFeedbackServer::~DPDFeedbackServer()
+{
+    m_running.store(false);
+
+    rx_burst_thread.interrupt();
+    if (rx_burst_thread.joinable()) {
+        rx_burst_thread.join();
+    }
+
+    burst_tcp_thread.interrupt();
+    if (burst_tcp_thread.joinable()) {
+        burst_tcp_thread.join();
+    }
+}
+
+void DPDFeedbackServer::set_tx_frame(
+        const std::vector<uint8_t> &buf,
+        const struct frame_timestamp &buf_ts)
+{
+    if (not m_running) {
+        throw runtime_error("DPDFeedbackServer not running");
+    }
+
+    boost::mutex::scoped_lock lock(burstRequest.mutex);
+
+    if (buf.size() % sizeof(complexf) != 0) {
+        throw std::logic_error("Buffer for tx frame has incorrect size");
+    }
+
+    if (burstRequest.state == BurstRequestState::SaveTransmitFrame) {
+        const size_t n = std::min(
+                burstRequest.num_samples * sizeof(complexf), buf.size());
+
+        burstRequest.num_samples = n / sizeof(complexf);
+
+        burstRequest.tx_samples.clear();
+        burstRequest.tx_samples.resize(n);
+        // A frame will always begin with the NULL symbol, which contains
+        // no power. Instead of taking n samples at the beginning of the
+        // frame, we take them at the end and adapt the timestamp accordingly.
+
+        const size_t start_ix = buf.size() - n;
+        copy(buf.begin() + start_ix, buf.end(), burstRequest.tx_samples.begin());
+
+        frame_timestamp ts = buf_ts;
+        ts += (1.0 * start_ix) / (sizeof(complexf) * m_sampleRate);
+
+        burstRequest.tx_second = ts.timestamp_sec;
+        burstRequest.tx_pps = ts.timestamp_pps;
+
+        // Prepare the next state
+        burstRequest.rx_second = ts.timestamp_sec;
+        burstRequest.rx_pps = ts.timestamp_pps;
+        burstRequest.state = BurstRequestState::SaveReceiveFrame;
+
+        lock.unlock();
+        burstRequest.mutex_notification.notify_one();
+    }
+    else {
+        lock.unlock();
+    }
+}
+
+void DPDFeedbackServer::ReceiveBurstThread()
+{
+    try {
+        set_thread_name("dpdreceiveburst");
+
+        while (m_running) {
+            boost::mutex::scoped_lock lock(burstRequest.mutex);
+            while (burstRequest.state != BurstRequestState::SaveReceiveFrame) {
+                if (not m_running) break;
+                burstRequest.mutex_notification.wait(lock);
+            }
+
+            if (not m_running) break;
+
+            const size_t num_samps = burstRequest.num_samples;
+
+            frame_timestamp ts;
+            ts.timestamp_sec = burstRequest.rx_second;
+            ts.timestamp_pps = burstRequest.rx_pps;
+            ts.timestamp_valid = true;
+
+            // We need to free the mutex while we recv(), because otherwise we block the
+            // TX thread
+            lock.unlock();
+
+            const double device_time = m_device->get_real_secs();
+            const double cmd_time = ts.get_real_secs();
+
+            std::vector<uint8_t> buf(num_samps * sizeof(complexf));
+
+            const double timeout = 60;
+            size_t samples_read = m_device->receive_frame(
+                    reinterpret_cast<complexf*>(buf.data()),
+                    num_samps, ts, timeout);
+
+            lock.lock();
+            burstRequest.rx_samples = std::move(buf);
+            burstRequest.rx_samples.resize(samples_read * sizeof(complexf));
+
+            // The recv might have happened at another time than requested
+            burstRequest.rx_second = ts.timestamp_sec;
+            burstRequest.rx_pps = ts.timestamp_pps;
+
+            etiLog.level(debug) << "DPD: acquired " << samples_read <<
+                " RX feedback samples " <<
+                "at time " << burstRequest.tx_second << " + " <<
+                std::fixed << burstRequest.tx_pps / 16384000.0 <<
+                " Delta=" << cmd_time - device_time;
+
+            burstRequest.state = BurstRequestState::Acquired;
+
+            lock.unlock();
+            burstRequest.mutex_notification.notify_one();
+        }
+    }
+    catch (const runtime_error &e) {
+        etiLog.level(error) << "DPD Feedback RX runtime error: " << e.what();
+    }
+    catch (const std::exception &e) {
+        etiLog.level(error) << "DPD Feedback RX exception: " << e.what();
+    }
+    catch (...) {
+        etiLog.level(error) << "DPD Feedback RX unknown exception!";
+    }
+
+    m_running.store(false);
+}
+
+void DPDFeedbackServer::ServeFeedback()
+{
+    TCPSocket m_server_sock;
+    m_server_sock.listen(m_port);
+
+    etiLog.level(info) << "DPD Feedback server listening on port " << m_port;
+
+    while (m_running) {
+        struct sockaddr_in client;
+        TCPSocket client_sock = m_server_sock.accept_with_timeout(1000, &client);
+
+        if (not client_sock.valid()) {
+            // No connection request received
+            continue;
+        }
+
+        uint8_t request_version = 0;
+        ssize_t read = client_sock.recv(&request_version, 1, 0);
+        if (!read) break; // done reading
+        if (read < 0) {
+            etiLog.level(info) <<
+                "DPD Feedback Server Client read request version failed: " << strerror(errno);
+            break;
+        }
+
+        if (request_version != 1) {
+            etiLog.level(info) << "DPD Feedback Server wrong request version";
+            break;
+        }
+
+        uint32_t num_samples = 0;
+        read = client_sock.recv(&num_samples, 4, 0);
+        if (!read) break; // done reading
+        if (read < 0) {
+            etiLog.level(info) <<
+                "DPD Feedback Server Client read num samples failed";
+            break;
+        }
+
+        // We are ready to issue the request now
+        {
+            boost::mutex::scoped_lock lock(burstRequest.mutex);
+            burstRequest.num_samples = num_samples;
+            burstRequest.state = BurstRequestState::SaveTransmitFrame;
+
+            lock.unlock();
+        }
+
+        // Wait for the result to be ready
+        boost::mutex::scoped_lock lock(burstRequest.mutex);
+        while (burstRequest.state != BurstRequestState::Acquired) {
+            if (not m_running) break;
+            burstRequest.mutex_notification.wait(lock);
+        }
+
+        burstRequest.state = BurstRequestState::None;
+        lock.unlock();
+
+        burstRequest.num_samples = std::min(burstRequest.num_samples,
+                std::min(
+                    burstRequest.tx_samples.size() / sizeof(complexf),
+                    burstRequest.rx_samples.size() / sizeof(complexf)));
+
+        uint32_t num_samples_32 = burstRequest.num_samples;
+        if (client_sock.sendall(&num_samples_32, sizeof(num_samples_32)) < 0) {
+            etiLog.level(info) <<
+                "DPD Feedback Server Client send num_samples failed";
+            break;
+        }
+
+        if (client_sock.sendall(
+                    &burstRequest.tx_second,
+                    sizeof(burstRequest.tx_second)) < 0) {
+            etiLog.level(info) <<
+                "DPD Feedback Server Client send tx_second failed";
+            break;
+        }
+
+        if (client_sock.sendall(
+                    &burstRequest.tx_pps,
+                    sizeof(burstRequest.tx_pps)) < 0) {
+            etiLog.level(info) <<
+                "DPD Feedback Server Client send tx_pps failed";
+            break;
+        }
+
+        const size_t frame_bytes = burstRequest.num_samples * sizeof(complexf);
+
+        if (burstRequest.tx_samples.size() < frame_bytes) {
+            throw logic_error("DPD Feedback burstRequest invalid: not enough TX samples");
+        }
+
+        if (client_sock.sendall(
+                    &burstRequest.tx_samples[0],
+                    frame_bytes) < 0) {
+            etiLog.level(info) <<
+                "DPD Feedback Server Client send tx_frame failed";
+            break;
+        }
+
+        if (client_sock.sendall(
+                    &burstRequest.rx_second,
+                    sizeof(burstRequest.rx_second)) < 0) {
+            etiLog.level(info) <<
+                "DPD Feedback Server Client send rx_second failed";
+            break;
+        }
+
+        if (client_sock.sendall(
+                    &burstRequest.rx_pps,
+                    sizeof(burstRequest.rx_pps)) < 0) {
+            etiLog.level(info) <<
+                "DPD Feedback Server Client send rx_pps failed";
+            break;
+        }
+
+        if (burstRequest.rx_samples.size() < frame_bytes) {
+            throw logic_error("DPD Feedback burstRequest invalid: not enough RX samples");
+        }
+
+        if (client_sock.sendall(
+                    &burstRequest.rx_samples[0],
+                    frame_bytes) < 0) {
+            etiLog.level(info) <<
+                "DPD Feedback Server Client send rx_frame failed";
+            break;
+        }
+    }
+}
+
+void DPDFeedbackServer::ServeFeedbackThread()
+{
+    set_thread_name("dpdfeedbackserver");
+
+    while (m_running) {
+        try {
+            ServeFeedback();
+        }
+        catch (const runtime_error &e) {
+            etiLog.level(error) << "DPD Feedback Server runtime error: " << e.what();
+        }
+        catch (const std::exception &e) {
+            etiLog.level(error) << "DPD Feedback Server exception: " << e.what();
+        }
+        catch (...) {
+            etiLog.level(error) << "DPD Feedback Server unknown exception!";
+        }
+
+        boost::this_thread::sleep(boost::posix_time::seconds(5));
+    }
+
+    m_running.store(false);
+}
+
+} // namespace Output
diff --git a/src/output/Feedback.h b/src/output/Feedback.h
new file mode 100644
index 0000000..2cad508
--- /dev/null
+++ b/src/output/Feedback.h
@@ -0,0 +1,116 @@
+/*
+   Copyright (C) 2005, 2006, 2007, 2008, 2009, 2010 Her Majesty the
+   Queen in Right of Canada (Communications Research Center Canada)
+
+   Copyright (C) 2017
+   Matthias P. Braendli, matthias.braendli@mpb.li
+
+    http://opendigitalradio.org
+
+DESCRIPTION:
+   This presents a TCP socket to an external tool which calculates
+   a Digital Predistortion model from a short sequence of transmit
+   samples and corresponding receive samples.
+*/
+
+/*
+   This file is part of ODR-DabMod.
+
+   ODR-DabMod 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-DabMod 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-DabMod.  If not, see <http://www.gnu.org/licenses/>.
+ */
+
+#pragma once
+
+#ifdef HAVE_CONFIG_H
+#   include <config.h>
+#endif
+
+#include <boost/thread.hpp>
+#include <memory>
+#include <string>
+#include <atomic>
+
+#include "Log.h"
+#include "TimestampDecoder.h"
+#include "output/SDRDevice.h"
+
+namespace Output {
+
+enum class BurstRequestState {
+    None, // To pending request
+    SaveTransmitFrame, // The TX thread has to save an outgoing frame
+    SaveReceiveFrame, // The RX thread has to save an incoming frame
+    Acquired, // Both TX and RX frames are ready
+};
+
+struct FeedbackBurstRequest {
+    // All fields in this struct are protected
+    mutable boost::mutex mutex;
+    boost::condition_variable mutex_notification;
+
+    BurstRequestState state = BurstRequestState::None;
+
+    // In the SaveTransmit states, num_samples complexf samples are saved into
+    // the vectors
+    size_t num_samples = 0;
+
+    // The timestamp of the first sample of the TX buffers
+    uint32_t tx_second = 0;
+    uint32_t tx_pps = 0; // in units of 1/16384000s
+
+    // Samples contain complexf, but since our internal representation is uint8_t
+    // we keep it like that
+    std::vector<uint8_t> tx_samples;
+
+    // The timestamp of the first sample of the RX buffers
+    uint32_t rx_second = 0;
+    uint32_t rx_pps = 0;
+
+    std::vector<uint8_t> rx_samples; // Also, actually complexf
+};
+
+// Serve TX samples and RX feedback samples over a TCP connection
+class DPDFeedbackServer {
+    public:
+        DPDFeedbackServer(
+                std::shared_ptr<SDRDevice> device,
+                uint16_t port, // Set to 0 to disable the Feedbackserver
+                uint32_t sampleRate);
+        DPDFeedbackServer(const DPDFeedbackServer& other) = delete;
+        DPDFeedbackServer& operator=(const DPDFeedbackServer& other) = delete;
+        ~DPDFeedbackServer();
+
+        void set_tx_frame(const std::vector<uint8_t> &buf,
+                const struct frame_timestamp& ts);
+
+    private:
+        // Thread that reacts to burstRequests and receives from the SDR device
+        void ReceiveBurstThread(void);
+
+        // Thread that listens for requests over TCP to get TX and RX feedback
+        void ServeFeedbackThread(void);
+        void ServeFeedback(void);
+
+        boost::thread rx_burst_thread;
+        boost::thread burst_tcp_thread;
+
+        FeedbackBurstRequest burstRequest;
+
+        std::atomic_bool m_running;
+        uint16_t m_port = 0;
+        uint32_t m_sampleRate = 0;
+        std::shared_ptr<SDRDevice> m_device;
+};
+
+} // namespace Output
diff --git a/src/output/SDR.cpp b/src/output/SDR.cpp
index 2ad8e57..66e93e2 100644
--- a/src/output/SDR.cpp
+++ b/src/output/SDR.cpp
@@ -67,6 +67,13 @@ SDR::SDR(SDRDeviceConfig& config, std::shared_ptr<SDRDevice> device) :
     m_config.muting = true;
 
     m_device_thread = std::thread(&SDR::process_thread_entry, this);
+
+    if (m_config.dpdFeedbackServerPort != 0) {
+        m_dpd_feedback_server = make_shared<DPDFeedbackServer>(
+                m_device,
+                m_config.dpdFeedbackServerPort,
+                m_config.sampleRate);
+    }
 }
 
 SDR::~SDR()
@@ -108,10 +115,21 @@ int SDR::process(Buffer *dataIn)
 
         if (frame.ts.fct == -1) {
             etiLog.level(info) <<
-                "OutputUHD: dropping one frame with invalid FCT";
+                "SDR output: dropping one frame with invalid FCT";
         }
         else {
-            // TODO setup Feedback and set tx_frame
+            try {
+                m_dpd_feedback_server->set_tx_frame(frame.buf, frame.ts);
+            }
+            catch (const runtime_error& e) {
+                etiLog.level(warn) <<
+                    "SDR output: Feedback server failed, restarting...";
+
+                m_dpd_feedback_server = std::make_shared<DPDFeedbackServer>(
+                        m_device,
+                        m_config.dpdFeedbackServerPort,
+                        m_config.sampleRate);
+            }
 
             size_t num_frames = m_queue.push_wait_if_full(frame,
                     FRAMES_MAX_SIZE);
diff --git a/src/output/SDR.h b/src/output/SDR.h
index 370bc38..5593ce3 100644
--- a/src/output/SDR.h
+++ b/src/output/SDR.h
@@ -36,92 +36,13 @@ DESCRIPTION:
 
 #include "ModPlugin.h"
 #include "EtiReader.h"
+#include "output/SDRDevice.h"
+#include "output/Feedback.h"
 
 namespace Output {
 
 using complexf = std::complex<float>;
 
-enum refclk_lock_loss_behaviour_t { CRASH, IGNORE };
-
-/* This structure is used as initial configuration for all SDR devices.
- * It must also contain all remote-controllable settings, otherwise
- * they will get lost on a modulator restart. */
-struct SDRDeviceConfig {
-    std::string device;
-    std::string subDevice; // For UHD
-
-    long masterClockRate = 32768000;
-    unsigned sampleRate = 2048000;
-    double frequency = 0.0;
-    double lo_offset = 0.0;
-    double txgain = 0.0;
-    double rxgain = 0.0;
-    bool enableSync = false;
-
-    // When working with timestamps, mute the frames that
-    // do not have a timestamp
-    bool muteNoTimestamps = false;
-    unsigned dabMode = 0;
-    unsigned maxGPSHoldoverTime = 0;
-
-    /* allowed values for UHD : auto, int, sma, mimo */
-    std::string refclk_src;
-
-    /* allowed values for UHD : int, sma, mimo */
-    std::string pps_src;
-
-    /* allowed values for UHD : pos, neg */
-    std::string pps_polarity;
-
-    /* What to do when the reference clock PLL loses lock */
-    refclk_lock_loss_behaviour_t refclk_lock_loss_behaviour;
-
-    // muting can only be changed using the remote control
-    bool muting = false;
-
-    // TCP port on which to serve TX and RX samples for the
-    // digital pre distortion learning tool
-    uint16_t dpdFeedbackServerPort = 0;
-};
-
-// Each frame contains one OFDM frame, and its
-// associated timestamp
-struct FrameData {
-    // Buffer holding frame data
-    std::vector<uint8_t> buf;
-
-    // A full timestamp contains a TIST according to standard
-    // and time information within MNSC with tx_second.
-    struct frame_timestamp ts;
-};
-
-
-// All SDR Devices must implement the SDRDevice interface
-class SDRDevice {
-    public:
-        struct RunStatistics {
-            size_t num_underruns;
-            size_t num_late_packets;
-            size_t num_overruns;
-            size_t num_frames_modulated; //TODO increment
-        };
-
-        // TODO make some functions const
-        virtual void tune(double lo_offset, double frequency) = 0;
-        virtual double get_tx_freq(void) = 0;
-        virtual void set_txgain(double txgain) = 0;
-        virtual double get_txgain(void) = 0;
-        virtual void transmit_frame(const struct FrameData& frame) = 0;
-        virtual RunStatistics get_run_statistics(void) = 0;
-        virtual double get_real_secs(void) = 0;
-
-
-        // Return true if GPS and reference clock inputs are ok
-        virtual bool is_clk_source_ok(void) = 0;
-
-        virtual const char* device_name(void) = 0;
-};
-
 class SDR : public ModOutput, public RemoteControllable {
     public:
         SDR(SDRDeviceConfig& config, std::shared_ptr<SDRDevice> device);
@@ -159,6 +80,8 @@ class SDR : public ModOutput, public RemoteControllable {
         std::shared_ptr<SDRDevice> m_device;
         std::string m_name;
 
+        std::shared_ptr<DPDFeedbackServer> m_dpd_feedback_server;
+
         EtiSource *m_eti_source = nullptr;
         bool     sourceContainsTimestamp = false;
         bool     last_tx_time_initialised = false;
diff --git a/src/output/SDRDevice.h b/src/output/SDRDevice.h
new file mode 100644
index 0000000..856233b
--- /dev/null
+++ b/src/output/SDRDevice.h
@@ -0,0 +1,136 @@
+/*
+   Copyright (C) 2005, 2006, 2007, 2008, 2009, 2010 Her Majesty the
+   Queen in Right of Canada (Communications Research Center Canada)
+
+   Copyright (C) 2017
+   Matthias P. Braendli, matthias.braendli@mpb.li
+
+    http://opendigitalradio.org
+
+DESCRIPTION:
+   Common interface for all SDR outputs
+*/
+
+/*
+   This file is part of ODR-DabMod.
+
+   ODR-DabMod 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-DabMod 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-DabMod.  If not, see <http://www.gnu.org/licenses/>.
+ */
+
+#pragma once
+
+#ifdef HAVE_CONFIG_H
+#   include <config.h>
+#endif
+
+#include <cstdint>
+#include <string>
+#include <vector>
+#include <complex>
+
+#include "TimestampDecoder.h"
+
+namespace Output {
+
+enum refclk_lock_loss_behaviour_t { CRASH, IGNORE };
+
+using complexf = std::complex<float>;
+
+/* This structure is used as initial configuration for all SDR devices.
+ * It must also contain all remote-controllable settings, otherwise
+ * they will get lost on a modulator restart. */
+struct SDRDeviceConfig {
+    std::string device;
+    std::string subDevice; // For UHD
+
+    long masterClockRate = 32768000;
+    unsigned sampleRate = 2048000;
+    double frequency = 0.0;
+    double lo_offset = 0.0;
+    double txgain = 0.0;
+    double rxgain = 0.0;
+    bool enableSync = false;
+
+    // When working with timestamps, mute the frames that
+    // do not have a timestamp
+    bool muteNoTimestamps = false;
+    unsigned dabMode = 0;
+    unsigned maxGPSHoldoverTime = 0;
+
+    /* allowed values for UHD : auto, int, sma, mimo */
+    std::string refclk_src;
+
+    /* allowed values for UHD : int, sma, mimo */
+    std::string pps_src;
+
+    /* allowed values for UHD : pos, neg */
+    std::string pps_polarity;
+
+    /* What to do when the reference clock PLL loses lock */
+    refclk_lock_loss_behaviour_t refclk_lock_loss_behaviour;
+
+    // muting can only be changed using the remote control
+    bool muting = false;
+
+    // TCP port on which to serve TX and RX samples for the
+    // digital pre distortion learning tool
+    uint16_t dpdFeedbackServerPort = 0;
+};
+
+// Each frame contains one OFDM frame, and its
+// associated timestamp
+struct FrameData {
+    // Buffer holding frame data
+    std::vector<uint8_t> buf;
+
+    // A full timestamp contains a TIST according to standard
+    // and time information within MNSC with tx_second.
+    struct frame_timestamp ts;
+};
+
+
+// All SDR Devices must implement the SDRDevice interface
+class SDRDevice {
+    public:
+        struct RunStatistics {
+            size_t num_underruns;
+            size_t num_late_packets;
+            size_t num_overruns;
+            size_t num_frames_modulated; //TODO increment
+        };
+
+        // TODO make some functions const
+        virtual void tune(double lo_offset, double frequency) = 0;
+        virtual double get_tx_freq(void) = 0;
+        virtual void set_txgain(double txgain) = 0;
+        virtual double get_txgain(void) = 0;
+        virtual void transmit_frame(const struct FrameData& frame) = 0;
+        virtual RunStatistics get_run_statistics(void) = 0;
+        virtual double get_real_secs(void) = 0;
+        virtual void set_rxgain(double rxgain) = 0;
+        virtual double get_rxgain(void) = 0;
+        virtual size_t receive_frame(
+                complexf *buf,
+                size_t num_samples,
+                struct frame_timestamp& ts,
+                double timeout_secs) = 0;
+
+
+        // Return true if GPS and reference clock inputs are ok
+        virtual bool is_clk_source_ok(void) = 0;
+
+        virtual const char* device_name(void) = 0;
+};
+
+} // namespace Output
diff --git a/src/output/Soapy.cpp b/src/output/Soapy.cpp
index d6b5953..e0a6f10 100644
--- a/src/output/Soapy.cpp
+++ b/src/output/Soapy.cpp
@@ -88,17 +88,18 @@ Soapy::Soapy(SDRDeviceConfig& config) :
     etiLog.level(info) << "SoapySDR:Actual tx gain: " <<
         m_device->getGain(SOAPY_SDR_TX, 0);
 
-    std::vector<size_t> channels;
-    channels.push_back(0);
-    m_stream = m_device->setupStream(SOAPY_SDR_TX, "CF32", channels);
-    m_device->activateStream(m_stream);
+    const std::vector<size_t> channels({0});
+    m_tx_stream = m_device->setupStream(SOAPY_SDR_TX, "CF32", channels);
+    m_device->activateStream(m_tx_stream);
+
+    m_rx_stream = m_device->setupStream(SOAPY_SDR_RX, "CF32", channels);
 }
 
 Soapy::~Soapy()
 {
     if (m_device != nullptr) {
-        if (m_stream != nullptr) {
-            m_device->closeStream(m_stream);
+        if (m_tx_stream != nullptr) {
+            m_device->closeStream(m_tx_stream);
         }
         SoapySDR::Device::unmake(m_device);
     }
@@ -156,6 +157,46 @@ double Soapy::get_real_secs(void)
     }
 }
 
+void Soapy::set_rxgain(double rxgain)
+{
+    m_device->setGain(SOAPY_SDR_RX, 0, m_conf.rxgain);
+    m_conf.rxgain = m_device->getGain(SOAPY_SDR_RX, 0);
+}
+
+double Soapy::get_rxgain(void)
+{
+    return m_device->getGain(SOAPY_SDR_RX, 0);
+}
+
+size_t Soapy::receive_frame(
+        complexf *buf,
+        size_t num_samples,
+        struct frame_timestamp& ts,
+        double timeout_secs)
+{
+    int flags = 0;
+    long long timeNs = ts.get_ns();
+    const size_t numElems = num_samples;
+
+    void *buffs[1];
+    buffs[0] = buf;
+
+    m_device->activateStream(m_rx_stream, flags, timeNs, numElems);
+
+    auto ret = m_device->readStream(m_tx_stream, buffs, num_samples, flags, timeNs);
+
+    m_device->deactivateStream(m_rx_stream);
+
+    // TODO update effective receive ts
+
+    if (ret < 0) {
+        throw runtime_error("Soapy readStream error: " + to_string(ret));
+    }
+
+    return ret;
+}
+
+
 bool Soapy::is_clk_source_ok()
 {
     // TODO
@@ -181,7 +222,7 @@ void Soapy::transmit_frame(const struct FrameData& frame)
     }
 
     // Stream MTU is in samples, not bytes.
-    const size_t mtu = m_device->getStreamMTU(m_stream);
+    const size_t mtu = m_device->getStreamMTU(m_tx_stream);
 
     size_t num_acc_samps = 0;
     while (num_acc_samps < numSamples) {
@@ -192,7 +233,7 @@ void Soapy::transmit_frame(const struct FrameData& frame)
 
         int flags = 0;
 
-        auto ret = m_device->writeStream(m_stream, buffs, samps_to_send, flags);
+        auto ret = m_device->writeStream(m_tx_stream, buffs, samps_to_send, flags);
 
         if (ret == SOAPY_SDR_TIMEOUT) {
             continue;
diff --git a/src/output/Soapy.h b/src/output/Soapy.h
index 97076b5..c603193 100644
--- a/src/output/Soapy.h
+++ b/src/output/Soapy.h
@@ -68,6 +68,14 @@ class Soapy : public Output::SDRDevice
         virtual RunStatistics get_run_statistics(void) override;
         virtual double get_real_secs(void) override;
 
+        virtual void set_rxgain(double rxgain) override;
+        virtual double get_rxgain(void) override;
+        virtual size_t receive_frame(
+                complexf *buf,
+                size_t num_samples,
+                struct frame_timestamp& ts,
+                double timeout_secs) override;
+
         // Return true if GPS and reference clock inputs are ok
         virtual bool is_clk_source_ok(void) override;
         virtual const char* device_name(void) override;
@@ -75,7 +83,8 @@ class Soapy : public Output::SDRDevice
     private:
         SDRDeviceConfig& m_conf;
         SoapySDR::Device *m_device = nullptr;
-        SoapySDR::Stream *m_stream = nullptr;
+        SoapySDR::Stream *m_tx_stream = nullptr;
+        SoapySDR::Stream *m_rx_stream = nullptr;
 
         size_t underflows = 0;
         size_t overflows = 0;
diff --git a/src/output/UHD.cpp b/src/output/UHD.cpp
index 13a0a57..3769c60 100644
--- a/src/output/UHD.cpp
+++ b/src/output/UHD.cpp
@@ -235,6 +235,10 @@ UHD::UHD(
             m_usrp, m_conf.dpdFeedbackServerPort, m_conf.sampleRate);
     */
 
+    const uhd::stream_args_t stream_args("fc32"); //complex floats
+    m_rx_stream = m_usrp->get_rx_stream(stream_args);
+    m_tx_stream = m_usrp->get_tx_stream(stream_args);
+
     MDEBUG("OutputUHD:UHD ready.\n");
 }
 
@@ -342,6 +346,42 @@ double UHD::get_real_secs(void)
     return m_usrp->get_time_now().get_real_secs();
 }
 
+void UHD::set_rxgain(double rxgain)
+{
+    m_usrp->set_rx_gain(m_conf.rxgain);
+    m_conf.rxgain = m_usrp->get_rx_gain();
+}
+
+double UHD::get_rxgain()
+{
+    return m_usrp->get_rx_gain();
+}
+
+size_t UHD::receive_frame(
+        complexf *buf,
+        size_t num_samples,
+        struct frame_timestamp& ts,
+        double timeout_secs)
+{
+    uhd::stream_cmd_t cmd(
+            uhd::stream_cmd_t::stream_mode_t::STREAM_MODE_NUM_SAMPS_AND_DONE);
+    cmd.num_samps = num_samples;
+    cmd.stream_now = false;
+    cmd.time_spec = uhd::time_spec_t(ts.timestamp_sec, ts.pps_offset());
+
+    m_rx_stream->issue_stream_cmd(cmd);
+
+    uhd::rx_metadata_t md;
+
+    constexpr double timeout = 60;
+    size_t samples_read = m_rx_stream->recv(buf, num_samples, md, timeout);
+
+    // Update the ts with the effective receive TS
+    ts.timestamp_sec = md.time_spec.get_full_secs();
+    ts.timestamp_pps = md.time_spec.get_frac_secs() * 16384000.0;
+    return samples_read;
+}
+
 // Return true if GPS and reference clock inputs are ok
 bool UHD::is_clk_source_ok(void)
 {
diff --git a/src/output/UHD.h b/src/output/UHD.h
index 220e4c8..3742924 100644
--- a/src/output/UHD.h
+++ b/src/output/UHD.h
@@ -84,6 +84,14 @@ class UHD : public Output::SDRDevice
         virtual RunStatistics get_run_statistics(void) override;
         virtual double get_real_secs(void) override;
 
+        virtual void set_rxgain(double rxgain) override;
+        virtual double get_rxgain(void) override;
+        virtual size_t receive_frame(
+                complexf *buf,
+                size_t num_samples,
+                struct frame_timestamp& ts,
+                double timeout_secs) override;
+
         // Return true if GPS and reference clock inputs are ok
         virtual bool is_clk_source_ok(void) override;
         virtual const char* device_name(void) override;
@@ -92,6 +100,7 @@ class UHD : public Output::SDRDevice
         SDRDeviceConfig& m_conf;
         uhd::usrp::multi_usrp::sptr m_usrp;
         uhd::tx_streamer::sptr m_tx_stream;
+        uhd::rx_streamer::sptr m_rx_stream;
 
         size_t num_underflows = 0;
         size_t num_overflows = 0;