Bufferef IO & propagate configuration inside ION source

src/algorithms/signal_source/adapters/ion_gsms_signal_source.cc

@@ -66,7 +66,7 @@ IONGSMSSignalSource::IONGSMSSignalSource(const ConfigurationInterface* configura
             LOG(ERROR) << "A signal source does not have an input stream";
         }
 
-    sources_ = metadata_.make_stream_sources(stream_ids_);
+    sources_ = metadata_.make_stream_sources(configuration, role, stream_ids_);
 
     for (const auto& source : sources_)
         {

src/algorithms/signal_source/libs/ion_gsms.cc

@@ -25,8 +25,11 @@
 #include <utility>
 #endif
 
+using namespace std::string_literals;
 
 IONGSMSFileSource::IONGSMSFileSource(
+    const ConfigurationInterface* configuration,
+    const std::string& role,
     const std::filesystem::path& metadata_filepath,
     const GnssMetadata::File& file,
     const GnssMetadata::Block& block,
@@ -36,7 +39,9 @@ IONGSMSFileSource::IONGSMSFileSource(
           gr::io_signature::make(0, 0, 0),
           make_output_signature(block, stream_ids)),
       file_metadata_(file),
-      block_metadata_(block)
+      block_metadata_(block),
+      io_buffer_offset_(0),
+      chunk_cycle_length_(0)
 {
     std::filesystem::path data_filepath = metadata_filepath.parent_path() / file.Url().Value();
     fd_ = std::fopen(data_filepath.c_str(), "rb");
@@ -47,6 +52,7 @@ IONGSMSFileSource::IONGSMSFileSource(
     for (const auto& chunk : block.Chunks())
         {
             chunk_data_.emplace_back(std::make_shared<IONGSMSChunkData>(chunk, stream_ids, output_stream_offset));
+            chunk_cycle_length_ += chunk.CountWords() * chunk.SizeWord();
             const std::size_t out_count = chunk_data_.back()->output_stream_count();
             output_stream_offset += out_count;
             for (std::size_t i = 0; i < out_count; ++i)
@@ -55,6 +61,8 @@ IONGSMSFileSource::IONGSMSFileSource(
                 }
         }
     output_stream_count_ = output_stream_offset;
+
+    io_buffer_.resize(1024 * chunk_cycle_length_);
 }
 
 IONGSMSFileSource::~IONGSMSFileSource()
@@ -67,16 +75,39 @@ int IONGSMSFileSource::work(
     gr_vector_const_void_star& input_items,
     gr_vector_void_star& output_items)
 {
-    // for (int i = 0; i < noutput_items; ++i)
+    io_buffer_offset_ = 0;
-        // {
+    std::fread(io_buffer_.data(), sizeof(decltype(io_buffer_)::value_type), io_buffer_.size(), fd_);
+    std::vector<int> items_produced{};
+
+    items_produced.resize(output_items.size());
+    for (int i = 0; i < items_produced.size(); ++i)
+        {
+            items_produced[i] = 0;
+        }
+
+
+    while (io_buffer_offset_ < io_buffer_.size())
+        {
             for (auto& c : chunk_data_)
                 {
-                    c->read_from_file(fd_);
+                    io_buffer_offset_ += c->read_from_buffer(io_buffer_.data(), io_buffer_offset_);
                     c->write_to_output(output_items, [&](int output, int nitems) {
-                        produce(output, nitems);
+                        items_produced[output] += nitems;
+
+                        // if (nitems_written(output) % 100 == 0)
+                            // {
+                                // add_item_tag(output, nitems_written(output), pmt::mp("tag_test"), pmt::from_uint64(nitems_written(output)));
+                            // }
                     });
                 }
-        // }
+        }
+
+    std::cout << "produced: " << std::to_string(items_produced[0]) << std::endl;
+    for (int i = 0; i < items_produced.size(); ++i)
+        {
+            produce(i, items_produced[i]);
+        }
+
     return WORK_CALLED_PRODUCE;
 }
 

src/algorithms/signal_source/libs/ion_gsms.h

@@ -19,10 +19,12 @@
 
 #include "gnss_block_interface.h"
 #include "ion_gsms_chunk_data.h"
+#include "configuration_interface.h"
 #include <gnuradio/block.h>
 #include <gnuradio/sync_block.h>
 #include <filesystem>
 #include <string>
+#include <iostream>
 
 
 class IONGSMSFileSource : public gr::sync_block
@@ -31,6 +33,8 @@ public:
     using sptr = gnss_shared_ptr<IONGSMSFileSource>;
 
     IONGSMSFileSource(
+        const ConfigurationInterface* configuration,
+        const std::string& role,
         const std::filesystem::path& metadata_filepath,
         const GnssMetadata::File& file,
         const GnssMetadata::Block& block,
@@ -53,9 +57,12 @@ private:
     const GnssMetadata::File& file_metadata_;
     const GnssMetadata::Block& block_metadata_;
     FILE* fd_;
+    std::vector<uint8_t> io_buffer_;
+    std::size_t io_buffer_offset_;
     std::size_t output_stream_count_;
     std::vector<std::size_t> output_stream_item_sizes_;
     std::vector<std::shared_ptr<IONGSMSChunkData>> chunk_data_;
+    std::size_t chunk_cycle_length_;
 };
 
 #include "ion_gsms_metadata_handler.h"

src/algorithms/signal_source/libs/ion_gsms_chunk_data.cc

@@ -76,9 +76,11 @@ IONGSMSChunkData::~IONGSMSChunkData()
     });
 }
 
-void IONGSMSChunkData::read_from_file(FILE* fd)
+std::size_t IONGSMSChunkData::read_from_buffer(uint8_t* buffer, std::size_t offset)
 {
-    std::fread(buffer_, sizeword_, countwords_, fd);
+    memset(buffer_, 0, sizeword_ * countwords_);
+    memcpy(buffer_, &buffer[offset], sizeword_ * countwords_);
+    return sizeword_ * countwords_;
 }
 
 void IONGSMSChunkData::write_to_output(gr_vector_void_star& outputs, const std::function<void(int output, int nitems)>& produce)
@@ -89,6 +91,7 @@ void IONGSMSChunkData::write_to_output(gr_vector_void_star& outputs, const std::
     });
 }
 
+
 std::size_t IONGSMSChunkData::output_stream_count() const
 {
     return output_stream_count_;
@@ -203,6 +206,8 @@ void IONGSMSChunkData::write_n_samples(
                     ++sample;
                 }
         }
+
+    out += sample_count * sizeof(OT);
 }
 
 

src/algorithms/signal_source/libs/ion_gsms_chunk_data.h

@@ -87,7 +87,7 @@ public:
     IONGSMSChunkData(IONGSMSChunkData&& rhl) = delete;
     IONGSMSChunkData& operator=(IONGSMSChunkData&& rhl) = delete;
 
-    void read_from_file(FILE* fd);
+    std::size_t read_from_buffer(uint8_t* buffer, std::size_t offset);
 
     void write_to_output(gr_vector_void_star& outputs, const std::function<void(int output, int nitems)>& produce);
 

src/algorithms/signal_source/libs/ion_gsms_metadata_handler.cc

@@ -53,7 +53,7 @@ void IONGSMSMetadataHandler::load_metadata()
         }
 }
 
-std::vector<IONGSMSFileSource::sptr> IONGSMSMetadataHandler::make_stream_sources(const std::vector<std::string>& stream_ids) const
+std::vector<IONGSMSFileSource::sptr> IONGSMSMetadataHandler::make_stream_sources(const ConfigurationInterface* configuration, const std::string& role, const std::vector<std::string>& stream_ids) const
 {
     std::vector<IONGSMSFileSource::sptr> sources{};
     for (const auto& file : metadata_.Files())
@@ -75,6 +75,8 @@ std::vector<IONGSMSFileSource::sptr> IONGSMSMetadataHandler::make_stream_sources
                                                                 }))
                                                                 {
                                                                     auto source = gnss_make_shared<IONGSMSFileSource>(
+                                                                        configuration,
+                                                                        role,
                                                                         metadata_filepath_,
                                                                         file,
                                                                         block,

src/algorithms/signal_source/libs/ion_gsms_metadata_handler.h

@@ -18,6 +18,7 @@
 #define ION_GSMS_METADATA_HANDLER_H
 
 #include "GnssMetadata.h"
+#include "configuration_interface.h"
 #include <gnuradio/block.h>
 
 #if USE_GLOG_AND_GFLAGS
@@ -31,7 +32,7 @@ class IONGSMSMetadataHandler
 public:
     explicit IONGSMSMetadataHandler(const std::string& metadata_filepath);
 
-    std::vector<IONGSMSFileSource::sptr> make_stream_sources(const std::vector<std::string>& stream_ids) const;
+    std::vector<IONGSMSFileSource::sptr> make_stream_sources(const ConfigurationInterface* configuration, const std::string& role, const std::vector<std::string>& stream_ids) const;
 
 public:  // Getters
     const std::string& metadata_filepath() const;