From 03967733d70220e2de7af3cdad320aec5c82ede1 Mon Sep 17 00:00:00 2001 From: "Matthias P. Braendli" Date: Tue, 25 Jun 2019 10:50:23 +0200 Subject: Add more EDI input improvements --- src/input/Edi.h | 82 +++++++++++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 82 insertions(+) create mode 100644 src/input/Edi.h (limited to 'src/input/Edi.h') diff --git a/src/input/Edi.h b/src/input/Edi.h new file mode 100644 index 0000000..7b3dc04 --- /dev/null +++ b/src/input/Edi.h @@ -0,0 +1,82 @@ +/* + Copyright (C) 2009 Her Majesty the Queen in Right of Canada (Communications + Research Center Canada) + + Copyright (C) 2019 + 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 . + */ + +#pragma once + +#include +#include +#include +#include +#include +#include "Socket.h" +#include "input/inputs.h" +#include "edi/STIDecoder.hpp" +#include "edi/STIWriter.hpp" +#include "ThreadsafeQueue.h" + +namespace Inputs { + +/* + * Receives EDI from UDP or TCP in a separate thread and pushes that data + * into the STIDecoder. Complete frames are then put into a queue for the consumer. + * + * This way, the EDI decoding happens in a separate thread. + */ +class Edi : public InputBase { + public: + Edi(); + Edi(const Edi&) = delete; + Edi& operator=(const Edi&) = delete; + ~Edi(); + + virtual int open(const std::string& name); + virtual int readFrame(uint8_t* buffer, size_t size); + virtual int setBitrate(int bitrate); + virtual int close(); + + protected: + void m_run(); + + std::mutex m_mutex; + + enum class InputUsed { Invalid, UDP, TCP }; + InputUsed m_input_used = InputUsed::Invalid; + Socket::UDPSocket m_udp_sock; + Socket::TCPReceiveServer m_tcp_receive_server; + + EdiDecoder::STIWriter m_sti_writer; + EdiDecoder::STIDecoder m_sti_decoder; + std::thread m_thread; + std::atomic m_running = ATOMIC_VAR_INIT(false); + ThreadsafeQueue > m_frames; + + bool m_is_prebuffering = true; + + std::string m_name; +}; + +}; + -- cgit v1.2.3 From 14f69f9c915cf644147a52b803d79ff8f40a4ea1 Mon Sep 17 00:00:00 2001 From: "Matthias P. Braendli" Date: Mon, 29 Jul 2019 15:27:32 +0200 Subject: First prototype taking EDI TIST into account for contribution --- lib/edi/STIWriter.cpp | 1 + lib/edi/common.cpp | 25 ++++++++++++++++++++++++- lib/edi/common.hpp | 4 ++++ src/input/Edi.cpp | 48 ++++++++++++++++++++++++++++++++++++++++-------- src/input/Edi.h | 3 ++- src/utils.cpp | 2 +- 6 files changed, 72 insertions(+), 11 deletions(-) (limited to 'src/input/Edi.h') diff --git a/lib/edi/STIWriter.cpp b/lib/edi/STIWriter.cpp index 6964eb1..399922a 100644 --- a/lib/edi/STIWriter.cpp +++ b/lib/edi/STIWriter.cpp @@ -120,6 +120,7 @@ void STIWriter::assemble() m_stiFrame.frame = m_payload.istd; m_stiFrame.timestamp.seconds = m_seconds; m_stiFrame.timestamp.utco = m_utco; + m_stiFrame.timestamp.tsta = m_management_data.tsta; } sti_frame_t STIWriter::getFrame() diff --git a/lib/edi/common.cpp b/lib/edi/common.cpp index bc0fa1b..b4b0c79 100644 --- a/lib/edi/common.cpp +++ b/lib/edi/common.cpp @@ -25,18 +25,31 @@ #include #include #include +#include #include namespace EdiDecoder { using namespace std; +bool frame_timestamp_t::valid() const +{ + return tsta != 0xFFFFFF; +} + string frame_timestamp_t::to_string() const { const time_t seconds_in_unix_epoch = to_unix_epoch(); stringstream ss; - ss << "Timestamp: " << std::put_time(std::gmtime(&seconds_in_unix_epoch), "%c %Z"); + if (valid()) { + ss << "Timestamp: "; + } + else { + ss << "Timestamp not valid: "; + } + ss << std::put_time(std::gmtime(&seconds_in_unix_epoch), "%c %Z") << + " + " << ((double)tsta / 16384000.0); return ss.str(); } @@ -48,6 +61,16 @@ time_t frame_timestamp_t::to_unix_epoch() const return 946684800 + seconds - utco; } +std::chrono::system_clock::time_point frame_timestamp_t::to_system_clock() const +{ + auto ts = chrono::system_clock::from_time_t(to_unix_epoch()); + + // PPS offset in seconds = tsta / 16384000 + ts += chrono::nanoseconds(std::lrint(tsta / 0.016384)); + + return ts; +} + TagDispatcher::TagDispatcher( std::function&& af_packet_completed, bool verbose) : diff --git a/lib/edi/common.hpp b/lib/edi/common.hpp index 1433004..887bc3d 100644 --- a/lib/edi/common.hpp +++ b/lib/edi/common.hpp @@ -23,6 +23,7 @@ #include "PFT.hpp" #include #include +#include #include #include #include @@ -33,9 +34,12 @@ namespace EdiDecoder { struct frame_timestamp_t { uint32_t seconds = 0; uint32_t utco = 0; + uint32_t tsta = 0; // According to EN 300 797 Annex B + bool valid() const; std::string to_string() const; time_t to_unix_epoch() const; + std::chrono::system_clock::time_point to_system_clock() const; }; struct decode_state_t { diff --git a/src/input/Edi.cpp b/src/input/Edi.cpp index 8aee296..765a355 100644 --- a/src/input/Edi.cpp +++ b/src/input/Edi.cpp @@ -27,6 +27,7 @@ #include "input/Edi.h" #include +#include #include #include #include @@ -41,7 +42,7 @@ namespace Inputs { constexpr bool VERBOSE = false; constexpr size_t TCP_BLOCKSIZE = 2048; -constexpr size_t MAX_FRAMES_QUEUED = 10; +constexpr size_t MAX_FRAMES_QUEUED = 1000; Edi::Edi() : m_tcp_receive_server(TCP_BLOCKSIZE), @@ -96,23 +97,53 @@ int Edi::open(const std::string& name) int Edi::readFrame(uint8_t* buffer, size_t size) { - vector frame; + if (m_pending_sti_frame.frame.empty()) { + m_frames.try_pop(m_pending_sti_frame); + } + + if (not m_pending_sti_frame.frame.empty()) { + if (m_pending_sti_frame.frame.size() != size) { + etiLog.level(debug) << "EDI input " << m_name << " size mismatch: " << + m_pending_sti_frame.frame.size() << " received, " << size << " requested"; + memset(buffer, 0, size * sizeof(*buffer)); + } + else { + const auto now = chrono::system_clock::now(); + + if (m_pending_sti_frame.timestamp.to_system_clock() <= now) { + etiLog.level(debug) << "EDI input take frame with TS " << + m_pending_sti_frame.timestamp.to_string() << " queue size " << m_frames.size(); + + std::copy(m_pending_sti_frame.frame.cbegin(), m_pending_sti_frame.frame.cend(), buffer); + m_pending_sti_frame.frame.clear(); + } + else { + etiLog.level(debug) << "EDI input skip frame with TS " << + m_pending_sti_frame.timestamp.to_string() << " queue size " << m_frames.size(); + } + } + } + + return size; + +#if 0 + EdiDecoder::sti_frame_t sti; if (m_is_prebuffering) { m_is_prebuffering = m_frames.size() < 10; if (not m_is_prebuffering) { etiLog.level(info) << "EDI input " << m_name << " pre-buffering complete."; } } - else if (m_frames.try_pop(frame)) { - if (frame.size() == 0) { + else if (m_frames.try_pop(sti)) { + if (sti.frame.size() == 0) { etiLog.level(debug) << "EDI input " << m_name << " empty frame"; return 0; } - else if (frame.size() == size) { - std::copy(frame.cbegin(), frame.cend(), buffer); + else if (sti.frame.size() == size) { + std::copy(sti.frame.cbegin(), sti.frame.cend(), buffer); } else { - etiLog.level(debug) << "EDI input " << m_name << " size mismatch: " << frame.size() << + etiLog.level(debug) << "EDI input " << m_name << " size mismatch: " << sti.frame.size() << " received, " << size << " requested"; memset(buffer, 0, size * sizeof(*buffer)); } @@ -123,6 +154,7 @@ int Edi::readFrame(uint8_t* buffer, size_t size) etiLog.level(info) << "EDI input " << m_name << " re-enabling pre-buffering"; } return size; +#endif } void Edi::m_run() @@ -159,7 +191,7 @@ void Edi::m_run() const auto sti = m_sti_writer.getFrame(); if (not sti.frame.empty()) { - m_frames.push_wait_if_full(move(sti.frame), MAX_FRAMES_QUEUED); + m_frames.push_wait_if_full(move(sti), MAX_FRAMES_QUEUED); work_done = true; } diff --git a/src/input/Edi.h b/src/input/Edi.h index 7b3dc04..66ff682 100644 --- a/src/input/Edi.h +++ b/src/input/Edi.h @@ -71,7 +71,8 @@ class Edi : public InputBase { EdiDecoder::STIDecoder m_sti_decoder; std::thread m_thread; std::atomic m_running = ATOMIC_VAR_INIT(false); - ThreadsafeQueue > m_frames; + ThreadsafeQueue m_frames; + EdiDecoder::sti_frame_t m_pending_sti_frame; bool m_is_prebuffering = true; diff --git a/src/utils.cpp b/src/utils.cpp index 721c145..3e3e86e 100644 --- a/src/utils.cpp +++ b/src/utils.cpp @@ -328,7 +328,7 @@ void printSubchannels(const vec_sp_subchannel& subchannels) etiLog.level(info) << " URI: " << subchannel->inputUri; switch (subchannel->type) { case subchannel_type_t::DABAudio: - etiLog.log(info, " type: DAbAudio"); + etiLog.log(info, " type: DABAudio"); break; case subchannel_type_t::DABPlusAudio: etiLog.log(info, " type: DABPlusAudio"); -- cgit v1.2.3