git.gag.com Git - debian/gnuradio/blob - gnuradio-core/src/python/gnuradio/gr/qa_stream_mux.py

   1 #!/usr/bin/env python
   2 #
   3 # Copyright 2004,2005 Free Software Foundation, Inc.
   4 #
   5 # This file is part of GNU Radio
   6 #
   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)
  10 # any later version.
  11 #
  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.
  16 #
  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.
  21 #
  22
  23 from gnuradio import gr, gr_unittest
  24
  25 class test_head (gr_unittest.TestCase):
  26
  27     def setUp (self):
  28         self.fg = gr.flow_graph ()
  29
  30     def tearDown (self):
  31         self.fg = None
  32
  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)
  36
  37         mux = gr.stream_mux(gr.sizeof_float, stream_sizes)
  38
  39         dst = gr.vector_sink_f ()
  40
  41         self.fg.connect (v0, (mux,0))
  42         self.fg.connect (v1, (mux,1))
  43         self.fg.connect (mux, dst)
  44         self.fg.run ()
  45
  46         return dst.data ()
  47
  48     def help_stream_ramp_2ff(self, N, stream_sizes):
  49         r1 = range(N)
  50         r2 = range(N)
  51         r2.reverse()
  52
  53         v0 = gr.vector_source_f(r1, False)
  54         v1 = gr.vector_source_f(r2, False)
  55
  56         mux = gr.stream_mux(gr.sizeof_float, stream_sizes)
  57
  58         dst = gr.vector_sink_f ()
  59
  60         self.fg.connect (v0, (mux,0))
  61         self.fg.connect (v1, (mux,1))
  62         self.fg.connect (mux, dst)
  63         self.fg.run ()
  64
  65         return dst.data ()
  66
  67     def test_stream_2NN_ff(self):
  68         N = 40
  69         stream_sizes = [10, 10]
  70         result_data = self.help_stream_2ff(N, stream_sizes)
  71
  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)
  81
  82     def test_stream_ramp_2NN_ff(self):
  83         N = 40
  84         stream_sizes = [10, 10]
  85         result_data = self.help_stream_ramp_2ff(N, stream_sizes)
  86
  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)
  96
  97
  98     def test_stream_2NM_ff(self):
  99         N = 40
 100         stream_sizes = [7, 9]
 101         self.help_stream_2ff(N, stream_sizes)
 102
 103         result_data = self.help_stream_2ff(N, stream_sizes)
 104
 105         exp_data = (1.0, 1.0, 1.0, 1.0, 1.0, 1.0, 1.0,
 106                     2.0, 2.0, 2.0, 2.0, 2.0, 2.0, 2.0, 2.0, 2.0,
 107                     1.0, 1.0, 1.0, 1.0, 1.0, 1.0, 1.0,
 108                     2.0, 2.0, 2.0, 2.0, 2.0, 2.0, 2.0, 2.0, 2.0,
 109                     1.0, 1.0, 1.0, 1.0, 1.0, 1.0, 1.0,
 110                     2.0, 2.0, 2.0, 2.0, 2.0, 2.0, 2.0, 2.0, 2.0,
 111                     1.0, 1.0, 1.0, 1.0, 1.0, 1.0, 1.0,
 112                     2.0, 2.0, 2.0, 2.0, 2.0, 2.0, 2.0, 2.0, 2.0,
 113                     1.0, 1.0, 1.0, 1.0, 1.0, 1.0, 1.0,
 114                     2.0, 2.0, 2.0, 2.0)
 115
 116         self.assertEqual (exp_data, result_data)
 117
 118
 119     def test_stream_2MN_ff(self):
 120         N = 37
 121         stream_sizes = [7, 9]
 122         self.help_stream_2ff(N, stream_sizes)
 123
 124         result_data = self.help_stream_2ff(N, stream_sizes)
 125
 126         exp_data = (1.0, 1.0, 1.0, 1.0, 1.0, 1.0, 1.0,
 127                     2.0, 2.0, 2.0, 2.0, 2.0, 2.0, 2.0, 2.0, 2.0,
 128                     1.0, 1.0, 1.0, 1.0, 1.0, 1.0, 1.0,
 129                     2.0, 2.0, 2.0, 2.0, 2.0, 2.0, 2.0, 2.0, 2.0,
 130                     1.0, 1.0, 1.0, 1.0, 1.0, 1.0, 1.0,
 131                     2.0, 2.0, 2.0, 2.0, 2.0, 2.0, 2.0, 2.0, 2.0,
 132                     1.0, 1.0, 1.0, 1.0, 1.0, 1.0, 1.0,
 133                     2.0, 2.0, 2.0, 2.0, 2.0, 2.0, 2.0, 2.0, 2.0,
 134                     1.0, 1.0, 1.0, 1.0, 1.0, 1.0, 1.0,
 135                     2.0)
 136
 137         self.assertEqual (exp_data, result_data)
 138
 139     def test_stream_2N0_ff(self):
 140         N = 30
 141         stream_sizes = [7, 0]
 142         self.help_stream_2ff(N, stream_sizes)
 143
 144         result_data = self.help_stream_2ff(N, stream_sizes)
 145
 146         exp_data = (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,
 149                     1.0, 1.0, 1.0, 1.0, 1.0, 1.0, 1.0,
 150                     1.0, 1.0)
 151
 152         self.assertEqual (exp_data, result_data)
 153
 154     def test_stream_20N_ff(self):
 155         N = 30
 156         stream_sizes = [0, 9]
 157         self.help_stream_2ff(N, stream_sizes)
 158
 159         result_data = self.help_stream_2ff(N, stream_sizes)
 160
 161         exp_data = (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,
 163                     2.0, 2.0, 2.0, 2.0, 2.0, 2.0, 2.0, 2.0, 2.0,
 164                     2.0, 2.0, 2.0)
 165
 166         self.assertEqual (exp_data, result_data)
 167
 168
 169 if __name__ == '__main__':
 170     gr_unittest.main ()