3 # Copyright 2004,2005,2007,2010 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
25 class test_stream_mux (gr_unittest.TestCase):
28 self.tb = gr.top_block ()
33 def help_stream_2ff(self, N, stream_sizes):
34 v0 = gr.vector_source_f(N*[1,], False)
35 v1 = gr.vector_source_f(N*[2,], False)
37 mux = gr.stream_mux(gr.sizeof_float, stream_sizes)
39 dst = gr.vector_sink_f ()
41 self.tb.connect (v0, (mux,0))
42 self.tb.connect (v1, (mux,1))
43 self.tb.connect (mux, dst)
48 def help_stream_ramp_2ff(self, N, stream_sizes):
53 v0 = gr.vector_source_f(r1, False)
54 v1 = gr.vector_source_f(r2, False)
56 mux = gr.stream_mux(gr.sizeof_float, stream_sizes)
58 dst = gr.vector_sink_f ()
60 self.tb.connect (v0, (mux,0))
61 self.tb.connect (v1, (mux,1))
62 self.tb.connect (mux, dst)
67 def test_stream_2NN_ff(self):
69 stream_sizes = [10, 10]
70 result_data = self.help_stream_2ff(N, stream_sizes)
72 exp_data = (1.0, 1.0, 1.0, 1.0, 1.0, 1.0, 1.0, 1.0, 1.0, 1.0,
73 2.0, 2.0, 2.0, 2.0, 2.0, 2.0, 2.0, 2.0, 2.0, 2.0,
74 1.0, 1.0, 1.0, 1.0, 1.0, 1.0, 1.0, 1.0, 1.0, 1.0,
75 2.0, 2.0, 2.0, 2.0, 2.0, 2.0, 2.0, 2.0, 2.0, 2.0,
76 1.0, 1.0, 1.0, 1.0, 1.0, 1.0, 1.0, 1.0, 1.0, 1.0,
77 2.0, 2.0, 2.0, 2.0, 2.0, 2.0, 2.0, 2.0, 2.0, 2.0,
78 1.0, 1.0, 1.0, 1.0, 1.0, 1.0, 1.0, 1.0, 1.0, 1.0,
79 2.0, 2.0, 2.0, 2.0, 2.0, 2.0, 2.0, 2.0, 2.0, 2.0)
80 self.assertEqual (exp_data, result_data)
82 def test_stream_ramp_2NN_ff(self):
84 stream_sizes = [10, 10]
85 result_data = self.help_stream_ramp_2ff(N, stream_sizes)
87 exp_data = ( 0.0, 1.0, 2.0, 3.0, 4.0, 5.0, 6.0, 7.0, 8.0, 9.0,
88 39.0, 38.0, 37.0, 36.0, 35.0, 34.0, 33.0, 32.0, 31.0, 30.0,
89 10.0, 11.0, 12.0, 13.0, 14.0, 15.0, 16.0, 17.0, 18.0, 19.0,
90 29.0, 28.0, 27.0, 26.0, 25.0, 24.0, 23.0, 22.0, 21.0, 20.0,
91 20.0, 21.0, 22.0, 23.0, 24.0, 25.0, 26.0, 27.0, 28.0, 29.0,
92 19.0, 18.0, 17.0, 16.0, 15.0, 14.0, 13.0, 12.0, 11.0, 10.0,
93 30.0, 31.0, 32.0, 33.0, 34.0, 35.0, 36.0, 37.0, 38.0, 39.0,
94 9.0, 8.0, 7.0, 6.0, 5.0, 4.0, 3.0, 2.0, 1.0, 0.0)
95 self.assertEqual (exp_data, result_data)
97 def test_stream_2NM_ff(self):
100 self.help_stream_2ff(N, stream_sizes)
102 result_data = self.help_stream_2ff(N, stream_sizes)
104 exp_data = (1.0, 1.0, 1.0, 1.0, 1.0, 1.0, 1.0,
105 2.0, 2.0, 2.0, 2.0, 2.0, 2.0, 2.0, 2.0, 2.0,
106 1.0, 1.0, 1.0, 1.0, 1.0, 1.0, 1.0,
107 2.0, 2.0, 2.0, 2.0, 2.0, 2.0, 2.0, 2.0, 2.0,
108 1.0, 1.0, 1.0, 1.0, 1.0, 1.0, 1.0,
109 2.0, 2.0, 2.0, 2.0, 2.0, 2.0, 2.0, 2.0, 2.0,
110 1.0, 1.0, 1.0, 1.0, 1.0, 1.0, 1.0,
111 2.0, 2.0, 2.0, 2.0, 2.0, 2.0, 2.0, 2.0, 2.0,
112 1.0, 1.0, 1.0, 1.0, 1.0, 1.0, 1.0,
115 self.assertEqual (exp_data, result_data)
118 def test_stream_2MN_ff(self):
120 stream_sizes = [7, 9]
121 self.help_stream_2ff(N, stream_sizes)
123 result_data = self.help_stream_2ff(N, stream_sizes)
125 exp_data = (1.0, 1.0, 1.0, 1.0, 1.0, 1.0, 1.0,
126 2.0, 2.0, 2.0, 2.0, 2.0, 2.0, 2.0, 2.0, 2.0,
127 1.0, 1.0, 1.0, 1.0, 1.0, 1.0, 1.0,
128 2.0, 2.0, 2.0, 2.0, 2.0, 2.0, 2.0, 2.0, 2.0,
129 1.0, 1.0, 1.0, 1.0, 1.0, 1.0, 1.0,
130 2.0, 2.0, 2.0, 2.0, 2.0, 2.0, 2.0, 2.0, 2.0,
131 1.0, 1.0, 1.0, 1.0, 1.0, 1.0, 1.0,
132 2.0, 2.0, 2.0, 2.0, 2.0, 2.0, 2.0, 2.0, 2.0,
133 1.0, 1.0, 1.0, 1.0, 1.0, 1.0, 1.0,
136 self.assertEqual (exp_data, result_data)
138 def test_stream_2N0_ff(self):
140 stream_sizes = [7, 0]
141 self.help_stream_2ff(N, stream_sizes)
143 result_data = self.help_stream_2ff(N, stream_sizes)
145 exp_data = (1.0, 1.0, 1.0, 1.0, 1.0, 1.0, 1.0,
146 1.0, 1.0, 1.0, 1.0, 1.0, 1.0, 1.0,
147 1.0, 1.0, 1.0, 1.0, 1.0, 1.0, 1.0,
148 1.0, 1.0, 1.0, 1.0, 1.0, 1.0, 1.0,
151 self.assertEqual (exp_data, result_data)
153 def test_stream_20N_ff(self):
155 stream_sizes = [0, 9]
156 self.help_stream_2ff(N, stream_sizes)
158 result_data = self.help_stream_2ff(N, stream_sizes)
160 exp_data = (2.0, 2.0, 2.0, 2.0, 2.0, 2.0, 2.0, 2.0, 2.0,
161 2.0, 2.0, 2.0, 2.0, 2.0, 2.0, 2.0, 2.0, 2.0,
162 2.0, 2.0, 2.0, 2.0, 2.0, 2.0, 2.0, 2.0, 2.0,
165 self.assertEqual (exp_data, result_data)
167 if __name__ == '__main__':
168 gr_unittest.run(test_stream_mux, "test_stream_mux.xml")