blocks: Added tagged stream mux feature to preserve "head" position of tags on other inputs than 0

Author: Martin Braun

gr-blocks/grc/blocks_tagged_stream_mux.xml

@@ -2,7 +2,7 @@
   <name>Tagged Stream Mux</name>
   <key>blocks_tagged_stream_mux</key>
   <import>from gnuradio import blocks</import>
-  <make>blocks.tagged_stream_mux($type.size*$vlen, $lengthtagname)</make>
+  <make>blocks.tagged_stream_mux($type.size*$vlen, $lengthtagname, $tag_preserve_head_pos)</make>
 	<param>
 		<name>IO Type</name>
 		<key>type</key>
@@ -49,6 +49,13 @@
     <value>1</value>
     <type>int</type>
   </param>
+  <param>
+    <name>Tags: Preserve head position on input</name>
+    <key>tag_preserve_head_pos</key>
+    <value>0</value>
+    <type>int</type>
+    <hide>part</hide>
+  </param>
   <sink>
     <name>in</name>
     <type>$type</type>

gr-blocks/include/gnuradio/blocks/tagged_stream_mux.h

@@ -41,6 +41,11 @@ namespace gr {
      *
      * All other tags are propagated as expected, i.e. they stay associated
      * with the same input item.
+     * There are cases when this behaviour is undesirable. One special case
+     * is when a tag at the first element (the head item) of one input port
+     * must stay on the head item of the output port. To achieve this,
+     * set \p tag_preserve_head_pos to the port that will receive these special
+     * tags.
      */
     class BLOCKS_API tagged_stream_mux : virtual public tagged_stream_block
     {
@@ -52,8 +57,9 @@ namespace gr {
        *
        * \param itemsize Items size (number of bytes per item)
        * \param lengthtagname Length tag key
+       * \param tag_preserve_head_pos Preserves the head position of tags on this input port
        */
-      static sptr make(size_t itemsize, const std::string &lengthtagname);
+      static sptr make(size_t itemsize, const std::string &lengthtagname, unsigned int tag_preserve_head_pos=0);
     };
 
   } // namespace blocks

gr-blocks/lib/tagged_stream_mux_impl.cc

@@ -31,17 +31,18 @@ namespace gr {
   namespace blocks {
 
     tagged_stream_mux::sptr
-    tagged_stream_mux::make(size_t itemsize, const std::string &lengthtagname)
+    tagged_stream_mux::make(size_t itemsize, const std::string &lengthtagname, unsigned int tag_preserve_head_pos)
     {
-      return gnuradio::get_initial_sptr (new tagged_stream_mux_impl(itemsize, lengthtagname));
+      return gnuradio::get_initial_sptr (new tagged_stream_mux_impl(itemsize, lengthtagname, tag_preserve_head_pos));
     }
 
-    tagged_stream_mux_impl::tagged_stream_mux_impl(size_t itemsize, const std::string &lengthtagname)
+    tagged_stream_mux_impl::tagged_stream_mux_impl(size_t itemsize, const std::string &lengthtagname, unsigned int tag_preserve_head_pos)
       : tagged_stream_block("tagged_stream_mux",
                  io_signature::make(1, -1, itemsize),
                  io_signature::make(1,  1, itemsize),
 		 lengthtagname),
-        d_itemsize(itemsize)
+        d_itemsize(itemsize),
+	d_tag_preserve_head_pos(tag_preserve_head_pos)
     {
       set_tag_propagation_policy(TPP_DONT);
     }
@@ -77,7 +78,10 @@ namespace gr {
 	std::vector<tag_t> tags;
 	get_tags_in_range(tags, i, nitems_read(i), nitems_read(i)+ninput_items[i]);
 	for (unsigned int j = 0; j < tags.size(); j++) {
-	  const uint64_t offset = tags[j].offset - nitems_read(i) + nitems_written(0) + n_produced;
+	  uint64_t offset = tags[j].offset - nitems_read(i) + nitems_written(0) + n_produced;
+	  if (i == d_tag_preserve_head_pos && tags[j].offset == nitems_read(i)) {
+	    offset -= n_produced;
+	  }
 	  add_item_tag(0, offset, tags[j].key, tags[j].value);
 	}
 	memcpy((void *) out, (const void *) in, ninput_items[i] * d_itemsize);

gr-blocks/lib/tagged_stream_mux_impl.h

@@ -33,12 +33,13 @@ namespace gr {
     {
      private:
       size_t d_itemsize;
+      unsigned int d_tag_preserve_head_pos;
 
      protected:
-      int calculate_output_stream_length(const std::vector<int> &ninput_items);
+      int calculate_output_stream_length(const gr_vector_int &ninput_items);
 
      public:
-      tagged_stream_mux_impl(size_t itemsize, const std::string &lengthtagname);
+      tagged_stream_mux_impl(size_t itemsize, const std::string &lengthtagname, unsigned int d_tag_preserve_head_pos);
       ~tagged_stream_mux_impl();
 
       int work(int noutput_items,

gr-blocks/python/blocks/qa_tagged_stream_mux.py

@@ -104,6 +104,40 @@ def test_1(self):
         ]
         self.assertEqual(tags, tags_expected)
 
+    def test_preserve_tag_head_pos(self):
+        """ Test the 'preserve head position' function.
+        This will add a 'special' tag to item 0 on stream 1.
+        It should be on item 0 of the output stream. """
+        special_tag = gr.tag_t()
+        special_tag.key = pmt.string_to_symbol('spam')
+        special_tag.offset = 0
+        special_tag.value = pmt.to_pmt('eggs')
+        len_tag_key = "length"
+        packet_len_1 = 5
+        packet_len_2 = 3
+        mux = blocks.tagged_stream_mux(gr.sizeof_float, len_tag_key, 1)
+        sink = blocks.vector_sink_f()
+        self.tb.connect(
+            blocks.vector_source_f(range(packet_len_1)),
+            blocks.stream_to_tagged_stream(gr.sizeof_float, 1, packet_len_1, len_tag_key),
+            (mux, 0)
+        )
+        self.tb.connect(
+            blocks.vector_source_f(range(packet_len_2), False, 1, (special_tag,)),
+            blocks.stream_to_tagged_stream(gr.sizeof_float, 1, packet_len_2, len_tag_key),
+            (mux, 1)
+        )
+        self.tb.connect(mux, sink)
+        self.tb.run()
+        self.assertEqual(sink.data(), tuple(range(packet_len_1) + range(packet_len_2)))
+        tags = [gr.tag_to_python(x) for x in sink.tags()]
+        tags = sorted([(x.offset, x.key, x.value) for x in tags])
+        tags_expected = [
+                (0, 'length', packet_len_1 + packet_len_2),
+                (0, 'spam', 'eggs'),
+        ]
+        self.assertEqual(tags, tags_expected)
+
 
 if __name__ == '__main__':
     gr_unittest.run(qa_tagged_stream_mux, "qa_tagged_stream_mux.xml")