CsvDumper

baranovmv · Apr 14, 2024 · 96fbd6a · 96fbd6a
1 parent 5dd9dd1
commit 96fbd6a
Show file tree

Hide file tree

Showing 25 changed files with 150 additions and 77 deletions.
diff --git a/src/internal_modules/roc_audio/feedback_monitor.cpp b/src/internal_modules/roc_audio/feedback_monitor.cpp
@@ -21,7 +21,7 @@ FeedbackMonitor::FeedbackMonitor(IFrameWriter& writer,
                                  const FeedbackConfig& feedback_config,
                                  const LatencyConfig& latency_config,
                                  const SampleSpec& sample_spec)
-    : tuner_(latency_config, sample_spec)
+    : tuner_(latency_config, sample_spec, NULL)
     , use_packetizer_(false)
     , has_feedback_(false)
     , last_feedback_ts_(0)

diff --git a/src/internal_modules/roc_audio/freq_estimator.cpp b/src/internal_modules/roc_audio/freq_estimator.cpp
@@ -10,7 +10,6 @@
 #include "roc_core/log.h"
 #include "roc_core/panic.h"
 #include "roc_core/time.h"
-#include <fstream>
 
 namespace roc {
 namespace audio {
@@ -66,7 +65,8 @@ double dot_prod(const double* coeff,
 } // namespace
 
 FreqEstimator::FreqEstimator(FreqEstimatorProfile profile,
-                             packet::stream_timestamp_t target_latency)
+                             packet::stream_timestamp_t target_latency,
+                             core::CsvDumper* dumper)
     : config_(make_config(profile))
     , target_(target_latency)
     , dec1_ind_(0)
@@ -75,7 +75,8 @@ FreqEstimator::FreqEstimator(FreqEstimatorProfile profile,
     , accum_(0)
     , coeff_(1)
     , stable_(false)
-    , last_unstable_time_(core::timestamp(core::ClockMonotonic)) {
+    , last_unstable_time_(core::timestamp(core::ClockMonotonic))
+    , dumper_(dumper) {
     roc_log(LogDebug, "freq estimator: initializing: P=%e I=%e dc1=%lu dc2=%lu",
             config_.P, config_.I, (unsigned long)config_.decimation_factor1,
             (unsigned long)config_.decimation_factor2);
@@ -108,18 +109,20 @@ float FreqEstimator::freq_coeff() const {
 }
 
 void FreqEstimator::update(packet::stream_timestamp_t current) {
-    static std::ofstream fout("/tmp/fe.log", std::ios::out);
-
     double filtered;
 
     if (run_decimators_(current, filtered)) {
-        fout << core::timestamp(core::ClockUnix)
-            << ", " << filtered
-            << ", " << target_
-            << ", " << (filtered - target_) * config_.P
-            << ", " << accum_ * config_.I
-            << std::endl;
-        fout.flush();
+        if (dumper_) {
+            core::CsvEntry e;
+            e.type = 'f';
+            e.n_fields = 5;
+            e.fields[0] = core::timestamp(core::ClockUnix);
+            e.fields[1] = filtered;
+            e.fields[2] = target_;
+            e.fields[3] = (filtered - target_) * config_.P;
+            e.fields[4] = accum_ * config_.I;
+            dumper_->write(e);
+        }
         coeff_ = run_controller_(filtered);
     }
 }

diff --git a/src/internal_modules/roc_audio/freq_estimator.h b/src/internal_modules/roc_audio/freq_estimator.h
@@ -14,6 +14,7 @@
 
 #include "roc_audio/freq_estimator_decim.h"
 #include "roc_audio/sample.h"
+#include "roc_core/csv_dumper.h"
 #include "roc_core/noncopyable.h"
 #include "roc_packet/units.h"
 
@@ -72,7 +73,8 @@ class FreqEstimator : public core::NonCopyable<> {
     //!  - @p profile defines configuration preset.
     //!  - @p target_latency defines latency we want to archive.
     FreqEstimator(FreqEstimatorProfile profile,
-                  packet::stream_timestamp_t target_latency);
+                  packet::stream_timestamp_t target_latency,
+                  roc::core::CsvDumper* dumper);
 
     //! Get current frequecy coefficient.
     float freq_coeff() const;
@@ -107,6 +109,8 @@ class FreqEstimator : public core::NonCopyable<> {
     bool stable_; // True if FreqEstimator has stabilized.
     // Last time when FreqEstimator was out of range.
     core::nanoseconds_t last_unstable_time_;
+
+    core::CsvDumper* dumper_;
 };
 
 } // namespace audio

diff --git a/src/internal_modules/roc_audio/latency_monitor.cpp b/src/internal_modules/roc_audio/latency_monitor.cpp
@@ -25,8 +25,9 @@ LatencyMonitor::LatencyMonitor(IFrameReader& frame_reader,
                                ResamplerReader* resampler,
                                const LatencyConfig& config,
                                const SampleSpec& packet_sample_spec,
-                               const SampleSpec& frame_sample_spec)
-    : tuner_(config, frame_sample_spec)
+                               const SampleSpec& frame_sample_spec,
+                               core::CsvDumper* dumper)
+    : tuner_(config, frame_sample_spec, dumper)
     , frame_reader_(frame_reader)
     , incoming_queue_(incoming_queue)
     , depacketizer_(depacketizer)

diff --git a/src/internal_modules/roc_audio/latency_monitor.h b/src/internal_modules/roc_audio/latency_monitor.h
@@ -19,6 +19,7 @@
 #include "roc_audio/resampler_reader.h"
 #include "roc_audio/sample_spec.h"
 #include "roc_core/attributes.h"
+#include "roc_core/csv_dumper.h"
 #include "roc_core/noncopyable.h"
 #include "roc_core/optional.h"
 #include "roc_core/time.h"
@@ -157,7 +158,8 @@ class LatencyMonitor : public IFrameReader, public core::NonCopyable<> {
                    ResamplerReader* resampler,
                    const LatencyConfig& config,
                    const SampleSpec& packet_sample_spec,
-                   const SampleSpec& frame_sample_spec);
+                   const SampleSpec& frame_sample_spec,
+                   core::CsvDumper* dumper);
 
     //! Check if the object was initialized successfully.
     bool is_valid() const;

diff --git a/src/internal_modules/roc_audio/latency_tuner.cpp b/src/internal_modules/roc_audio/latency_tuner.cpp
@@ -10,7 +10,6 @@
 #include "roc_core/log.h"
 #include "roc_core/panic.h"
 #include "roc_core/time.h"
-#include <fstream>
 
 namespace roc {
 namespace audio {
@@ -146,7 +145,9 @@ void LatencyConfig::deduce_defaults(core::nanoseconds_t default_target_latency,
     }
 }
 
-LatencyTuner::LatencyTuner(const LatencyConfig& config, const SampleSpec& sample_spec)
+LatencyTuner::LatencyTuner(const LatencyConfig& config,
+                           const SampleSpec& sample_spec,
+                           core::CsvDumper* dumper)
     : stream_pos_(0)
     , scale_interval_(0)
     , scale_pos_(0)
@@ -176,7 +177,8 @@ LatencyTuner::LatencyTuner(const LatencyConfig& config, const SampleSpec& sample
     , lat_update_upper_thrsh_(config.upper_threshold_coef)
     , lat_update_dec_step_(upper_coef_to_step_lat_update_(config.upper_threshold_coef))
     , lat_update_inc_step_(lower_thrs_to_step_lat_update_(config.upper_threshold_coef))
-    , last_lat_limit_log_(0) {
+    , last_lat_limit_log_(0)
+    , dumper_(dumper) {
     roc_log(LogDebug,
             "latency tuner: initializing:"
             " target_latency=%ld(%.3fms) min_latency=%ld(%.3fms) max_latency=%ld(%.3fms)"
@@ -289,11 +291,11 @@ LatencyTuner::LatencyTuner(const LatencyConfig& config, const SampleSpec& sample
                     " upper_threshold_coef=%f", (double)config.upper_threshold_coef);
             }
 
-            fe_.reset(new (fe_)
-                          FreqEstimator(profile_ == LatencyTunerProfile_Responsive
-                                            ? FreqEstimatorProfile_Responsive
-                                            : FreqEstimatorProfile_Gradual,
-                                        (packet::stream_timestamp_t)target_latency_));
+            fe_.reset(new (fe_) FreqEstimator(profile_ == LatencyTunerProfile_Responsive
+                                                  ? FreqEstimatorProfile_Responsive
+                                                  : FreqEstimatorProfile_Gradual,
+                                              (packet::stream_timestamp_t)target_latency_,
+                                              dumper_));
             if (!fe_) {
                 return;
             }
@@ -309,8 +311,6 @@ bool LatencyTuner::is_valid() const {
 
 void LatencyTuner::write_metrics(const LatencyMetrics& latency_metrics,
                                  const packet::LinkMetrics& link_metrics) {
-    static std::ofstream fout("/tmp/tuner.log", std::ios::out);
-
     roc_panic_if(!is_valid());
 
     if (latency_metrics.niq_latency > 0 || latency_metrics.niq_stalling > 0
@@ -333,10 +333,15 @@ void LatencyTuner::write_metrics(const LatencyMetrics& latency_metrics,
                                latency_metrics.fec_block_duration);
     }
 
-    fout << core::timestamp(core::ClockUnix)
-        << ", " << niq_latency_
-        << ", " << target_latency_
-        << std::endl;
+    if (dumper_) {
+        core::CsvEntry e;
+        e.type = 't';
+        e.n_fields = 3;
+        e.fields[0] = core::timestamp(core::ClockUnix);
+        e.fields[1] = niq_latency_;
+        e.fields[2] = target_latency_;
+        dumper_->write(e);
+    }
 
     latency_metrics_ = latency_metrics;
     link_metrics_ = link_metrics;

diff --git a/src/internal_modules/roc_audio/latency_tuner.h b/src/internal_modules/roc_audio/latency_tuner.h
@@ -14,6 +14,7 @@
 
 #include "roc_audio/freq_estimator.h"
 #include "roc_audio/sample_spec.h"
+#include "roc_core/csv_dumper.h"
 #include "roc_core/noncopyable.h"
 #include "roc_core/optional.h"
 #include "roc_core/time.h"
@@ -183,7 +184,9 @@ struct LatencyMetrics {
 class LatencyTuner : public core::NonCopyable<> {
 public:
     //! Initialize.
-    LatencyTuner(const LatencyConfig& config, const SampleSpec& sample_spec);
+    LatencyTuner(const LatencyConfig& config,
+                 const SampleSpec& sample_spec,
+                 core::CsvDumper* dumper);
 
     //! Check if the object was initialized successfully.
     bool is_valid() const;
@@ -276,6 +279,8 @@ class LatencyTuner : public core::NonCopyable<> {
     const float lat_update_inc_step_;
 
     core::nanoseconds_t last_lat_limit_log_;
+
+    core::CsvDumper* dumper_;
 };
 
 //! Get string name of latency backend.

diff --git a/src/internal_modules/roc_pipeline/config.cpp b/src/internal_modules/roc_pipeline/config.cpp
@@ -55,7 +55,9 @@ void ReceiverSessionConfig::deduce_defaults() {
     resampler.deduce_defaults(latency.tuner_backend, latency.tuner_profile);
 }
 
-ReceiverSourceConfig::ReceiverSourceConfig() {
+ReceiverSourceConfig::ReceiverSourceConfig()
+    : dump_file(NULL)
+{
 }
 
 void ReceiverSourceConfig::deduce_defaults() {

diff --git a/src/internal_modules/roc_pipeline/config.h b/src/internal_modules/roc_pipeline/config.h
@@ -191,6 +191,9 @@ struct ReceiverSourceConfig {
     //! Default parameters for a session.
     ReceiverSessionConfig session_defaults;
 
+    //! File to a dump file in csv format with some run-time metrics.
+    const char* dump_file;
+
     //! Initialize config.
     ReceiverSourceConfig();
 

diff --git a/src/internal_modules/roc_pipeline/receiver_session.cpp b/src/internal_modules/roc_pipeline/receiver_session.cpp
@@ -22,9 +22,11 @@ ReceiverSession::ReceiverSession(
     packet::PacketFactory& packet_factory,
     core::BufferFactory<uint8_t>& byte_buffer_factory,
     core::BufferFactory<audio::sample_t>& sample_buffer_factory,
+    core::CsvDumper* dumper,
     core::IArena& arena)
     : core::RefCounted<ReceiverSession, core::ArenaAllocation>(arena)
     , frame_reader_(NULL)
+    , dumper_(dumper)
     , valid_(false) {
     const rtp::Encoding* pkt_encoding =
         encoding_map.find_by_pt(session_config.payload_type);
@@ -48,9 +50,8 @@ ReceiverSession::ReceiverSession(
     }
     pkt_writer = source_queue_.get();
 
-    source_meter_.reset(new (source_meter_)
-                            rtp::LinkMeter(arena, encoding_map, pkt_encoding->sample_spec,
-                                           session_config.latency));
+    source_meter_.reset(new (source_meter_) rtp::LinkMeter(
+        arena, encoding_map, pkt_encoding->sample_spec, session_config.latency, dumper_));
     if (!source_meter_) {
         return;
     }
@@ -93,9 +94,9 @@ ReceiverSession::ReceiverSession(
             return;
         }
 
-        repair_meter_.reset(new (repair_meter_)
-                                rtp::LinkMeter(arena, encoding_map, pkt_encoding->sample_spec,
-                                               session_config.latency));
+        repair_meter_.reset(new (repair_meter_) rtp::LinkMeter(
+            arena, encoding_map, pkt_encoding->sample_spec, session_config.latency,
+            dumper_));
         if (!repair_meter_) {
             return;
         }
@@ -216,7 +217,7 @@ ReceiverSession::ReceiverSession(
     latency_monitor_.reset(new (latency_monitor_) audio::LatencyMonitor(
         *frm_reader, *source_queue_, *depacketizer_, *source_meter_, fec_reader_.get(),
         resampler_reader_.get(), session_config.latency, pkt_encoding->sample_spec,
-        common_config.output_sample_spec));
+        common_config.output_sample_spec, dumper_));
     if (!latency_monitor_ || !latency_monitor_->is_valid()) {
         return;
     }

diff --git a/src/internal_modules/roc_pipeline/receiver_session.h b/src/internal_modules/roc_pipeline/receiver_session.h
@@ -22,6 +22,7 @@
 #include "roc_audio/resampler_reader.h"
 #include "roc_audio/watchdog.h"
 #include "roc_core/buffer_factory.h"
+#include "roc_core/csv_dumper.h"
 #include "roc_core/iarena.h"
 #include "roc_core/list_node.h"
 #include "roc_core/optional.h"
@@ -65,6 +66,7 @@ class ReceiverSession : public core::RefCounted<ReceiverSession, core::ArenaAllo
                     packet::PacketFactory& packet_factory,
                     core::BufferFactory<uint8_t>& byte_buffer_factory,
                     core::BufferFactory<audio::sample_t>& sample_buffer_factory,
+                    core::CsvDumper* dumper,
                     core::IArena& arena);
 
     //! Check if the session was succefully constructed.
@@ -149,6 +151,8 @@ class ReceiverSession : public core::RefCounted<ReceiverSession, core::ArenaAllo
 
     core::Optional<audio::LatencyMonitor> latency_monitor_;
 
+    core::CsvDumper* dumper_;
+
     bool valid_;
 };
 

diff --git a/src/internal_modules/roc_pipeline/receiver_session_group.cpp b/src/internal_modules/roc_pipeline/receiver_session_group.cpp
@@ -8,6 +8,7 @@
 
 #include "roc_pipeline/receiver_session_group.h"
 #include "roc_address/socket_addr_to_str.h"
+#include "roc_core/csv_dumper.h"
 #include "roc_core/log.h"
 #include "roc_core/panic.h"
 #include "roc_rtcp/participant_info.h"
@@ -25,6 +26,7 @@ ReceiverSessionGroup::ReceiverSessionGroup(
     packet::PacketFactory& packet_factory,
     core::BufferFactory<uint8_t>& byte_buffer_factory,
     core::BufferFactory<audio::sample_t>& sample_buffer_factory,
+    core::CsvDumper* dumper,
     core::IArena& arena)
     : source_config_(source_config)
     , slot_config_(slot_config)
@@ -36,6 +38,7 @@ ReceiverSessionGroup::ReceiverSessionGroup(
     , byte_buffer_factory_(byte_buffer_factory)
     , sample_buffer_factory_(sample_buffer_factory)
     , session_router_(arena)
+    , dumper_(dumper)
     , valid_(false) {
     identity_.reset(new (identity_) rtp::Identity());
     if (!identity_ || !identity_->is_valid()) {
@@ -381,7 +384,7 @@ ReceiverSessionGroup::create_session_(const packet::PacketPtr& packet) {
 
     core::SharedPtr<ReceiverSession> sess = new (arena_) ReceiverSession(
         sess_config, source_config_.common, encoding_map_, packet_factory_,
-        byte_buffer_factory_, sample_buffer_factory_, arena_);
+        byte_buffer_factory_, sample_buffer_factory_, dumper_, arena_);
 
     if (!sess || !sess->is_valid()) {
         roc_log(LogError, "session group: can't create session, initialization failed");

diff --git a/src/internal_modules/roc_pipeline/receiver_session_group.h b/src/internal_modules/roc_pipeline/receiver_session_group.h
@@ -13,6 +13,7 @@
 #define ROC_PIPELINE_RECEIVER_SESSION_GROUP_H_
 
 #include "roc_audio/mixer.h"
+#include "roc_core/csv_dumper.h"
 #include "roc_core/iarena.h"
 #include "roc_core/list.h"
 #include "roc_core/noncopyable.h"
@@ -54,6 +55,7 @@ class ReceiverSessionGroup : public core::NonCopyable<>, private rtcp::IParticip
                          packet::PacketFactory& packet_factory,
                          core::BufferFactory<uint8_t>& byte_buffer_factory,
                          core::BufferFactory<audio::sample_t>& sample_buffer_factory,
+                         core::CsvDumper* dumper,
                          core::IArena& arena);
 
     ~ReceiverSessionGroup();
@@ -153,6 +155,8 @@ class ReceiverSessionGroup : public core::NonCopyable<>, private rtcp::IParticip
     core::List<ReceiverSession> sessions_;
     ReceiverSessionRouter session_router_;
 
+    core::CsvDumper* dumper_;
+
     bool valid_;
 };