Merge pull request #37 from anarkiwi/zstwrite

write FFT points.

.github/workflows/config/test-shell.nix

@@ -1,8 +1,9 @@
+# TODO: use released nix packages when boost/zstd available.
 with (import (builtins.fetchGit {
-  name = "nixpkgs-01-27-2023";
+  name = "nixpkgs-03-08-2023";
   url = "https://github.com/nixos/nixpkgs/";
-  ref = "refs/tags/22.11";
-  rev = "bd15cafc53d0aecd90398dd3ffc83a908bceb734";
+  ref = "refs/heads/nixpkgs-unstable";
+  rev = "1e383aada51b416c6c27d4884d2e258df201bc11";
 }) {} );
 let
   test-py-packages = python-packages: with python-packages; [
@@ -28,9 +29,11 @@ mkShell {
     gnuradio.python.pkgs.pybind11
     gnuradio.python.pkgs.pandas
     cmake
-
+
+    zstd
+    boost
     libsndfile
     soapysdr
     cppcheck
   ];
-}
+}

grc/iqtlabs_retune_fft.block.yml

@@ -29,6 +29,8 @@ documentation: |-
     fft_roll: if True, the consider the center of the FFT window to be at the beginning of the FFT vector.
     fft_min: clip FFT values to this minimum.
     fft_max: clip FFT values to this maximum.
+    sdir: directory to write raw FFT points to.
+    write_step_fft: if > 0, write N vectors of FFT points at each retune step.
 
   example JSON output:
     { "ts": <epoch time>, "config": { "freq_start": 10e6, "freq_end": 20e6, ... }, "buckets": { "10e6": 1, "10.1e6", ... } }

lib/CMakeLists.txt

@@ -19,7 +19,7 @@ if(NOT iqtlabs_sources)
 endif(NOT iqtlabs_sources)
 
 add_library(gnuradio-iqtlabs SHARED ${iqtlabs_sources})
-target_link_libraries(gnuradio-iqtlabs gnuradio::gnuradio-runtime ${Boost_LIBRARIES})
+target_link_libraries(gnuradio-iqtlabs ${Boost_LIBRARIES} gnuradio::gnuradio-runtime )
 target_include_directories(gnuradio-iqtlabs
     PUBLIC $<BUILD_INTERFACE:${CMAKE_CURRENT_SOURCE_DIR}/../include>
     PUBLIC $<INSTALL_INTERFACE:include>

lib/retune_fft_impl.cc

@@ -203,11 +203,10 @@
  */
 
 #include <chrono>
+#include <fstream>
 #include <ios>
 #include <iostream>
 #include <sstream>
-#include <boost/filesystem.hpp>
-#include <boost/iostreams/filter/zstd.hpp>
 
 #include <gnuradio/io_signature.h>
 #include "retune_fft_impl.h"
@@ -252,13 +251,48 @@ namespace gr {
               pending_retune_(0),
               total_tune_count_(0),
               sdir_(sdir),
-              write_step_fft_(write_step_fft)
+              write_step_fft_(write_step_fft),
+              write_step_fft_count_(write_step_fft)
         {
+            outbuf_p.reset(new boost::iostreams::filtering_ostream());
             message_port_register_out(TUNE);
         }
 
         retune_fft_impl::~retune_fft_impl()
         {
+            close_();
+        }
+
+        std::string retune_fft_impl::get_prefix_file_(const std::string &file, const std::string &prefix) {
+            boost::filesystem::path orig_path(file);
+            std::string basename(orig_path.filename().c_str());
+            std::string dirname(boost::filesystem::canonical(orig_path.parent_path()).c_str());
+            return dirname + "/" + prefix + basename;
+        }
+
+        std::string retune_fft_impl::get_dotfile_(const std::string &file) {
+            return get_prefix_file_(file, ".");
+        }
+
+        void retune_fft_impl::write_(const char *data, size_t len) {
+            if (!outbuf_p->empty()) {
+                outbuf_p->write(data, len);
+            }
+        }
+
+        void retune_fft_impl::open_(const std::string &file, size_t zlevel) {
+            close_();
+            file_ = file;
+            dotfile_ = get_dotfile_(file_);
+            outbuf_p->push(boost::iostreams::zstd_compressor(boost::iostreams::zstd_params(zlevel)));
+            outbuf_p->push(boost::iostreams::file_sink(file_));
+        }
+
+        void retune_fft_impl::close_() {
+             if (!outbuf_p->empty()) {
+                 outbuf_p->reset();
+                 rename(dotfile_.c_str(), file_.c_str());
+             }
         }
 
         uint64_t retune_fft_impl::host_now_()
@@ -302,13 +336,17 @@ namespace gr {
                         --skip_fft_count_;
                         continue;
                     }
+                    if (write_step_fft_count_) {
+                        write_((const char*)in, sizeof(input_type) * nfft_);
+                        --write_step_fft_count_;
+                    }
                     for (size_t k = 0; k < nfft_; ++k) {
-                        double s = *in++;
-                        sample_[k] += s;
+                        sample_[k] += *in++;
                     }
                     if (++fft_count_ >= tune_step_fft_ && (pending_retune_ == 0 || total_tune_count_ == 0)) {
                         fft_count_ = 0;
                         skip_fft_count_ = skip_tune_step_fft_;
+                        write_step_fft_count_ = write_step_fft_;
                         retune_now_();
                     }
                     ++sample_count_;
@@ -383,6 +421,8 @@ namespace gr {
                     --pending_retune_;
                     if (last_rx_freq_ && sample_count_) {
                         const uint64_t host_now = host_now_();
+                        std::string bucket_path = sdir_ + "/fft_" + std::to_string(host_now) + "_" + std::to_string(uint64_t(rx_freq)) + "Hz.zst";
+                        open_(bucket_path, 1);
                         const double bucket_size = samp_rate_ / vlen_;
                         std::stringstream ss("", std::ios_base::app | std::ios_base::out);
                         ss << "{" <<

lib/retune_fft_impl.h

@@ -205,8 +205,14 @@
 #ifndef INCLUDED_IQTLABS_RETUNE_FFT_IMPL_H
 #define INCLUDED_IQTLABS_RETUNE_FFT_IMPL_H
 
+#include <boost/filesystem.hpp>
+#include <boost/iostreams/device/file.hpp>
+#include <boost/iostreams/filtering_stream.hpp>
+#include <boost/iostreams/filter/zstd.hpp>
+#include <boost/scoped_ptr.hpp>
 #include <gnuradio/iqtlabs/retune_fft.h>
 
+
 namespace gr {
   namespace iqtlabs {
     using input_type = float;
@@ -219,6 +225,11 @@ namespace gr {
       void retune_now_();
       void sum_samples_(size_t c, const input_type* &in);
       void output_buckets_(const std::string &name, const std::list<std::pair<double, double>> &buckets, std::stringstream &ss);
+      std::string get_prefix_file_(const std::string &file, const std::string &prefix);
+      std::string get_dotfile_(const std::string &file);
+      void write_(const char *data, size_t len);
+      void open_(const std::string &file, size_t zlevel);
+      void close_();
 
       pmt::pmt_t tag_;
       size_t vlen_;
@@ -229,6 +240,9 @@ namespace gr {
       uint64_t tune_step_hz_;
       uint64_t tune_step_fft_;
       uint64_t skip_tune_step_fft_;
+      uint64_t write_step_fft_;
+      std::string sdir_;
+
       bool fft_roll_;
       double fft_min_;
       double fft_max_;
@@ -244,8 +258,11 @@ namespace gr {
       uint64_t skip_fft_count_;
       uint64_t total_tune_count_;
       uint64_t pending_retune_;
-      uint64_t write_step_fft_;
-      std::string sdir_;
+      uint64_t write_step_fft_count_;
+
+      boost::scoped_ptr<boost::iostreams::filtering_ostream> outbuf_p;
+      std::string file_;
+      std::string dotfile_;
 
      public:
       retune_fft_impl(const std::string &tag, int vlen, int nfft, uint64_t samp_rate, uint64_t freq_start, uint64_t freq_end, int tune_step_hz, int tune_step_fft, int skip_tune_step_fft, bool fft_roll, double fft_min, double fft_max, const std::string &sdir, uint64_t write_step_fft);

python/iqtlabs/qa_retune_fft.py

@@ -202,8 +202,10 @@
 #    See the License for the specific language governing permissions and
 #    limitations under the License.
 #
+import glob
 import json
 import os
+import subprocess
 import time
 import tempfile
 import pandas as pd
@@ -236,11 +238,12 @@ def test_retune_fft(self):
         samp_rate = points * points
         freq_start = int(1e9 / samp_rate) * samp_rate
         freq_end = int(1.1e9 / samp_rate) * samp_rate
+        fft_write_count = 2
 
         with tempfile.TemporaryDirectory() as tmpdir:
             test_file = os.path.join(tmpdir, "samples.csv")
             iqtlabs_tuneable_test_source_0 = tuneable_test_source(freq_end)
-            iqtlabs_retune_fft_0 = retune_fft("rx_freq", points, points, int(samp_rate), int(freq_start), int(freq_end), int(samp_rate), 64, 2, False, 1e-4, 1e4, "", 0)
+            iqtlabs_retune_fft_0 = retune_fft("rx_freq", points, points, int(samp_rate), int(freq_start), int(freq_end), int(samp_rate), 64, 2, False, 1e-4, 1e4, tmpdir, fft_write_count)
             fft_vxx_0 = fft.fft_vcc(points, True, window.blackmanharris(points), True, 1)
             blocks_throttle_0 = blocks.throttle(gr.sizeof_gr_complex*1, samp_rate, True)
             blocks_stream_to_vector_0 = blocks.stream_to_vector(gr.sizeof_gr_complex*1, points)
@@ -289,6 +292,14 @@ def test_retune_fft(self):
             self.assertGreater(f_count_min, 1)
             self.assertTrue(non_unique_v.empty, (non_unique_v, df))
 
+            zst_fft_files = sorted(glob.glob(os.path.join(tmpdir, '*.zst')))[:10]
+            self.assertTrue(zst_fft_files)
+            output = subprocess.check_output(['zstd', '-tv'] + zst_fft_files).decode("utf8")
+            bytes_match = '%u bytes' % fft_write_count * 4
+            for file in output.splitlines():
+                # points output correct size (floats)
+                self.assertIn(bytes_match, file, file)
+
 
 if __name__ == '__main__':
     gr_unittest.run(qa_retune_fft)