Begin tag stream to vector stream

5 files changed, 146 insertions, 86 deletions

diff --git a/tests/correlator/correlator.grc b/tests/correlator/correlator.grc
index 9fc6699..555ad15 100644
--- a/tests/correlator/correlator.grc
+++ b/tests/correlator/correlator.grc
@@ -251,7 +251,7 @@ blocks:
     bus_sink: false
     bus_source: false
     bus_structure: null
-    coordinate: [512, 1252.0]
+    coordinate: [496, 1252.0]
     rotation: 0
     state: true
 - name: blocks_stream_mux_0
@@ -273,24 +273,25 @@ blocks:
     coordinate: [288, 256.0]
     rotation: 0
     state: enabled
-- name: blocks_tagged_stream_align_0
-  id: blocks_tagged_stream_align
+- name: blocks_tag_debug_0
+  id: blocks_tag_debug
   parameters:
     affinity: ''
     alias: ''
     comment: ''
-    lengthtagname: frame_start
-    maxoutbuf: '0'
-    minoutbuf: '0'
+    display: 'True'
+    filter: '""'
+    name: ''
+    num_inputs: '1'
     type: byte
     vlen: '1'
   states:
     bus_sink: false
     bus_source: false
     bus_structure: null
-    coordinate: [768, 1260.0]
+    coordinate: [744, 1164.0]
     rotation: 0
-    state: true
+    state: disabled
 - name: blocks_throttle_0
   id: blocks_throttle
   parameters:
@@ -471,56 +472,57 @@ blocks:
 - name: epy_block_0
   id: epy_block
   parameters:
-    _source_code: "import pmt\n\nimport numpy as np\nfrom gnuradio import gr\n\n\n\
-      class blk(gr.sync_block):\n    \"\"\"\n    Apply phase and frequency correction\
-      \ where there is a correlation peak tag.\n\n    The correlation peak tags are\
-      \ NOT propagated, and instead replaced with a\n    frame_start tag.\n    \"\"\
-      \"\n    def __init__(self):\n        gr.sync_block.__init__(\n            self,\n\
-      \            name='Phase and Frequency Correction',\n            in_sig=[np.complex64],\n\
-      \            out_sig=[np.complex64]\n        )\n\n        # tags should not\
-      \ be propagated, we then output our own tags\n        self.set_tag_propagation_policy(gr.TPP_DONT)\n\
-      \n        # because we do block processing, we need to keep track of the last\
-      \ tag\n        # of the previous block to correct the first values of the next\
-      \ block\n        self.last = None\n        self.lastfreq = 0\n        self.lastnback\
-      \ = 0\n        self.lastnsamples = 0\n\n    def block_phase(self, start, end):\n\
-      \        \"\"\"\n        Compute a vector for the phase and frequency correction\
-      \ for the samples\n        between two tags (start and end).\n\n        @param\
-      \ start Tag where the samples should start to be corrected\n        @param end\
-      \   Tag where to stop correcting\n\n        @return A vector of phase values\
-      \ for each sample. To correct the samples\n                the data should be\
-      \ multiplied with np.exp(-1j * phase)\n        \"\"\"\n        # compute number\
-      \ of samples between tags\n        nsamples = end.offset - start.offset\n\n\
-      \        # debugging, see last lines of self.work()\n        self.lastnsamples\
-      \ = nsamples\n\n        # unpack pmt values into start and end phase\n     \
-      \   sphase = pmt.to_python(start.value)\n        ephase = pmt.to_python(end.value)\n\
-      \n        # compute frequency offset between start and end\n        phasediff\
-      \ = (ephase - sphase) % (2 * np.pi)\n        freq = phasediff / nsamples\n\n\
-      \        # save this one for the last block (see variable `end' in self.work)\n\
-      \        self.lastfreq = freq\n\n        # debugging\n        print(f\"Correction\
-      \ for chunk of {nsamples:2d} samples is \" \\\n              f\"sphase={sphase:\
-      \ .4f} rad and freq={freq*1e3: .4f}e-3 rad / sample\")\n\n        # compute\
-      \ chunk values\n        return sphase * np.ones(nsamples) + freq * np.arange(0,\
-      \ nsamples)\n\n    def work(self, input_items, output_items):\n        counter\
-      \ = self.nitems_written(0)\n\n        # nicer aliases\n        inp = input_items[0]\n\
-      \        out = output_items[0]\n\n        # read phase tags\n        is_phase\
-      \ = lambda tag: pmt.to_python(tag.key) == \"phase_est\"\n        tags = list(filter(is_phase,\
-      \ self.get_tags_in_window(0, 0, len(inp))))\n\n        if not tags:\n      \
-      \      print(f\"There were no tags in {len(inp)} samples!\")\n            out[:]\
-      \ = inp\n            return len(out)\n\n        # debugging\n        print(f\"\
-      Processing {len(tags)} tags = {tags[-1].offset - tags[0].offset} \" \\\n   \
-      \           f\"samples out of {len(inp)} input samples\")\n\n        # compute\
-      \ \"the middle\"\n        enough_samples = lambda pair: ((pair[1].offset - pair[0].offset)\
-      \ > 0)\n        pairs = list(filter(enough_samples, zip(tags, tags[1:])))\n\
-      \        chunks = [ self.block_phase(start, end) for (start, end) in pairs ]\n\
-      \        middle = np.concatenate(chunks) if chunks else []\n\n        # compute\
-      \ values at the end, we do not have informations about the future\n        #\
-      \ but we can use the frequency of the last tag to approximate\n        nback\
-      \ = len(inp) - (tags[-1].offset - counter)\n        print(f\"Processing {nback}\
-      \ samples at the back of the buffer\")\n        end = np.ones(nback) * pmt.to_python(tags[-1].value)\
-      \ \\\n                + self.lastfreq * np.arange(0, nback)\n\n\n        # compute\
-      \ the \"start\", using the last tag from the previous call\n        nfront =\
-      \ tags[0].offset - counter\n        print(f\"Processing {nfront} samples at\
-      \ the front of the buffer\")\n        start = self.block_phase(self.last, tags[0])[-nfront:]\
+    _source_code: "import pmt\n\nimport numpy as np\nfrom gnuradio import gr\n\nprint\
+      \ = lambda x: None\n\nclass blk(gr.sync_block):\n    \"\"\"\n    Apply phase\
+      \ and frequency correction where there is a correlation peak tag.\n\n    The\
+      \ correlation peak tags are NOT propagated, and instead replaced with a\n  \
+      \  frame_start tag.\n    \"\"\"\n    def __init__(self):\n        gr.sync_block.__init__(\n\
+      \            self,\n            name='Phase and Frequency Correction',\n   \
+      \         in_sig=[np.complex64],\n            out_sig=[np.complex64]\n     \
+      \   )\n\n        # tags should not be propagated, we then output our own tags\n\
+      \        self.set_tag_propagation_policy(gr.TPP_DONT)\n\n        # because we\
+      \ do block processing, we need to keep track of the last tag\n        # of the\
+      \ previous block to correct the first values of the next block\n        self.last\
+      \ = None\n        self.lastfreq = 0\n        self.lastnback = 0\n        self.lastnsamples\
+      \ = 0\n\n    def block_phase(self, start, end):\n        \"\"\"\n        Compute\
+      \ a vector for the phase and frequency correction for the samples\n        between\
+      \ two tags (start and end).\n\n        @param start Tag where the samples should\
+      \ start to be corrected\n        @param end   Tag where to stop correcting\n\
+      \n        @return A vector of phase values for each sample. To correct the samples\n\
+      \                the data should be multiplied with np.exp(-1j * phase)\n  \
+      \      \"\"\"\n        # compute number of samples between tags\n        nsamples\
+      \ = end.offset - start.offset\n\n        # debugging, see last lines of self.work()\n\
+      \        self.lastnsamples = nsamples\n\n        # unpack pmt values into start\
+      \ and end phase\n        sphase = pmt.to_python(start.value)\n        ephase\
+      \ = pmt.to_python(end.value)\n\n        # compute frequency offset between start\
+      \ and end\n        phasediff = (ephase - sphase) % (2 * np.pi)\n        freq\
+      \ = phasediff / nsamples\n\n        # save this one for the last block (see\
+      \ variable `end' in self.work)\n        self.lastfreq = freq\n\n        # debugging\n\
+      \        print(f\"Correction for chunk of {nsamples:2d} samples is \" \\\n \
+      \             f\"sphase={sphase: .4f} rad and freq={freq*1e3: .4f}e-3 rad /\
+      \ sample\")\n\n        # compute chunk values\n        return sphase * np.ones(nsamples)\
+      \ + freq * np.arange(0, nsamples)\n\n    def work(self, input_items, output_items):\n\
+      \        counter = self.nitems_written(0)\n\n        # nicer aliases\n     \
+      \   inp = input_items[0]\n        out = output_items[0]\n\n        # read phase\
+      \ tags\n        is_phase = lambda tag: pmt.to_python(tag.key) == \"phase_est\"\
+      \n        tags = list(filter(is_phase, self.get_tags_in_window(0, 0, len(inp))))\n\
+      \n        if not tags:\n            print(f\"There were no tags in {len(inp)}\
+      \ samples!\")\n            out[:] = inp\n            return len(out)\n\n   \
+      \     # debugging\n        print(f\"Processing {len(tags)} tags = {tags[-1].offset\
+      \ - tags[0].offset} \" \\\n              f\"samples out of {len(inp)} input\
+      \ samples\")\n\n        # compute \"the middle\"\n        enough_samples = lambda\
+      \ pair: ((pair[1].offset - pair[0].offset) > 0)\n        pairs = list(filter(enough_samples,\
+      \ zip(tags, tags[1:])))\n        chunks = [ self.block_phase(start, end) for\
+      \ (start, end) in pairs ]\n        middle = np.concatenate(chunks) if chunks\
+      \ else []\n\n        # compute values at the end, we do not have informations\
+      \ about the future\n        # but we can use the frequency of the last tag to\
+      \ approximate\n        nback = len(inp) - (tags[-1].offset - counter)\n    \
+      \    print(f\"Processing {nback} samples at the back of the buffer\")\n    \
+      \    end = np.ones(nback) * pmt.to_python(tags[-1].value) \\\n             \
+      \   + self.lastfreq * np.arange(0, nback)\n\n\n        # compute the \"start\"\
+      , using the last tag from the previous call\n        nfront = tags[0].offset\
+      \ - counter\n        print(f\"Processing {nfront} samples at the front of the\
+      \ buffer\")\n        start = self.block_phase(self.last, tags[0])[-nfront:]\
       \ \\\n                if self.last and nfront else np.zeros(nfront)\n\n    \
       \    # debugging\n        if nfront + self.lastnback != self.lastnsamples:\n\
       \            print(f\"Something went wrong: {self.lastnback + nfront} != self.lastnsamples\"\
@@ -549,27 +551,55 @@ blocks:
 - name: epy_block_1
   id: epy_block
   parameters:
-    _source_code: "\"\"\"\nEmbedded Python Blocks:\n\nEach time this file is saved,\
-      \ GRC will instantiate the first class it finds\nto get ports and parameters\
-      \ of your block. The arguments to __init__  will\nbe the parameters. All of\
-      \ them are required to have default values!\n\"\"\"\n\nimport numpy as np\n\
-      from gnuradio import gr\n\nnp.set_printoptions(formatter={'int':hex})\n\nclass\
-      \ blk(gr.sync_block):\n    def __init__(self):\n        gr.sync_block.__init__(\n\
-      \            self,\n            name='Printer',\n            in_sig=[np.byte],\n\
-      \            out_sig=[]\n        )\n\n    def work(self, input_items, output_items):\n\
-      \        inp = np.array(input_items[0], dtype=np.uint8)\n        print(f\"Decoded\
-      \ {len(inp)} samples:\\n{inp}\")\n\n        return len(inp)\n"
+    _source_code: "import numpy as np\nfrom gnuradio import gr\n\nnp.set_printoptions(formatter={'int':hex})\n\
+      \nclass blk(gr.sync_block):\n    def __init__(self, vlen=1):\n        dt = np.byte\
+      \ if vlen == 1 else (np.byte, vlen)\n\n        gr.sync_block.__init__(\n   \
+      \         self,\n            name='Printer',\n            in_sig=[(np.byte,\
+      \ vlen)],\n            out_sig=[]\n        )\n\n    def work(self, input_items,\
+      \ output_items):\n        inp = np.array(input_items[0], dtype=np.uint8)\n \
+      \       print(f\"Decoded {len(inp)} samples:\\n{inp}\")\n\n        return len(inp)\n"
+    affinity: ''
+    alias: ''
+    comment: ''
+    maxoutbuf: '0'
+    minoutbuf: '0'
+    vlen: '20'
+  states:
+    _io_cache: ('Printer', 'blk', [('vlen', '1')], [('0', 'byte', 1)], [], '', [])
+    bus_sink: false
+    bus_source: false
+    bus_structure: null
+    coordinate: [952, 1260.0]
+    rotation: 0
+    state: true
+- name: epy_block_2
+  id: epy_block
+  parameters:
+    _source_code: "import pmt\nimport numpy as np\nfrom gnuradio import gr\n\n\nclass\
+      \ blk(gr.sync_block):\n\n    def __init__(self, tag=\"frame_start\", vlen=1):\n\
+      \        dt = np.byte if vlen == 1 else (np.byte, vlen)\n\n        gr.sync_block.__init__(\n\
+      \            self,\n            name='Split at tag',\n            in_sig=[np.byte],\n\
+      \            out_sig=[(np.byte, vlen)]\n        )\n\n        self.tag = tag\n\
+      \        self.vlen = vlen\n\n    def work(self, input_items, output_items):\n\
+      \        inp = input_items[0]\n\n        is_frame_start = lambda tag: pmt.to_python(tag.key)\
+      \ == self.tag\n        tags = filter(is_frame_start, self.get_tags_in_window(0,\
+      \ 0, len(inp)))\n\n        counter = self.nitems_written(0)\n        offsets\
+      \ = map(lambda t: t.offset - counter, tags)\n\n        print(list(offsets))\n\
+      \n        output_items[0][:] = inp\n        return len(output_items[0])\n"
     affinity: ''
     alias: ''
     comment: ''
     maxoutbuf: '0'
     minoutbuf: '0'
+    tag: '''frame_start'''
+    vlen: '20'
   states:
-    _io_cache: ('Printer', 'blk', [], [('0', 'byte', 1)], [], '', [])
+    _io_cache: ('Split at tag', 'blk', [('tag', "'frame_start'"), ('vlen', '1')],
+      [('0', 'byte', 1)], [('0', 'byte', 1)], '', ['tag', 'vlen'])
     bus_sink: false
     bus_source: false
     bus_structure: null
-    coordinate: [1072, 1264.0]
+    coordinate: [752, 1252.0]
     rotation: 0
     state: true
 - name: fir_filter_xxx_1
@@ -1349,9 +1379,9 @@ connections:
 - [blocks_complex_to_magphase_0_0, '1', blocks_null_sink_3, '0']
 - [blocks_multiply_const_vxx_0_0, '0', qtgui_time_sink_x_2_0, '0']
 - [blocks_null_source_0, '0', blocks_stream_mux_0, '0']
-- [blocks_repack_bits_bb_0, '0', blocks_tagged_stream_align_0, '0']
+- [blocks_repack_bits_bb_0, '0', blocks_tag_debug_0, '0']
+- [blocks_repack_bits_bb_0, '0', epy_block_2, '0']
 - [blocks_stream_mux_0, '0', digital_constellation_modulator_0, '0']
-- [blocks_tagged_stream_align_0, '0', epy_block_1, '0']
 - [blocks_throttle_0, '0', virtual_sink_0, '0']
 - [blocks_vector_source_x_0, '0', blocks_stream_mux_0, '1']
 - [channels_channel_model_0, '0', blocks_throttle_0, '0']
@@ -1370,6 +1400,7 @@ connections:
 - [digital_pfb_clock_sync_xxx_0, '0', digital_cma_equalizer_cc_0, '0']
 - [epy_block_0, '0', qtgui_const_sink_x_0_0, '0']
 - [epy_block_0, '0', virtual_sink_3, '0']
+- [epy_block_2, '0', epy_block_1, '0']
 - [fir_filter_xxx_1, '0', blocks_complex_to_magphase_0_0, '0']
 - [virtual_source_0, '0', digital_pfb_clock_sync_xxx_0, '0']
 - [virtual_source_1, '0', digital_constellation_decoder_cb_0, '0']
diff --git a/tests/correlator/correlator.py b/tests/correlator/correlator.py
index 3f3b8fc..c84e4d7 100755
--- a/tests/correlator/correlator.py
+++ b/tests/correlator/correlator.py
@@ -36,6 +36,7 @@ from gnuradio.eng_arg import eng_float, intx
 from gnuradio import eng_notation
 import epy_block_0
 import epy_block_1
+import epy_block_2
 import numpy as np
 
 from gnuradio import qtgui
@@ -329,7 +330,8 @@ class correlator(gr.top_block, Qt.QWidget):
             self.top_grid_layout.setRowStretch(r, 1)
         for c in range(1, 2):
             self.top_grid_layout.setColumnStretch(c, 1)
-        self.epy_block_1 = epy_block_1.blk()
+        self.epy_block_2 = epy_block_2.blk(tag='frame_start', vlen=20)
+        self.epy_block_1 = epy_block_1.blk(vlen=20)
         self.epy_block_0 = epy_block_0.blk()
         self.digital_pfb_clock_sync_xxx_0 = digital.pfb_clock_sync_ccf(sps, timing_loop_bw, rrc_taps, nfilts, 16, 1.5, 1)
         self.digital_costas_loop_cc_0 = digital.costas_loop_cc(2 * 3.141592653589793 / 100, 4, False)
@@ -353,7 +355,6 @@ class correlator(gr.top_block, Qt.QWidget):
             block_tags=False)
         self.blocks_vector_source_x_0 = blocks.vector_source_b(testvec * 500, True, 1, [])
         self.blocks_throttle_0 = blocks.throttle(gr.sizeof_gr_complex*1, samp_rate,True)
-        self.blocks_tagged_stream_align_0 = blocks.tagged_stream_align(gr.sizeof_char*1, 'frame_start')
         self.blocks_stream_mux_0 = blocks.stream_mux(gr.sizeof_char*1, [0, len(testvec)])
         self.blocks_repack_bits_bb_0 = blocks.repack_bits_bb(2, 8, "", False, gr.GR_MSB_FIRST)
         self.blocks_null_source_0 = blocks.null_source(gr.sizeof_char*1)
@@ -368,9 +369,8 @@ class correlator(gr.top_block, Qt.QWidget):
         self.connect((self.blocks_complex_to_magphase_0_0, 1), (self.blocks_null_sink_3, 0))
         self.connect((self.blocks_complex_to_magphase_0_0, 0), (self.qtgui_time_sink_x_0_0, 0))
         self.connect((self.blocks_null_source_0, 0), (self.blocks_stream_mux_0, 0))
-        self.connect((self.blocks_repack_bits_bb_0, 0), (self.blocks_tagged_stream_align_0, 0))
+        self.connect((self.blocks_repack_bits_bb_0, 0), (self.epy_block_2, 0))
         self.connect((self.blocks_stream_mux_0, 0), (self.digital_constellation_modulator_0, 0))
-        self.connect((self.blocks_tagged_stream_align_0, 0), (self.epy_block_1, 0))
         self.connect((self.blocks_throttle_0, 0), (self.digital_pfb_clock_sync_xxx_0, 0))
         self.connect((self.blocks_vector_source_x_0, 0), (self.blocks_stream_mux_0, 1))
         self.connect((self.channels_channel_model_0, 0), (self.blocks_throttle_0, 0))
@@ -387,6 +387,7 @@ class correlator(gr.top_block, Qt.QWidget):
         self.connect((self.digital_pfb_clock_sync_xxx_0, 0), (self.digital_cma_equalizer_cc_0, 0))
         self.connect((self.epy_block_0, 0), (self.digital_constellation_decoder_cb_0, 0))
         self.connect((self.epy_block_0, 0), (self.qtgui_const_sink_x_0_0, 0))
+        self.connect((self.epy_block_2, 0), (self.epy_block_1, 0))
 
 
     def closeEvent(self, event):
diff --git a/tests/correlator/epy_block_0.py b/tests/correlator/epy_block_0.py
index 7821b3e..e32d72b 100644
--- a/tests/correlator/epy_block_0.py
+++ b/tests/correlator/epy_block_0.py
@@ -3,6 +3,7 @@ import pmt
 import numpy as np
 from gnuradio import gr
 
+print = lambda x: None
 
 class blk(gr.sync_block):
     """
diff --git a/tests/correlator/epy_block_1.py b/tests/correlator/epy_block_1.py
index d30c2eb..7e722fc 100644
--- a/tests/correlator/epy_block_1.py
+++ b/tests/correlator/epy_block_1.py
@@ -1,22 +1,16 @@
-"""
-Embedded Python Blocks:
-
-Each time this file is saved, GRC will instantiate the first class it finds
-to get ports and parameters of your block. The arguments to __init__  will
-be the parameters. All of them are required to have default values!
-"""
-
 import numpy as np
 from gnuradio import gr
 
 np.set_printoptions(formatter={'int':hex})
 
 class blk(gr.sync_block):
-    def __init__(self):
+    def __init__(self, vlen=1):
+        dt = np.byte if vlen == 1 else (np.byte, vlen)
+
         gr.sync_block.__init__(
             self,
             name='Printer',
-            in_sig=[np.byte],
+            in_sig=[(np.byte, vlen)],
             out_sig=[]
         )
 
diff --git a/tests/correlator/epy_block_2.py b/tests/correlator/epy_block_2.py
new file mode 100644
index 0000000..210be12
--- /dev/null
+++ b/tests/correlator/epy_block_2.py
@@ -0,0 +1,33 @@
+import pmt
+import numpy as np
+from gnuradio import gr
+
+
+class blk(gr.sync_block):
+
+    def __init__(self, tag="frame_start", vlen=1):
+        dt = np.byte if vlen == 1 else (np.byte, vlen)
+
+        gr.sync_block.__init__(
+            self,
+            name='Split at tag',
+            in_sig=[np.byte],
+            out_sig=[(np.byte, vlen)]
+        )
+
+        self.tag = tag
+        self.vlen = vlen
+
+    def work(self, input_items, output_items):
+        inp = input_items[0]
+
+        is_frame_start = lambda tag: pmt.to_python(tag.key) == self.tag
+        tags = filter(is_frame_start, self.get_tags_in_window(0, 0, len(inp)))
+
+        counter = self.nitems_written(0)
+        offsets = map(lambda t: t.offset - counter, tags)
+
+        print(list(offsets))
+
+        output_items[0][:] = inp
+        return len(output_items[0])