[Commit-gnuradio] [gnuradio] 01/11: gr-uhd: added tagged stream support

[git]

This is an automated email from the git hooks/post-receive script.

jcorgan pushed a commit to branch master
in repository gnuradio.

commit e05164d15e8921760ff3649ae058517cab9dee40
Author: Sean Nowlan <address@hidden>
Date:   Sun Feb 23 23:02:32 2014 -0500

    gr-uhd: added tagged stream support
---
 gr-uhd/examples/c++/tag_source_demo.h   | 42 ++++++++++++++++++++++---
 gr-uhd/examples/c++/tags_demo.cc        |  8 +++--
 gr-uhd/include/gnuradio/uhd/usrp_sink.h | 22 +++++++++++--
 gr-uhd/lib/usrp_sink_impl.cc            | 56 +++++++++++++++++++++++++++------
 gr-uhd/lib/usrp_sink_impl.h             |  5 ++-
 5 files changed, 113 insertions(+), 20 deletions(-)

diff --git a/gr-uhd/examples/c++/tag_source_demo.h 
b/gr-uhd/examples/c++/tag_source_demo.h
index 71fb944..904087c 100644
--- a/gr-uhd/examples/c++/tag_source_demo.h
+++ b/gr-uhd/examples/c++/tag_source_demo.h
@@ -35,7 +35,8 @@ public:
         const double start_fracs,
         const double samp_rate,
         const double idle_duration,
-        const double burst_duration
+        const double burst_duration,
+        const std::string &length_tag_key = ""
     ):
         sync_block(
             "uhd tag source demo",
@@ -48,7 +49,9 @@ public:
         _samps_per_burst(samp_rate*burst_duration),
         _cycle_duration(idle_duration + burst_duration),
         _samps_left_in_burst(1), //immediate EOB
-        _do_new_burst(false)
+        _do_new_burst(false),
+        _firstrun(not length_tag_key.empty()),
+        _length_tag_key(length_tag_key)
     {
         //NOP
     }
@@ -80,6 +83,19 @@ public:
         this->add_item_tag(0/*chan0*/, tag_count, key, value, srcid);
     }
 
+    void make_length_tag(const uint64_t tag_count, const uint64_t burst_len)
+    {
+        if (_length_tag_key.empty()){
+            //no length_tag was specified at initialization; make a tx_sob tag 
instead
+            this->make_sob_tag(tag_count);
+            return;
+        }
+        const pmt::pmt_t key = pmt::string_to_symbol(_length_tag_key);
+        const pmt::pmt_t value = pmt::from_long((long)burst_len);
+        const pmt::pmt_t srcid = pmt::string_to_symbol(this->name());
+        this->add_item_tag(0/*chan0*/, tag_count, key, value, srcid);
+    }
+
     int work(
         int noutput_items,
         gr_vector_const_void_star &input_items,
@@ -98,7 +114,18 @@ public:
             _do_new_burst = false;
             _samps_left_in_burst = _samps_per_burst;
 
-            this->make_sob_tag(this->nitems_written(0));
+            if (_length_tag_key.empty())
+                this->make_sob_tag(this->nitems_written(0));
+            else
+#if 1
+                this->make_length_tag(this->nitems_written(0), 
_samps_left_in_burst);
+#else
+                //Test usrp_sink's ability to cancel remainder of burst if new 
length_tag is found early
+                //sets burst time to 10% greater than the cycle duration to 
guarantee early length_tag
+                //In a real implementation the user should guard against this 
so that the number of
+                //samples promised by the length_tag are actually processed by 
the usrp_sink.
+                this->make_length_tag(this->nitems_written(0), uint64_t(1.1 * 
_samp_rate * _cycle_duration));
+#endif
             this->make_time_tag(this->nitems_written(0));
 
             _time_fracs += _cycle_duration;
@@ -111,7 +138,12 @@ public:
         //Tag an end of burst and return early.
         //the next work call will be a start of burst.
         if (_samps_left_in_burst < size_t(noutput_items)){
-            this->make_eob_tag(this->nitems_written(0) + _samps_left_in_burst 
- 1);
+            if (_length_tag_key.empty())
+                this->make_eob_tag(this->nitems_written(0) + 
_samps_left_in_burst - 1);
+            else if (_firstrun){
+                _firstrun = false;
+                this->make_length_tag(this->nitems_written(0), 1);
+            }
             _do_new_burst = true;
             noutput_items = _samps_left_in_burst;
         }
@@ -128,5 +160,7 @@ private:
     const double _cycle_duration;
     uint64_t _samps_left_in_burst;
     bool _do_new_burst;
+    bool _firstrun;
+    const std::string _length_tag_key;
 
 };
diff --git a/gr-uhd/examples/c++/tags_demo.cc b/gr-uhd/examples/c++/tags_demo.cc
index 0f87109..a4e2317 100644
--- a/gr-uhd/examples/c++/tags_demo.cc
+++ b/gr-uhd/examples/c++/tags_demo.cc
@@ -43,7 +43,7 @@ void sig_int_handler(int){stop_signal_called = true;}
  **********************************************************************/
 int main(int argc, char *argv[]){
 
-    std::string device_addr;
+    std::string device_addr, length_tag;
     double center_freq, samp_rate, burst_dur, idle_dur;
 
     //setup the program options
@@ -55,6 +55,7 @@ int main(int argc, char *argv[]){
         ("freq", po::value<double>(&center_freq)->default_value(10e6), "the 
center frequency in Hz")
         ("burst", po::value<double>(&burst_dur)->default_value(0.1), "the 
duration of each burst in seconds")
         ("idle", po::value<double>(&idle_dur)->default_value(0.05), "idle time 
between bursts in seconds")
+        ("length_tag", po::value<std::string>(&length_tag)->default_value(""), 
"the length tag key name")
     ;
     po::variables_map vm;
     po::store(po::parse_command_line(argc, argv, desc), vm);
@@ -95,14 +96,15 @@ int main(int argc, char *argv[]){
     //-- make the usrp sink test blocks
     //------------------------------------------------------------------
     gr::uhd::usrp_sink::sptr usrp_sink = gr::uhd::usrp_sink::make
-      (device_addr, uhd::stream_args_t("fc32"));
+      (device_addr, uhd::stream_args_t("fc32"), length_tag);
     usrp_sink->set_samp_rate(samp_rate);
     usrp_sink->set_center_freq(center_freq);
     const uhd::time_spec_t time_now = usrp_sink->get_time_now();
+    const double actual_samp_rate = usrp_sink->get_samp_rate();
 
     boost::shared_ptr<tag_source_demo> tag_source = 
boost::make_shared<tag_source_demo>(
         time_now.get_full_secs() + 1, time_now.get_frac_secs(), //time now + 1 
second
-        samp_rate, idle_dur, burst_dur
+        actual_samp_rate, idle_dur, burst_dur
     );
 
     //------------------------------------------------------------------
diff --git a/gr-uhd/include/gnuradio/uhd/usrp_sink.h 
b/gr-uhd/include/gnuradio/uhd/usrp_sink.h
index 18640fe..08bad49 100644
--- a/gr-uhd/include/gnuradio/uhd/usrp_sink.h
+++ b/gr-uhd/include/gnuradio/uhd/usrp_sink.h
@@ -72,10 +72,17 @@ namespace gr {
        *  - pmt::string_to_symbol("tx_sob")
        *  - pmt::string_to_symbol("tx_eob")
        *  - pmt::string_to_symbol("tx_time")
+       *  - pmt::string_to_symbol(length_tag_key)
        *
        * The sob and eob (start and end of burst) tag values are pmt booleans.
        * When present, burst tags should be set to true (pmt::PMT_T).
        *
+       * If length_tag_key is not an empty string, all "tx_sob" and "tx_eob" 
+       * tags will be ignored.
+       *
+       * The length tag value should be a pmt long specifying the number
+       * of samples contained in the corresponding tagged stream.
+       *
        * The timstamp tag value is a pmt tuple of the following:
        * (uint64 seconds, and double fractional seconds).
        *
@@ -85,11 +92,13 @@ namespace gr {
        * \param device_addr the address to identify the hardware
        * \param io_type the desired input data type
        * \param num_channels number of stream from the device
+       * \param length_tag_key the key of the tag identifying tagged stream 
length
        * \return a new USRP sink block object
        */
       static sptr make(const ::uhd::device_addr_t &device_addr,
                        const ::uhd::io_type_t &io_type,
-                       size_t num_channels);
+                       size_t num_channels,
+                       const std::string &length_tag_key = "");
 
       /*!
        * \brief Make a new USRP sink block.
@@ -103,10 +112,17 @@ namespace gr {
        *  - pmt::string_to_symbol("tx_sob")
        *  - pmt::string_to_symbol("tx_eob")
        *  - pmt::string_to_symbol("tx_time")
+       *  - pmt::string_to_symbol(length_tag_key)
        *
        * The sob and eob (start and end of burst) tag values are pmt booleans.
        * When present, burst tags should be set to true (pmt::PMT_T).
        *
+       * If length_tag_key is not an empty string, all "tx_sob" and "tx_eob" 
+       * tags will be ignored.
+       *
+       * The length tag value should be a pmt long specifying the number
+       * of samples contained in the corresponding tagged stream.
+       *
        * The timstamp tag value is a pmt tuple of the following:
        * (uint64 seconds, and double fractional seconds).
        *
@@ -115,10 +131,12 @@ namespace gr {
        *
        * \param device_addr the address to identify the hardware
        * \param stream_args the IO format and channel specification
+       * \param length_tag_key the key of the tag identifying tagged stream 
length
        * \return a new USRP sink block object
        */
       static sptr make(const ::uhd::device_addr_t &device_addr,
-                       const ::uhd::stream_args_t &stream_args);
+                       const ::uhd::stream_args_t &stream_args,
+                       const std::string &length_tag_key = "");
 
       /*!
        * Set the start time for outgoing samples.
diff --git a/gr-uhd/lib/usrp_sink_impl.cc b/gr-uhd/lib/usrp_sink_impl.cc
index 828537f..759019f 100644
--- a/gr-uhd/lib/usrp_sink_impl.cc
+++ b/gr-uhd/lib/usrp_sink_impl.cc
@@ -32,7 +32,8 @@ namespace gr {
     usrp_sink::sptr
     usrp_sink::make(const ::uhd::device_addr_t &device_addr,
                     const ::uhd::io_type_t &io_type,
-                    size_t num_channels)
+                    size_t num_channels,
+                    const std::string &length_tag_key)
     {
       //fill in the streamer args
       ::uhd::stream_args_t stream_args;
@@ -46,27 +47,31 @@ namespace gr {
       for(size_t chan = 0; chan < num_channels; chan++)
         stream_args.channels.push_back(chan); //linear mapping
 
-      return usrp_sink::make(device_addr, stream_args);
+      return usrp_sink::make(device_addr, stream_args, length_tag_key);
     }
 
     usrp_sink::sptr
     usrp_sink::make(const ::uhd::device_addr_t &device_addr,
-                    const ::uhd::stream_args_t &stream_args)
+                    const ::uhd::stream_args_t &stream_args,
+                    const std::string &length_tag_key)
     {
       check_abi();
       return usrp_sink::sptr
-        (new usrp_sink_impl(device_addr, stream_args_ensure(stream_args)));
+        (new usrp_sink_impl(device_addr, stream_args_ensure(stream_args), 
length_tag_key));
     }
 
     usrp_sink_impl::usrp_sink_impl(const ::uhd::device_addr_t &device_addr,
-                                   const ::uhd::stream_args_t &stream_args)
+                                   const ::uhd::stream_args_t &stream_args,
+                                   const std::string &length_tag_key)
       : sync_block("gr uhd usrp sink",
                       args_to_io_sig(stream_args),
                       io_signature::make(0, 0, 0)),
         _stream_args(stream_args),
         _nchan(stream_args.channels.size()),
         _stream_now(_nchan == 1),
-        _start_time_set(false)
+        _start_time_set(false),
+        _length_tag_key(length_tag_key),
+        _nitems_to_send(0)
     {
       if(stream_args.cpu_format == "fc32")
         _type = boost::make_shared< ::uhd::io_type_t 
>(::uhd::io_type_t::COMPLEX_FLOAT32);
@@ -476,6 +481,16 @@ namespace gr {
       if(not _tags.empty())
         this->tag_work(ninput_items);
 
+      //check if there is data left to send from a burst tagged with length_tag
+      //note: if a burst is started during this call to work(), tag_work() 
should
+      //have been called and we should have _nitems_to_send > 0.
+      if(not _length_tag_key.empty() and _nitems_to_send > 0) {
+        ninput_items = std::min<long>(_nitems_to_send, ninput_items);
+        //if we run out of items to send, it's the end of the burst
+        if(_nitems_to_send - long(ninput_items) == 0)
+          _metadata.end_of_burst = true;
+      }
+
 #ifdef GR_UHD_USE_STREAM_API
       //send all ninput_items with metadata
       const size_t num_sent = _tx_stream->send
@@ -486,6 +501,10 @@ namespace gr {
          *_type, ::uhd::device::SEND_MODE_FULL_BUFF, 1.0);
 #endif
 
+      //if using length_tags, decrement items left to send by the number of 
samples sent
+      if(not _length_tag_key.empty() and _nitems_to_send > 0)
+        _nitems_to_send -= long(num_sent);
+
       //increment the timespec by the number of samples sent
       _metadata.time_spec += ::uhd::time_spec_t(0, num_sent, _sample_rate);
       return num_sent;
@@ -528,18 +547,33 @@ namespace gr {
           break;
         }
 
-        //handle end of burst with a mini end of burst packet
-        else if(pmt::equal(key, EOB_KEY)) {
+        //handle end of burst with a mini end of burst packet; ignore if we 
have a nonempty length_tag_key string
+        else if(_length_tag_key.empty() and pmt::equal(key, EOB_KEY)) {
           _metadata.end_of_burst = pmt::to_bool(value);
           ninput_items = 1;
           return;
         }
 
-        //set the start of burst flag in the metadata
-        else if(pmt::equal(key, SOB_KEY)) {
+        //set the start of burst flag in the metadata; ignore if we have a 
nonempty length_tag_key string
+        else if(_length_tag_key.empty() and pmt::equal(key, SOB_KEY)) {
           _metadata.start_of_burst = pmt::to_bool(value);
         }
 
+        //length_tag found; set the start of burst flag in the metadata
+        else if(not _length_tag_key.empty() and pmt::equal(key, 
pmt::string_to_symbol(_length_tag_key))) {
+          //If there are still items left to send, we will truncate the 
previous burst 
+          //by setting the end of burst flag in a mini end of burst packet. 
The next 
+          //call to work will start at the new burst.
+          if(_nitems_to_send > 0) {
+              ninput_items = 0;
+              _nitems_to_send = 0;
+              _metadata.end_of_burst = true;
+              return;
+          }
+          _nitems_to_send = pmt::to_long(value);
+          _metadata.start_of_burst = true;
+        }
+
         //set the time specification in the metadata
         else if(pmt::equal(key, TIME_KEY)) {
           _metadata.has_time_spec = true;
@@ -570,6 +604,7 @@ namespace gr {
       _metadata.start_of_burst = true;
       _metadata.end_of_burst = false;
       _metadata.has_time_spec = not _stream_now;
+      _nitems_to_send = 0;
       if(_start_time_set) {
         _start_time_set = false; //cleared for next run
         _metadata.time_spec = _start_time;
@@ -597,6 +632,7 @@ namespace gr {
       _metadata.start_of_burst = false;
       _metadata.end_of_burst = true;
       _metadata.has_time_spec = false;
+      _nitems_to_send = 0;
 
 #ifdef GR_UHD_USE_STREAM_API
       _tx_stream->send(gr_vector_const_void_star(_nchan), 0, _metadata, 1.0);
diff --git a/gr-uhd/lib/usrp_sink_impl.h b/gr-uhd/lib/usrp_sink_impl.h
index c714eb5..873c01f 100644
--- a/gr-uhd/lib/usrp_sink_impl.h
+++ b/gr-uhd/lib/usrp_sink_impl.h
@@ -53,7 +53,8 @@ namespace gr {
     {
     public:
        usrp_sink_impl(const ::uhd::device_addr_t &device_addr,
-                      const ::uhd::stream_args_t &stream_args);
+                      const ::uhd::stream_args_t &stream_args,
+                      const std::string &length_tag_key);
       ~usrp_sink_impl();
 
       void setup_rpc();
@@ -135,6 +136,8 @@ namespace gr {
 
       //stream tags related stuff
       std::vector<tag_t> _tags;
+      const std::string _length_tag_key;
+      long _nitems_to_send;
     };
 
   } /* namespace uhd */