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,2007 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.tb = gr.top_block ()
  29
  30     def tearDown (self):
  31         self.tb = 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.tb.connect (v0, (mux,0))
  42         self.tb.connect (v1, (mux,1))
  43         self.tb.connect (mux, dst)
  44         self.tb.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.tb.connect (v0, (mux,0))
  61         self.tb.connect (v1, (mux,1))
  62         self.tb.connect (mux, dst)
  63         self.tb.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     def test_stream_2NM_ff(self):
  98         N = 40
  99         stream_sizes = [7, 9]
 100         self.help_stream_2ff(N, stream_sizes)
 101
 102         result_data = self.help_stream_2ff(N, stream_sizes)
 103
 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,
 113                     2.0, 2.0, 2.0, 2.0)
 114
 115         self.assertEqual (exp_data, result_data)
 116
 117
 118     def test_stream_2MN_ff(self):
 119         N = 37
 120         stream_sizes = [7, 9]
 121         self.help_stream_2ff(N, stream_sizes)
 122
 123         result_data = self.help_stream_2ff(N, stream_sizes)
 124
 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,
 134                     2.0)
 135
 136         self.assertEqual (exp_data, result_data)
 137
 138     def test_stream_2N0_ff(self):
 139         N = 30
 140         stream_sizes = [7, 0]
 141         self.help_stream_2ff(N, stream_sizes)
 142
 143         result_data = self.help_stream_2ff(N, stream_sizes)
 144
 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,
 149                     1.0, 1.0)
 150
 151         self.assertEqual (exp_data, result_data)
 152
 153     def test_stream_20N_ff(self):
 154         N = 30
 155         stream_sizes = [0, 9]
 156         self.help_stream_2ff(N, stream_sizes)
 157
 158         result_data = self.help_stream_2ff(N, stream_sizes)
 159
 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,
 163                     2.0, 2.0, 2.0)
 164
 165         self.assertEqual (exp_data, result_data)
 166
 167 if __name__ == '__main__':
 168     gr_unittest.main()