Merge pull request #289 from anarkiwi/pt

simplify process tags.
IQTLabs · Jun 24, 2024 · f97e202 · f97e202
2 parents 956fc37 + accd844
commit f97e202
Show file tree

Hide file tree

Showing 10 changed files with 94 additions and 162 deletions.
diff --git a/lib/base_impl.h b/lib/base_impl.h
@@ -235,23 +235,30 @@ typedef sigmf::SigMF<
                        pmt::from_double((double)rx_freq), _id);                \
   }
 
-#define PROCESS_TAGS(X)                                                        \
+#define FIND_TAGS                                                              \
+  std::vector<tag_t> all_tags, rx_freq_tags;                                   \
+  std::vector<TIME_T> rx_times;                                                \
+  get_tags_in_window(all_tags, 0, 0, in_count);                                \
+  get_tags(tag_, all_tags, rx_freq_tags, rx_times);
+
+#define PROCESS_TAGS(X, ...)                                                   \
+  FIND_TAGS                                                                    \
+  COUNT_T consumed = 0;                                                        \
   for (COUNT_T t = 0; t < rx_freq_tags.size(); ++t) {                          \
     const auto &tag = rx_freq_tags[t];                                         \
     auto rel = tag.offset - in_first;                                          \
     const TIME_T rx_time = rx_times[t];                                        \
     const FREQ_T rx_freq = GET_FREQ(tag);                                      \
     d_logger->debug("new rx_freq tag: {}", rx_freq);                           \
+    process_items_(rel, consumed, __VA_ARGS__);                                \
     {X} last_rx_freq_ = rx_freq;                                               \
     last_rx_time_ = rx_time;                                                   \
+    in_first += rel;                                                           \
+  }                                                                            \
+  if (consumed < in_count) {                                                   \
+    process_items_(in_count - consumed, consumed, __VA_ARGS__);                \
   }
 
-#define FIND_TAGS                                                              \
-  std::vector<tag_t> all_tags, rx_freq_tags;                                   \
-  std::vector<TIME_T> rx_times;                                                \
-  get_tags_in_window(all_tags, 0, 0, in_count);                                \
-  get_tags(tag_, all_tags, rx_freq_tags, rx_times);
-
 // A driver block might give us float style (e.g. 2.5e9) or unsigned lon.
 #define GET_FREQ(tag) (FREQ_T) pmt::to_double(tag.value)
 

diff --git a/lib/image_inference_impl.cc b/lib/image_inference_impl.cc
@@ -585,28 +585,13 @@ void image_inference_impl::run_inference_() {
 
 void image_inference_impl::process_tags_(COUNT_T in_count, COUNT_T in_first,
                                          const input_type *in) {
-  COUNT_T consumed = 0;
-
-  FIND_TAGS
-
-  if (rx_freq_tags.empty()) {
-    process_items_(in_count, consumed, in);
-  } else {
-    PROCESS_TAGS({
-      in_first += rel;
-
-      if (rel > 0) {
-        process_items_(rel, consumed, in);
-      }
-
-      if (rx_freq != last_rx_freq_) {
-        create_image_(true);
-      }
-    })
-    if (consumed < in_count) {
-      process_items_(in_count - consumed, consumed, in);
-    }
-  }
+  PROCESS_TAGS(
+      {
+        if (rx_freq != last_rx_freq_) {
+          create_image_(true);
+        }
+      },
+      in)
 }
 
 #pragma GCC diagnostic push

diff --git a/lib/iq_inference_impl.cc b/lib/iq_inference_impl.cc
@@ -367,9 +367,9 @@ void iq_inference_impl::run_inference_() {
   }
 }
 
-void iq_inference_impl::process_items_(COUNT_T power_in_count, COUNT_T in_first,
-                                       const float *&power_in,
-                                       COUNT_T &consumed) {
+void iq_inference_impl::process_items_(COUNT_T power_in_count,
+                                       COUNT_T &consumed, COUNT_T in_first,
+                                       const float *&power_in) {
   for (COUNT_T i = 0; i < power_in_count; i += n_vlen_, consumed += n_vlen_,
                power_in += batch_, samples_since_tag_ += batch_,
                sample_clock_ += batch_) {
@@ -436,36 +436,23 @@ void iq_inference_impl::process_tags_(COUNT_T in_first,
                                       COUNT_T in_count,
                                       const gr_complex *samples_in,
                                       const float *power_in) {
-  COUNT_T consumed = 0;
-
   for (COUNT_T i = 0; i < in_count; i += n_vlen_, samples_in += batch_) {
     COUNT_T j = (samples_in_first + i) % sample_buffer_;
     memcpy((void *)&samples_lookback_[j * batch_], samples_in,
            sizeof(gr_complex) * batch_);
   }
 
-  FIND_TAGS
-
-  if (rx_freq_tags.empty()) {
-    process_items_(in_count, in_first, power_in, consumed);
-  } else {
-    PROCESS_TAGS({
-      // TODO: in theory we might have a vector with more than one frequency's
-      // samples, as the SDR probably isn't vector aligned. In practice this
-      // should not happen in the most common Ettus low power workaround state,
-      // because tags are delayed until after re-tuning has been verified.
-      if (rel > 0) {
-        process_items_(rel, in_first, power_in, consumed);
-        in_first += rel;
-      }
-
-      last_rx_freq_sample_clock_ = sample_clock_;
-      samples_since_tag_ = 0;
-    })
-    if (consumed < in_count) {
-      process_items_(in_count - consumed, in_first, power_in, consumed);
-    }
-  }
+  PROCESS_TAGS(
+      {
+        // TODO: in theory we might have a vector with more than one frequency's
+        // samples, as the SDR probably isn't vector aligned. In practice this
+        // should not happen in the most common Ettus low power workaround
+        // state, because tags are delayed until after re-tuning has been
+        // verified.
+        last_rx_freq_sample_clock_ = sample_clock_;
+        samples_since_tag_ = 0;
+      },
+      in_first, power_in)
 }
 
 #pragma GCC diagnostic push

diff --git a/lib/iq_inference_impl.h b/lib/iq_inference_impl.h
@@ -248,8 +248,8 @@ class iq_inference_impl : public iq_inference, base_impl {
   boost::lockfree::spsc_queue<std::string> json_q_{MAX_INFERENCE};
   boost::scoped_ptr<std::thread> inference_thread_;
 
-  void process_items_(COUNT_T power_in_count, COUNT_T in_first,
-                      const float *&power_in, COUNT_T &consumed);
+  void process_items_(COUNT_T power_in_count, COUNT_T &consumed,
+                      COUNT_T in_first, const float *&power_in);
   void process_tags_(COUNT_T in_first, COUNT_T samples_in_first,
                      COUNT_T in_count, const gr_complex *samples_in,
                      const float *power_in);

diff --git a/lib/retune_fft_impl.cc b/lib/retune_fft_impl.cc
@@ -357,9 +357,10 @@ void retune_fft_impl::add_output_tags_(TIME_T rx_time, FREQ_T rx_freq,
   OUTPUT_TAGS(apply_rx_time_slew_(rx_time), rx_freq, 0, rel);
 }
 
-void retune_fft_impl::process_items_(COUNT_T c, const input_type *&in,
+void retune_fft_impl::process_items_(COUNT_T c, COUNT_T &consumed,
+                                     const input_type *&in,
                                      const input_type *&fft_output,
-                                     COUNT_T &consumed, COUNT_T &produced) {
+                                     COUNT_T &produced) {
   for (COUNT_T i = 0; i < c; ++i, in += nfft_) {
     ++consumed;
     if (skip_fft_count_) {
@@ -483,34 +484,19 @@ void retune_fft_impl::process_buckets_(FREQ_T rx_freq, TIME_T rx_time) {
 COUNT_T retune_fft_impl::process_tags_(const input_type *in, COUNT_T in_count,
                                        COUNT_T in_first,
                                        const input_type *fft_output) {
-  COUNT_T consumed = 0;
   COUNT_T produced = 0;
+  PROCESS_TAGS(
+      {
+        if (!reset_tags_) {
+          add_output_tags_(rx_time, rx_freq, produced);
+        }
 
-  FIND_TAGS
-
-  if (rx_freq_tags.empty()) {
-    process_items_(in_count, in, fft_output, consumed, produced);
-  } else {
-    PROCESS_TAGS({
-      in_first += rel;
-
-      if (rel > 0) {
-        process_items_(rel, in, fft_output, consumed, produced);
-      }
-
-      if (!reset_tags_) {
-        add_output_tags_(rx_time, rx_freq, produced);
-      }
-
-      if (pending_retune_) {
-        --pending_retune_;
-      }
-      process_buckets_(rx_freq, rx_time);
-    })
-    if (consumed < in_count) {
-      process_items_(in_count - consumed, in, fft_output, consumed, produced);
-    }
-  }
+        if (pending_retune_) {
+          --pending_retune_;
+        }
+        process_buckets_(rx_freq, rx_time);
+      },
+      in, fft_output, produced)
 
   return produced;
 }

diff --git a/lib/retune_fft_impl.h b/lib/retune_fft_impl.h
@@ -226,9 +226,8 @@ class retune_fft_impl : public retune_fft, base_impl, retuner_impl {
   void reset_items_();
   void calc_peaks_();
   void add_output_tags_(TIME_T rx_time, FREQ_T rx_freq, COUNT_T produced);
-  void process_items_(COUNT_T c, const input_type *&in,
-                      const input_type *&fft_output, COUNT_T &consumed,
-                      COUNT_T &produced);
+  void process_items_(COUNT_T c, COUNT_T &consumed, const input_type *&in,
+                      const input_type *&fft_output, COUNT_T &produced);
   void output_buckets_(const std::string &name,
                        const std::list<std::pair<double, double>> &buckets,
                        std::stringstream &ss);

diff --git a/lib/retune_pre_fft_impl.cc b/lib/retune_pre_fft_impl.cc
@@ -260,12 +260,13 @@ bool retune_pre_fft_impl::all_zeros_(const block_type *&in) {
   return *total_ == 0;
 }
 
-void retune_pre_fft_impl::process_items_(COUNT_T c, const block_type *&in,
+void retune_pre_fft_impl::process_items_(COUNT_T c, COUNT_T &consumed,
+                                         const block_type *&in,
                                          const block_type *&out,
-                                         COUNT_T &consumed, COUNT_T &produced) {
+                                         COUNT_T &produced) {
+  c /= nfft_;
   if (reset_tags_) {
-    for (COUNT_T i = 0; i < c; ++i, in += nfft_) {
-      ++consumed;
+    for (COUNT_T i = 0; i < c; ++i, in += nfft_, consumed += nfft_) {
       if (skip_fft_count_) {
         --skip_fft_count_;
         slew_samples_ += nfft_;
@@ -298,8 +299,7 @@ void retune_pre_fft_impl::process_items_(COUNT_T c, const block_type *&in,
       }
     }
   } else {
-    for (COUNT_T i = 0; i < c; ++i, in += nfft_) {
-      ++consumed;
+    for (COUNT_T i = 0; i < c; ++i, in += nfft_, consumed += nfft_) {
       if (skip_fft_count_) {
         --skip_fft_count_;
         slew_samples_ += nfft_;
@@ -315,41 +315,24 @@ void retune_pre_fft_impl::process_items_(COUNT_T c, const block_type *&in,
   }
 }
 
-COUNT_T retune_pre_fft_impl::process_tags_(COUNT_T in_nffts, COUNT_T in_count,
-                                           COUNT_T in_first,
+COUNT_T retune_pre_fft_impl::process_tags_(COUNT_T in_count, COUNT_T in_first,
                                            const block_type *in,
                                            const block_type *out) {
-  COUNT_T consumed = 0;
   COUNT_T produced = 0;
-
-  FIND_TAGS
-
-  if (rx_freq_tags.empty()) {
-    process_items_(in_nffts, in, out, consumed, produced);
-  } else {
-    // TODO: deprecate fft_batch_size, gr-wavelearner could use set_multiple
-    // abstraction like VkFFT
-    PROCESS_TAGS({
-      in_first += rel;
-      rel /= nfft_;
-
-      if (rel > 0) {
-        process_items_(rel, in, out, consumed, produced);
-      }
-
-      if (!reset_tags_) {
-        add_output_tags_(rx_time, rx_freq, produced);
-      }
-      if (pending_retune_) {
-        --pending_retune_;
-        fft_count_ = 0;
-        skip_fft_count_ = skip_tune_step_fft_;
-      }
-    })
-    if (consumed < in_nffts) {
-      process_items_(in_nffts - consumed, in, out, consumed, produced);
-    }
-  }
+  // TODO: deprecate fft_batch_size, gr-wavelearner could use set_multiple
+  // abstraction like VkFFT
+  PROCESS_TAGS(
+      {
+        if (!reset_tags_) {
+          add_output_tags_(rx_time, rx_freq, produced);
+        }
+        if (pending_retune_) {
+          --pending_retune_;
+          fft_count_ = 0;
+          skip_fft_count_ = skip_tune_step_fft_;
+        }
+      },
+      in, out, produced)
   return produced;
 }
 
@@ -369,7 +352,7 @@ int retune_pre_fft_impl::general_work(int noutput_items,
   }
   in_nffts = in_batches * fft_batch_size_;
   in_count = in_nffts * nfft_;
-  COUNT_T produced = process_tags_(in_nffts, in_count, in_first, in, out);
+  COUNT_T produced = process_tags_(in_count, in_first, in, out);
   consume_each(in_count);
   return produced / fft_batch_size_;
 }

diff --git a/lib/retune_pre_fft_impl.h b/lib/retune_pre_fft_impl.h
@@ -218,9 +218,9 @@ using block_type = gr_complex;
 class retune_pre_fft_impl : public retune_pre_fft, base_impl, retuner_impl {
 private:
   bool all_zeros_(const block_type *&in);
-  void process_items_(COUNT_T c, const block_type *&in, const block_type *&out,
-                      COUNT_T &consumed, COUNT_T &produced);
-  COUNT_T process_tags_(COUNT_T in_nffts, COUNT_T in_count, COUNT_T in_first,
+  void process_items_(COUNT_T c, COUNT_T &consumed, const block_type *&in,
+                      const block_type *&out, COUNT_T &produced);
+  COUNT_T process_tags_(COUNT_T in_count, COUNT_T in_first,
                         const block_type *in, const block_type *out);
   void add_output_tags_(TIME_T rx_time, FREQ_T rx_freq, COUNT_T rel);
 

diff --git a/lib/write_freq_samples_impl.cc b/lib/write_freq_samples_impl.cc
@@ -387,8 +387,8 @@ void write_freq_samples_impl::close_() {
   }
 }
 
-void write_freq_samples_impl::write_samples_(COUNT_T c, const char *&in,
-                                             COUNT_T &consumed) {
+void write_freq_samples_impl::process_items_(COUNT_T c, COUNT_T &consumed,
+                                             const char *&in) {
   for (COUNT_T i = 0; i < c;
        ++i, in += itemsize_ * vlen_, ++consumed, sample_clock_ += vlen_) {
     if (skip_tune_step_samples_count_) {
@@ -406,34 +406,19 @@ void write_freq_samples_impl::write_samples_(COUNT_T c, const char *&in,
 
 void write_freq_samples_impl::process_tags_(COUNT_T in_count, COUNT_T in_first,
                                             const char *in) {
-  COUNT_T consumed = 0;
-
-  FIND_TAGS
-
-  if (all_tags.empty()) {
-    write_samples_(in_count, in, consumed);
-  } else {
-    PROCESS_TAGS({
-      in_first += rel;
-
-      if (rel > 0) {
-        write_samples_(rel, in, consumed);
-      }
-
-      if (rotate_) {
-        open_(1);
-      }
-      if (sigmf_) {
-        capture_item_type capture_item;
-        capture_item.rx_freq = rx_freq;
-        capture_item.sample_clock = sample_clock_;
-        capture_q_.push(capture_item);
-      }
-    })
-    if (consumed < in_count) {
-      write_samples_(in_count - consumed, in, consumed);
-    }
-  }
+  PROCESS_TAGS(
+      {
+        if (rotate_) {
+          open_(1);
+        }
+        if (sigmf_) {
+          capture_item_type capture_item;
+          capture_item.rx_freq = rx_freq;
+          capture_item.sample_clock = sample_clock_;
+          capture_q_.push(capture_item);
+        }
+      },
+      in)
 }
 
 #pragma GCC diagnostic push

diff --git a/lib/write_freq_samples_impl.h b/lib/write_freq_samples_impl.h
@@ -237,7 +237,7 @@ class write_freq_samples_impl : public write_freq_samples, base_impl {
   void write_(const char *data, COUNT_T len);
   void open_(COUNT_T zlevel);
   void close_();
-  void write_samples_(COUNT_T c, const char *&in, COUNT_T &consumed);
+  void process_items_(COUNT_T c, COUNT_T &consumed, const char *&in);
   void recv_inference_(const pmt::pmt_t msg);
   void process_tags_(COUNT_T in_count, COUNT_T in_first, const char *in);