3 # Copyright 2005,2007 Free Software Foundation, Inc.
5 # This file is part of GNU Radio
7 # GNU Radio is free software; you can redistribute it and/or modify
8 # it under the terms of the GNU General Public License as published by
9 # the Free Software Foundation; either version 3, or (at your option)
12 # GNU Radio is distributed in the hope that it will be useful,
13 # but WITHOUT ANY WARRANTY; without even the implied warranty of
14 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
15 # GNU General Public License for more details.
17 # You should have received a copy of the GNU General Public License
18 # along with GNU Radio; see the file COPYING. If not, write to
19 # the Free Software Foundation, Inc., 51 Franklin Street,
20 # Boston, MA 02110-1301, USA.
23 from gnuradio import gr, gr_unittest
27 print "pid =", os.getpid()
28 raw_input("Attach, then press Enter to continue")
31 def calc_expected_result(src_data, n):
32 assert (len(src_data) % n) == 0
33 result = [list() for x in range(n)]
34 #print "len(result) =", len(result)
35 for i in xrange(len(src_data)):
36 (result[i % n]).append(src_data[i])
37 return [tuple(x) for x in result]
40 class test_pipe_fittings(gr_unittest.TestCase):
43 self.tb = gr.top_block ()
50 Test stream_to_streams.
54 src_data = range(src_len)
56 expected_results = calc_expected_result(src_data, n)
57 #print "expected results: ", expected_results
58 src = gr.vector_source_i(src_data)
59 op = gr.stream_to_streams(gr.sizeof_int, n)
60 self.tb.connect(src, op)
64 dst = gr.vector_sink_i()
65 self.tb.connect((op, i), (dst, 0))
71 self.assertEqual(expected_results[d], dsts[d].data())
75 Test streams_to_stream (using stream_to_streams).
79 src_data = tuple(range(src_len))
80 expected_results = src_data
82 src = gr.vector_source_i(src_data)
83 op1 = gr.stream_to_streams(gr.sizeof_int, n)
84 op2 = gr.streams_to_stream(gr.sizeof_int, n)
85 dst = gr.vector_sink_i()
87 self.tb.connect(src, op1)
89 self.tb.connect((op1, i), (op2, i))
90 self.tb.connect(op2, dst)
93 self.assertEqual(expected_results, dst.data())
97 Test streams_to_vector (using stream_to_streams & vector_to_stream).
101 src_data = tuple(range(src_len))
102 expected_results = src_data
104 src = gr.vector_source_i(src_data)
105 op1 = gr.stream_to_streams(gr.sizeof_int, n)
106 op2 = gr.streams_to_vector(gr.sizeof_int, n)
107 op3 = gr.vector_to_stream(gr.sizeof_int, n)
108 dst = gr.vector_sink_i()
110 self.tb.connect(src, op1)
112 self.tb.connect((op1, i), (op2, i))
113 self.tb.connect(op2, op3, dst)
116 self.assertEqual(expected_results, dst.data())
120 Test vector_to_streams.
124 src_data = tuple(range(src_len))
125 expected_results = src_data
127 src = gr.vector_source_i(src_data)
128 op1 = gr.stream_to_vector(gr.sizeof_int, n)
129 op2 = gr.vector_to_streams(gr.sizeof_int, n)
130 op3 = gr.streams_to_stream(gr.sizeof_int, n)
131 dst = gr.vector_sink_i()
133 self.tb.connect(src, op1, op2)
135 self.tb.connect((op2, i), (op3, i))
136 self.tb.connect(op3, dst)
139 self.assertEqual(expected_results, dst.data())
141 if __name__ == '__main__':