diff options
| author | Andrej Rode <mail@andrejro.de> | 2016-04-26 11:31:48 +0200 |
|---|---|---|
| committer | Johnathan Corgan <johnathan@corganlabs.com> | 2016-04-27 16:43:21 -0700 |
| commit | d7773175342aba3f559fb69801d085737c25463b (patch) | |
| tree | 1ee96c1b41e00efd28584bb51ad24087c4a8990b /gr-blocks/python | |
| parent | a72adb370bb3ebb24354c7c60f9ec6dc9c653900 (diff) | |
gr-blocks: stream_mux add tag_propagation qa
Diffstat (limited to 'gr-blocks/python')
| -rwxr-xr-x | gr-blocks/python/blocks/qa_stream_mux.py | 60 |
1 files changed, 59 insertions, 1 deletions
diff --git a/gr-blocks/python/blocks/qa_stream_mux.py b/gr-blocks/python/blocks/qa_stream_mux.py index 00e32e955e..3b470afa4c 100755 --- a/gr-blocks/python/blocks/qa_stream_mux.py +++ b/gr-blocks/python/blocks/qa_stream_mux.py @@ -21,7 +21,7 @@ # from gnuradio import gr, gr_unittest, blocks - +import pmt import os class test_stream_mux (gr_unittest.TestCase): @@ -67,6 +67,36 @@ class test_stream_mux (gr_unittest.TestCase): return dst.data () + def help_stream_tag_propagation(self, N, stream_sizes): + src_data1 = stream_sizes[0]*N*[1,] + src_data2 = stream_sizes[1]*N*[2,] + src_data3 = stream_sizes[2]*N*[3,] + # stream_mux scheme (3,2,4) + src1 = blocks.vector_source_f(src_data1) + src2 = blocks.vector_source_f(src_data2) + src3 = blocks.vector_source_f(src_data3) + tag_stream1 = blocks.stream_to_tagged_stream(gr.sizeof_float, 1, + stream_sizes[0], 'src1') + tag_stream2 = blocks.stream_to_tagged_stream(gr.sizeof_float, 1, + stream_sizes[1], 'src2') + tag_stream3 = blocks.stream_to_tagged_stream(gr.sizeof_float, 1, + stream_sizes[2], 'src3') + + mux = blocks.stream_mux(gr.sizeof_float, stream_sizes) + dst = blocks.vector_sink_f() + + self.tb.connect(src1, tag_stream1) + self.tb.connect(src2, tag_stream2) + self.tb.connect(src3, tag_stream3) + self.tb.connect(tag_stream1, (mux,0)) + self.tb.connect(tag_stream2, (mux,1)) + self.tb.connect(tag_stream3, (mux,2)) + self.tb.connect(mux, dst) + self.tb.run() + + return (dst.data (), dst.tags ()) + + def test_stream_2NN_ff(self): N = 40 stream_sizes = [10, 10] @@ -181,5 +211,33 @@ class test_stream_mux (gr_unittest.TestCase): self.tb.run () self.assertEqual (r1 + r2, dst.data()) + def test_tag_propagation(self): + N = 10 # Block length + stream_sizes = [1,2,3] + + expected_result = N*(stream_sizes[0]*[1,] + +stream_sizes[1]*[2,] + +stream_sizes[2]*[3,]) + # check the data + (result, tags) = self.help_stream_tag_propagation(N, stream_sizes) + self.assertFloatTuplesAlmostEqual(expected_result, result, places=6) + + # check the tags + expected_tag_offsets_src1 = [sum(stream_sizes)*i for i in range(N)] + expected_tag_offsets_src2 = [stream_sizes[0] + +sum(stream_sizes)*i for i in range(N)] + expected_tag_offsets_src3 = [stream_sizes[0]+stream_sizes[1] + +sum(stream_sizes)*i for i in range(N)] + tags_src1 = [tag for tag in tags if pmt.eq(tag.key, pmt.intern('src1'))] + tags_src2 = [tag for tag in tags if pmt.eq(tag.key, pmt.intern('src2'))] + tags_src3 = [tag for tag in tags if pmt.eq(tag.key, pmt.intern('src3'))] + + for i in range(len(expected_tag_offsets_src1)): + self.assertTrue(expected_tag_offsets_src1[i] == tags_src1[i].offset) + for i in range(len(expected_tag_offsets_src2)): + self.assertTrue(expected_tag_offsets_src2[i] == tags_src2[i].offset) + for i in range(len(expected_tag_offsets_src3)): + self.assertTrue(expected_tag_offsets_src3[i] == tags_src3[i].offset) + if __name__ == '__main__': gr_unittest.run(test_stream_mux, "test_stream_mux.xml") |