diff options
| author | Tom Rondeau <trondeau@vt.edu> | 2013-03-17 12:24:38 -0400 |
|---|---|---|
| committer | Tom Rondeau <trondeau@vt.edu> | 2013-03-17 15:40:32 -0400 |
| commit | d4f6b86a9bdea09c2c158b9982559a727f8c6a0b (patch) | |
| tree | 60b452d387bd494b461ee78b2b0faec1fbf6e5fe /gnuradio-core/src/python/gnuradio | |
| parent | ee2b700f72503d6e7f62adbfb7dff9997b9cb003 (diff) | |
blocks: converting references to vector source/sink, null source/sink, nop, copy, head, skiphead, vector_map, and annotator blocks to use gr-blocks.
Diffstat (limited to 'gnuradio-core/src/python/gnuradio')
14 files changed, 7 insertions, 1696 deletions
diff --git a/gnuradio-core/src/python/gnuradio/ctrlport/GrDataPlotter.py b/gnuradio-core/src/python/gnuradio/ctrlport/GrDataPlotter.py index dfa67feffe..8597ca6497 100644 --- a/gnuradio-core/src/python/gnuradio/ctrlport/GrDataPlotter.py +++ b/gnuradio-core/src/python/gnuradio/ctrlport/GrDataPlotter.py @@ -206,7 +206,7 @@ class GrDataPlotterC(GrDataPlotParent): return snk def get_vecsource(self): - return gr.vector_source_c([]) + return blocks.vector_source_c([]) def get_npts(self): self._npts = self.snk.nsamps() @@ -237,7 +237,7 @@ class GrDataPlotterF(GrDataPlotParent): return snk def get_vecsource(self): - return gr.vector_source_f([]) + return blocks.vector_source_f([]) def get_npts(self): self._npts = self.snk.nsamps() @@ -265,7 +265,7 @@ class GrDataPlotterConst(GrDataPlotParent): return snk def get_vecsource(self): - return gr.vector_source_c([]) + return blocks.vector_source_c([]) def get_npts(self): self._npts = self.snk.nsamps() @@ -304,7 +304,7 @@ class GrDataPlotterPsdC(GrDataPlotParent): return snk def get_vecsource(self): - return gr.vector_source_c([]) + return blocks.vector_source_c([]) def get_npts(self): self._npts = self.snk.fft_size() @@ -337,7 +337,7 @@ class GrDataPlotterPsdF(GrDataPlotParent): return snk def get_vecsource(self): - return gr.vector_source_f([]) + return blocks.vector_source_f([]) def get_npts(self): self._npts = self.snk.fft_size() @@ -367,7 +367,7 @@ class GrTimeRasterF(GrDataPlotParent): return snk def get_vecsource(self): - return gr.vector_source_f([]) + return blocks.vector_source_f([]) def get_npts(self): self._npts = self.snk.num_cols() @@ -396,7 +396,7 @@ class GrTimeRasterB(GrDataPlotParent): return snk def get_vecsource(self): - return gr.vector_source_b([]) + return blocks.vector_source_b([]) def get_npts(self): self._npts = self.snk.num_cols() diff --git a/gnuradio-core/src/python/gnuradio/ctrlport/qa_cpp_py_binding.py b/gnuradio-core/src/python/gnuradio/ctrlport/qa_cpp_py_binding.py deleted file mode 100755 index 6de5d5fc9f..0000000000 --- a/gnuradio-core/src/python/gnuradio/ctrlport/qa_cpp_py_binding.py +++ /dev/null @@ -1,172 +0,0 @@ -#!/usr/bin/env python -# -# Copyright 2012 Free Software Foundation, Inc. -# -# This file is part of GNU Radio -# -# GNU Radio is free software; you can redistribute it and/or modify -# it under the terms of the GNU General Public License as published by -# the Free Software Foundation; either version 3, or (at your option) -# any later version. -# -# GNU Radio is distributed in the hope that it will be useful, -# but WITHOUT ANY WARRANTY; without even the implied warranty of -# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the -# GNU General Public License for more details. -# -# You should have received a copy of the GNU General Public License -# along with GNU Radio; see the file COPYING. If not, write to -# the Free Software Foundation, Inc., 51 Franklin Street, -# Boston, MA 02110-1301, USA. -# - -# -# This program tests mixed python and c++ ctrlport exports in a single app -# - -import Ice -import sys, time, random, numpy -from gnuradio import gr, gr_unittest - -from gnuradio.ctrlport import GNURadio -from gnuradio import ctrlport -import os - -def get1(): - return "success" - -def get2(): - return "failure" - -class inc_class: - def __init__(self): - self.val = 1 - def pp(self): - self.val = self.val+1 - return self.val - -get3 = inc_class() - -def get4(): - random.seed(0) - rv = random.random() - return rv - -def get5(): - numpy.random.seed(0) - samp_t = numpy.random.randn(24)+1j*numpy.random.randn(24); - samp_f = numpy.fft.fft(samp_t); - log_pow_f = 20*numpy.log10(numpy.abs(samp_f)) - rv = list(log_pow_f) - return rv; - -def get6(): - numpy.random.seed(0) - samp_t = numpy.random.randn(1024)+1j*numpy.random.randn(1024); - rv = list(samp_t) - return rv; - -class test_cpp_py_binding(gr_unittest.TestCase): - - def setUp(self): - self.tb = gr.top_block() - os.environ['GR_CONF_CONTROLPORT_ON'] = 'True' - - def tearDown(self): - self.tb = None - - def test_001(self): - v1 = gr.RPC_get_string("pyland", "v1", "unit_1_string", - "Python Exported String", "", "", "", - gr.DISPNULL) - v1.activate(get1) - - v2 = gr.RPC_get_string("pyland", "v2", "unit_2_string", - "Python Exported String", "", "", "", - gr.DISPNULL) - v2.activate(get2) - - v3 = gr.RPC_get_int("pyland", "v3", "unit_3_int", - "Python Exported Int", 0, 100, 1, - gr.DISPNULL) - v3.activate(get3.pp) - - v4 = gr.RPC_get_double("pyland", "time", "unit_4_time_double", - "Python Exported Double", 0, 1000, 1, - gr.DISPNULL) - v4.activate(get4) - - v5 = gr.RPC_get_vector_float("pyland", "fvec", "unit_5_float_vector", - "Python Exported Float Vector", [], [], [], - gr.DISPTIME | gr.DISPOPTCPLX) - v5.activate(get5) - - v6 = gr.RPC_get_vector_gr_complex("pyland", "cvec", "unit_6_gr_complex_vector", - "Python Exported Complex Vector", [], [], [], - gr.DISPXY | gr.DISPOPTSCATTER) - v6.activate(get6) - - # print some variables locally - val = get1() - rval = v1.get() - self.assertEqual(val, rval) - - val = get2() - rval = v2.get() - self.assertEqual(val, rval) - - val = get3.pp() - rval = v3.get() - self.assertEqual(val+1, rval) - - val = get4() - rval = v4.get() - self.assertEqual(val, rval) - - val = get5() - rval = v5.get() - self.assertComplexTuplesAlmostEqual(val, rval, 5) - - val = get6() - rval = v6.get() - self.assertComplexTuplesAlmostEqual(val, rval, 5) - - def test_002(self): - data = range(1,9) - - self.src = gr.vector_source_c(data) - self.p1 = gr.ctrlport_probe_c("aaa","C++ exported variable") - self.p2 = gr.ctrlport_probe_c("bbb","C++ exported variable") - probe_name = self.p2.alias() - - self.tb.connect(self.src, self.p1) - self.tb.connect(self.src, self.p2) - self.tb.start() - - # Probes return complex values as list of floats with re, im - # Imaginary parts of this data set are 0. - expected_result = [1, 0, 2, 0, 3, 0, 4, 0, - 5, 0, 6, 0, 7, 0, 8, 0] - - # Make sure we have time for flowgraph to run - time.sleep(0.1) - - # Get available endpoint - ep = gr.rpcmanager_get().endpoints()[0] - - # Initialize a simple Ice client from endpoint - ic = Ice.initialize(sys.argv) - base = ic.stringToProxy(ep) - radio = GNURadio.ControlPortPrx.checkedCast(base) - - # Get all exported knobs - ret = radio.get([probe_name + "::bbb"]) - for name in ret.keys(): - result = ret[name].value - self.assertEqual(result, expected_result) - - self.tb.stop() - -if __name__ == '__main__': - gr_unittest.run(test_cpp_py_binding, "test_cpp_py_binding.xml") - diff --git a/gnuradio-core/src/python/gnuradio/ctrlport/qa_cpp_py_binding_set.py b/gnuradio-core/src/python/gnuradio/ctrlport/qa_cpp_py_binding_set.py deleted file mode 100755 index 4ca120099c..0000000000 --- a/gnuradio-core/src/python/gnuradio/ctrlport/qa_cpp_py_binding_set.py +++ /dev/null @@ -1,150 +0,0 @@ -#!/usr/bin/env python -# -# Copyright 2012 Free Software Foundation, Inc. -# -# This file is part of GNU Radio -# -# GNU Radio is free software; you can redistribute it and/or modify -# it under the terms of the GNU General Public License as published by -# the Free Software Foundation; either version 3, or (at your option) -# any later version. -# -# GNU Radio is distributed in the hope that it will be useful, -# but WITHOUT ANY WARRANTY; without even the implied warranty of -# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the -# GNU General Public License for more details. -# -# You should have received a copy of the GNU General Public License -# along with GNU Radio; see the file COPYING. If not, write to -# the Free Software Foundation, Inc., 51 Franklin Street, -# Boston, MA 02110-1301, USA. -# - -# -# This program tests mixed python and c++ GRCP sets in a single app -# - -import Ice -import sys, time, random, numpy -from gnuradio import gr, gr_unittest - -from gnuradio.ctrlport import GNURadio -from gnuradio import ctrlport -import os - -class inc_class: - def __init__(self,val): - self.val = val; - - def _get(self): - #print "returning get (val = %s)"%(str(self.val)); - return self.val; - - def _set(self,val): - #print "updating val to %s"%(str(val)); - self.val = val; - return; - -getset1 = inc_class(10); -getset2 = inc_class(100.0); -getset3 = inc_class("test"); - -class test_cpp_py_binding_set(gr_unittest.TestCase): - def setUp(self): - self.tb = gr.top_block() - os.environ['GR_CONF_CONTROLPORT_ON'] = 'True' - - def tearDown(self): - self.tb = None - - def test_001(self): - - g1 = gr.RPC_get_int("pyland", "v1", "unit_1_int", - "Python Exported Int", 0, 100, 10, - gr.DISPNULL) - g1.activate(getset1._get) - s1 = gr.RPC_get_int("pyland", "v1", "unit_1_int", - "Python Exported Int", 0, 100, 10, - gr.DISPNULL) - s1.activate(getset1._set) - time.sleep(0.01) - - # test int variables - getset1._set(21) - val = getset1._get() - rval = g1.get() - self.assertEqual(val, rval) - - g2 = gr.RPC_get_float("pyland", "v2", "unit_2_float", - "Python Exported Float", -100, 1000.0, 100.0, - gr.DISPNULL) - g2.activate(getset2._get) - s2 = gr.RPC_get_float("pyland", "v2", "unit_2_float", - "Python Exported Float", -100, 1000.0, 100.0, - gr.DISPNULL) - s2.activate(getset2._set) - time.sleep(0.01) - - # test float variables - getset2._set(123.456) - val = getset2._get() - rval = g2.get() - self.assertAlmostEqual(val, rval, 4) - - g3 = gr.RPC_get_string("pyland", "v3", "unit_3_string", - "Python Exported String", "", "", "", - gr.DISPNULL) - g3.activate(getset3._get) - s3 = gr.RPC_get_string("pyland", "v3", "unit_3_string", - "Python Exported String", "", "", "", - gr.DISPNULL) - s3.activate(getset3._set) - time.sleep(0.01) - - # test string variables - getset3._set("third test") - val = getset3._get() - rval = g3.get() - self.assertEqual(val, rval) - - - def test_002(self): - data = range(1, 10) - - self.src = gr.vector_source_c(data, True) - self.p = gr.nop(gr.sizeof_gr_complex) - self.p.set_ctrlport_test(0); - probe_info = self.p.alias() - - self.tb.connect(self.src, self.p) - - # Get available endpoint - ep = gr.rpcmanager_get().endpoints()[0] - - # Initialize a simple Ice client from endpoint - ic = Ice.initialize(sys.argv) - base = ic.stringToProxy(ep) - radio = GNURadio.ControlPortPrx.checkedCast(base) - - self.tb.start() - - # Make sure we have time for flowgraph to run - time.sleep(0.1) - - # Get all exported knobs - key_name_test = probe_info+"::test" - ret = radio.get([key_name_test,]) - - ret[key_name_test].value = 10 - radio.set({key_name_test: ret[key_name_test]}) - - ret = radio.get([]) - result_test = ret[key_name_test].value - self.assertEqual(result_test, 10) - - self.tb.stop() - self.tb.wait() - -if __name__ == '__main__': - gr_unittest.run(test_cpp_py_binding_set, "test_cpp_py_binding_set.xml") - diff --git a/gnuradio-core/src/python/gnuradio/gr/qa_block_gateway.py b/gnuradio-core/src/python/gnuradio/gr/qa_block_gateway.py deleted file mode 100644 index 2c8cba5439..0000000000 --- a/gnuradio-core/src/python/gnuradio/gr/qa_block_gateway.py +++ /dev/null @@ -1,255 +0,0 @@ -# -# Copyright 2011-2012 Free Software Foundation, Inc. -# -# This file is part of GNU Radio -# -# GNU Radio is free software; you can redistribute it and/or modify -# it under the terms of the GNU General Public License as published by -# the Free Software Foundation; either version 3, or (at your option) -# any later version. -# -# GNU Radio is distributed in the hope that it will be useful, -# but WITHOUT ANY WARRANTY; without even the implied warranty of -# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the -# GNU General Public License for more details. -# -# You should have received a copy of the GNU General Public License -# along with GNU Radio; see the file COPYING. If not, write to -# the Free Software Foundation, Inc., 51 Franklin Street, -# Boston, MA 02110-1301, USA. -# - -from gnuradio import gr, gr_unittest -import pmt -import numpy - -class add_2_f32_1_f32(gr.sync_block): - def __init__(self): - gr.sync_block.__init__( - self, - name = "add 2 f32", - in_sig = [numpy.float32, numpy.float32], - out_sig = [numpy.float32], - ) - - def work(self, input_items, output_items): - output_items[0][:] = input_items[0] + input_items[1] - return len(output_items[0]) - -class add_2_fc32_1_fc32(gr.sync_block): - def __init__(self): - gr.sync_block.__init__( - self, - name = "add 2 fc32", - in_sig = [numpy.complex64, numpy.complex64], - out_sig = [numpy.complex64], - ) - - def work(self, input_items, output_items): - output_items[0][:] = input_items[0] + input_items[1] - return len(output_items[0]) - -class convolve(gr.sync_block): - """ - A demonstration using block history to properly perform a convolution. - """ - def __init__(self): - gr.sync_block.__init__( - self, - name = "convolve", - in_sig = [numpy.float32], - out_sig = [numpy.float32] - ) - self._taps = [1, 0, 0, 0] - self.set_history(len(self._taps)) - - def work(self, input_items, output_items): - output_items[0][:] = numpy.convolve(input_items[0], self._taps, mode='valid') - return len(output_items[0]) - -class decim2x(gr.decim_block): - def __init__(self): - gr.decim_block.__init__( - self, - name = "decim2x", - in_sig = [numpy.float32], - out_sig = [numpy.float32], - decim = 2 - ) - - def work(self, input_items, output_items): - output_items[0][:] = input_items[0][::2] - return len(output_items[0]) - -class interp2x(gr.interp_block): - def __init__(self): - gr.interp_block.__init__( - self, - name = "interp2x", - in_sig = [numpy.float32], - out_sig = [numpy.float32], - interp = 2 - ) - - def work(self, input_items, output_items): - output_items[0][1::2] = input_items[0] - output_items[0][::2] = input_items[0] - return len(output_items[0]) - -class tag_source(gr.sync_block): - def __init__(self): - gr.sync_block.__init__( - self, - name = "tag source", - in_sig = None, - out_sig = [numpy.float32], - ) - - def work(self, input_items, output_items): - num_output_items = len(output_items[0]) - - #put code here to fill the output items... - - #make a new tag on the middle element every time work is called - count = self.nitems_written(0) + num_output_items/2 - key = pmt.string_to_symbol("example_key") - value = pmt.string_to_symbol("example_value") - self.add_item_tag(0, count, key, value) - - return num_output_items - -class tag_sink(gr.sync_block): - def __init__(self): - gr.sync_block.__init__( - self, - name = "tag sink", - in_sig = [numpy.float32], - out_sig = None, - ) - self.key = None - - def work(self, input_items, output_items): - num_input_items = len(input_items[0]) - - #put code here to process the input items... - - #print all the tags received in this work call - nread = self.nitems_read(0) - tags = self.get_tags_in_range(0, nread, nread+num_input_items) - for tag in tags: - #print tag.offset - #print pmt.symbol_to_string(tag.key) - #print pmt.symbol_to_string(tag.value) - self.key = pmt.symbol_to_string(tag.key) - - return num_input_items - -class fc32_to_f32_2(gr.sync_block): - def __init__(self): - gr.sync_block.__init__( - self, - name = "fc32_to_f32_2", - in_sig = [numpy.complex64], - out_sig = [(numpy.float32, 2)], - ) - - def work(self, input_items, output_items): - output_items[0][::,0] = numpy.real(input_items[0]) - output_items[0][::,1] = numpy.imag(input_items[0]) - return len(output_items[0]) - -class vector_to_stream(gr.interp_block): - def __init__(self, itemsize, nitems_per_block): - gr.interp_block.__init__( - self, - name = "vector_to_stream", - in_sig = [(itemsize, nitems_per_block)], - out_sig = [itemsize], - interp = nitems_per_block - ) - self.block_size = nitems_per_block - - def work(self, input_items, output_items): - n = 0 - for i in xrange(len(input_items[0])): - for j in xrange(self.block_size): - output_items[0][n] = input_items[0][i][j] - n += 1 - - return len(output_items[0]) - -class test_block_gateway(gr_unittest.TestCase): - - def test_add_f32(self): - tb = gr.top_block() - src0 = gr.vector_source_f([1, 3, 5, 7, 9], False) - src1 = gr.vector_source_f([0, 2, 4, 6, 8], False) - adder = add_2_f32_1_f32() - sink = gr.vector_sink_f() - tb.connect((src0, 0), (adder, 0)) - tb.connect((src1, 0), (adder, 1)) - tb.connect(adder, sink) - tb.run() - self.assertEqual(sink.data(), (1, 5, 9, 13, 17)) - - def test_add_fc32(self): - tb = gr.top_block() - src0 = gr.vector_source_c([1, 3j, 5, 7j, 9], False) - src1 = gr.vector_source_c([0, 2j, 4, 6j, 8], False) - adder = add_2_fc32_1_fc32() - sink = gr.vector_sink_c() - tb.connect((src0, 0), (adder, 0)) - tb.connect((src1, 0), (adder, 1)) - tb.connect(adder, sink) - tb.run() - self.assertEqual(sink.data(), (1, 5j, 9, 13j, 17)) - - def test_convolve(self): - tb = gr.top_block() - src = gr.vector_source_f([1, 2, 3, 4, 5, 6, 7, 8], False) - cv = convolve() - sink = gr.vector_sink_f() - tb.connect(src, cv, sink) - tb.run() - self.assertEqual(sink.data(), (1, 2, 3, 4, 5, 6, 7, 8)) - - def test_decim2x(self): - tb = gr.top_block() - src = gr.vector_source_f([1, 2, 3, 4, 5, 6, 7, 8], False) - d2x = decim2x() - sink = gr.vector_sink_f() - tb.connect(src, d2x, sink) - tb.run() - self.assertEqual(sink.data(), (1, 3, 5, 7)) - - def test_interp2x(self): - tb = gr.top_block() - src = gr.vector_source_f([1, 3, 5, 7, 9], False) - i2x = interp2x() - sink = gr.vector_sink_f() - tb.connect(src, i2x, sink) - tb.run() - self.assertEqual(sink.data(), (1, 1, 3, 3, 5, 5, 7, 7, 9, 9)) - - def test_tags(self): - src = tag_source() - sink = tag_sink() - head = gr.head(gr.sizeof_float, 50000) #should be enough items to get a tag through - tb = gr.top_block() - tb.connect(src, head, sink) - tb.run() - self.assertEqual(sink.key, "example_key") - - def test_fc32_to_f32_2(self): - tb = gr.top_block() - src = gr.vector_source_c([1+2j, 3+4j, 5+6j, 7+8j, 9+10j], False) - convert = fc32_to_f32_2() - v2s = vector_to_stream(numpy.float32, 2) - sink = gr.vector_sink_f() - tb.connect(src, convert, v2s, sink) - tb.run() - self.assertEqual(sink.data(), (1, 2, 3, 4, 5, 6, 7, 8, 9, 10)) - -if __name__ == '__main__': - gr_unittest.run(test_block_gateway, "test_block_gateway.xml") - diff --git a/gnuradio-core/src/python/gnuradio/gr/qa_copy.py b/gnuradio-core/src/python/gnuradio/gr/qa_copy.py deleted file mode 100755 index 68c8e451f3..0000000000 --- a/gnuradio-core/src/python/gnuradio/gr/qa_copy.py +++ /dev/null @@ -1,58 +0,0 @@ -#!/usr/bin/env python -# -# Copyright 2009,2010 Free Software Foundation, Inc. -# -# This file is part of GNU Radio -# -# GNU Radio is free software; you can redistribute it and/or modify -# it under the terms of the GNU General Public License as published by -# the Free Software Foundation; either version 3, or (at your option) -# any later version. -# -# GNU Radio is distributed in the hope that it will be useful, -# but WITHOUT ANY WARRANTY; without even the implied warranty of -# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the -# GNU General Public License for more details. -# -# You should have received a copy of the GNU General Public License -# along with GNU Radio; see the file COPYING. If not, write to -# the Free Software Foundation, Inc., 51 Franklin Street, -# Boston, MA 02110-1301, USA. -# - -from gnuradio import gr, gr_unittest - -class test_copy(gr_unittest.TestCase): - - def setUp (self): - self.tb = gr.top_block () - - def tearDown (self): - self.tb = None - - def test_copy (self): - src_data = (1, 2, 3, 4, 5, 6, 7, 8, 9, 10) - expected_result = (1, 2, 3, 4, 5, 6, 7, 8, 9, 10) - src = gr.vector_source_b(src_data) - op = gr.copy(gr.sizeof_char) - dst = gr.vector_sink_b() - self.tb.connect(src, op, dst) - self.tb.run() - dst_data = dst.data() - self.assertEqual(expected_result, dst_data) - - def test_copy_drop (self): - src_data = (1, 2, 3, 4, 5, 6, 7, 8, 9, 10) - expected_result = () - src = gr.vector_source_b(src_data) - op = gr.copy(gr.sizeof_char) - op.set_enabled(False) - dst = gr.vector_sink_b() - self.tb.connect(src, op, dst) - self.tb.run() - dst_data = dst.data() - self.assertEqual(expected_result, dst_data) - - -if __name__ == '__main__': - gr_unittest.run(test_copy, "test_copy.xml") diff --git a/gnuradio-core/src/python/gnuradio/gr/qa_endian_swap.py b/gnuradio-core/src/python/gnuradio/gr/qa_endian_swap.py deleted file mode 100644 index 4d2555cc4d..0000000000 --- a/gnuradio-core/src/python/gnuradio/gr/qa_endian_swap.py +++ /dev/null @@ -1,66 +0,0 @@ -#!/usr/bin/env python -# -# Copyright 2011,2012 Free Software Foundation, Inc. -# -# This file is part of GNU Radio -# -# GNU Radio is free software; you can redistribute it and/or modify -# it under the terms of the GNU General Public License as published by -# the Free Software Foundation; either version 3, or (at your option) -# any later version. -# -# GNU Radio is distributed in the hope that it will be useful, -# but WITHOUT ANY WARRANTY; without even the implied warranty of -# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the -# GNU General Public License for more details. -# -# You should have received a copy of the GNU General Public License -# along with GNU Radio; see the file COPYING. If not, write to -# the Free Software Foundation, Inc., 51 Franklin Street, -# Boston, MA 02110-1301, USA. -# - -from gnuradio import gr, gr_unittest -import ctypes - -class test_endian_swap (gr_unittest.TestCase): - - def setUp (self): - self.tb = gr.top_block () - - def tearDown (self): - self.tb = None - - def test_001(self): - - src_data = [1,2,3,4] - expected_result = [256, 512, 768, 1024]; - - src = gr.vector_source_s(src_data) - op = gr.endian_swap(2) - dst = gr.vector_sink_s() - - self.tb.connect(src, op, dst) - self.tb.run() - result_data = list(dst.data()) - - self.assertEqual(expected_result, result_data) - - def test_002(self): - - src_data = [1,2,3,4] - expected_result = [16777216, 33554432, 50331648, 67108864]; - - src = gr.vector_source_i(src_data) - op = gr.endian_swap(4) - dst = gr.vector_sink_i() - - self.tb.connect(src, op, dst) - self.tb.run() - result_data = list(dst.data()) - - self.assertEqual(expected_result, result_data) - -if __name__ == '__main__': - gr_unittest.run(test_endian_swap, "test_endian_swap.xml") - diff --git a/gnuradio-core/src/python/gnuradio/gr/qa_head.py b/gnuradio-core/src/python/gnuradio/gr/qa_head.py deleted file mode 100755 index d7cb354dcb..0000000000 --- a/gnuradio-core/src/python/gnuradio/gr/qa_head.py +++ /dev/null @@ -1,47 +0,0 @@ -#!/usr/bin/env python -# -# Copyright 2004,2007,2010 Free Software Foundation, Inc. -# -# This file is part of GNU Radio -# -# GNU Radio is free software; you can redistribute it and/or modify -# it under the terms of the GNU General Public License as published by -# the Free Software Foundation; either version 3, or (at your option) -# any later version. -# -# GNU Radio is distributed in the hope that it will be useful, -# but WITHOUT ANY WARRANTY; without even the implied warranty of -# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the -# GNU General Public License for more details. -# -# You should have received a copy of the GNU General Public License -# along with GNU Radio; see the file COPYING. If not, write to -# the Free Software Foundation, Inc., 51 Franklin Street, -# Boston, MA 02110-1301, USA. -# - -from gnuradio import gr, gr_unittest - -class test_head (gr_unittest.TestCase): - - def setUp (self): - self.tb = gr.top_block () - - def tearDown (self): - self.tb = None - - def test_head (self): - src_data = (1, 2, 3, 4, 5, 6, 7, 8, 9, 10) - expected_result = (1, 2, 3, 4) - src1 = gr.vector_source_i (src_data) - op = gr.head (gr.sizeof_int, 4) - dst1 = gr.vector_sink_i () - self.tb.connect (src1, op) - self.tb.connect (op, dst1) - self.tb.run () - dst_data = dst1.data () - self.assertEqual (expected_result, dst_data) - - -if __name__ == '__main__': - gr_unittest.run(test_head, "test_head.xml") diff --git a/gnuradio-core/src/python/gnuradio/gr/qa_hier_block2.py b/gnuradio-core/src/python/gnuradio/gr/qa_hier_block2.py deleted file mode 100755 index 9ba48599af..0000000000 --- a/gnuradio-core/src/python/gnuradio/gr/qa_hier_block2.py +++ /dev/null @@ -1,397 +0,0 @@ -#!/usr/bin/env python - -from gnuradio import gr, gr_unittest -import numpy - -class add_ff(gr.sync_block): - def __init__(self): - gr.sync_block.__init__( - self, - name = "add_ff", - in_sig = [numpy.float32, numpy.float32], - out_sig = [numpy.float32], - ) - - def work(self, input_items, output_items): - output_items[0][:] = input_items[0] + input_items[1] - return len(output_items[0]) - -class multiply_const_ff(gr.sync_block): - def __init__(self, k): - gr.sync_block.__init__( - self, - name = "multiply_ff", - in_sig = [numpy.float32], - out_sig = [numpy.float32], - ) - self.k = k - - def work(self, input_items, output_items): - output_items[0][:] = map(lambda x: self.k*x, input_items[0]) - return len(output_items[0]) - -class test_hier_block2(gr_unittest.TestCase): - - def setUp(self): - pass - - def tearDown(self): - pass - - def test_001_make(self): - hblock = gr.hier_block2("test_block", - gr.io_signature(1,1,gr.sizeof_int), - gr.io_signature(1,1,gr.sizeof_int)) - self.assertEqual("test_block", hblock.name()) - self.assertEqual(1, hblock.input_signature().max_streams()) - self.assertEqual(1, hblock.output_signature().min_streams()) - self.assertEqual(1, hblock.output_signature().max_streams()) - self.assertEqual(gr.sizeof_int, hblock.output_signature().sizeof_stream_item(0)) - - def test_002_connect_input(self): - hblock = gr.hier_block2("test_block", - gr.io_signature(1,1,gr.sizeof_int), - gr.io_signature(1,1,gr.sizeof_int)) - nop1 = gr.nop(gr.sizeof_int) - hblock.connect(hblock, nop1) - - def test_004_connect_output(self): - hblock = gr.hier_block2("test_block", - gr.io_signature(1,1,gr.sizeof_int), - gr.io_signature(1,1,gr.sizeof_int)) - nop1 = gr.nop(gr.sizeof_int) - hblock.connect(nop1, hblock) - - def test_005_connect_output_in_use(self): - hblock = gr.hier_block2("test_block", - gr.io_signature(1,1,gr.sizeof_int), - gr.io_signature(1,1,gr.sizeof_int)) - nop1 = gr.nop(gr.sizeof_int) - nop2 = gr.nop(gr.sizeof_int) - hblock.connect(nop1, hblock) - self.assertRaises(ValueError, - lambda: hblock.connect(nop2, hblock)) - - def test_006_connect_invalid_src_port_neg(self): - hblock = gr.hier_block2("test_block", - gr.io_signature(1,1,gr.sizeof_int), - gr.io_signature(1,1,gr.sizeof_int)) - nop1 = gr.nop(gr.sizeof_int) - self.assertRaises(ValueError, - lambda: hblock.connect((hblock, -1), nop1)) - - def test_005_connect_invalid_src_port_exceeds(self): - hblock = gr.hier_block2("test_block", - gr.io_signature(1,1,gr.sizeof_int), - gr.io_signature(1,1,gr.sizeof_int)) - nop1 = gr.nop(gr.sizeof_int) - self.assertRaises(ValueError, - lambda: hblock.connect((hblock, 1), nop1)) - - def test_007_connect_invalid_dst_port_neg(self): - hblock = gr.hier_block2("test_block", - gr.io_signature(1,1,gr.sizeof_int), - gr.io_signature(1,1,gr.sizeof_int)) - nop1 = gr.nop(gr.sizeof_int) - nop2 = gr.nop(gr.sizeof_int) - self.assertRaises(ValueError, - lambda: hblock.connect(nop1, (nop2, -1))) - - def test_008_connect_invalid_dst_port_exceeds(self): - hblock = gr.hier_block2("test_block", - gr.io_signature(1,1,gr.sizeof_int), - gr.io_signature(1,1,gr.sizeof_int)) - nop1 = gr.null_sink(gr.sizeof_int) - nop2 = gr.null_sink(gr.sizeof_int) - self.assertRaises(ValueError, - lambda: hblock.connect(nop1, (nop2, 1))) - - def test_009_check_topology(self): - hblock = gr.top_block("test_block") - hblock.check_topology(0, 0) - - def test_010_run(self): - expected = (1.0, 2.0, 3.0, 4.0) - hblock = gr.top_block("test_block") - src = gr.vector_source_f(expected, False) - sink1 = gr.vector_sink_f() - sink2 = gr.vector_sink_f() - hblock.connect(src, sink1) - hblock.connect(src, sink2) - hblock.run() - actual1 = sink1.data() - actual2 = sink2.data() - self.assertEquals(expected, actual1) - self.assertEquals(expected, actual2) - - def test_012_disconnect_input(self): - hblock = gr.hier_block2("test_block", - gr.io_signature(1,1,gr.sizeof_int), - gr.io_signature(1,1,gr.sizeof_int)) - nop1 = gr.nop(gr.sizeof_int) - hblock.connect(hblock, nop1) - hblock.disconnect(hblock, nop1) - - def test_013_disconnect_input_not_connected(self): - hblock = gr.hier_block2("test_block", - gr.io_signature(1,1,gr.sizeof_int), - gr.io_signature(1,1,gr.sizeof_int)) - nop1 = gr.nop(gr.sizeof_int) - nop2 = gr.nop(gr.sizeof_int) - hblock.connect(hblock, nop1) - self.assertRaises(ValueError, - lambda: hblock.disconnect(hblock, nop2)) - - def test_014_disconnect_input_neg(self): - hblock = gr.hier_block2("test_block", - gr.io_signature(1,1,gr.sizeof_int), - gr.io_signature(1,1,gr.sizeof_int)) - nop1 = gr.nop(gr.sizeof_int) - hblock.connect(hblock, nop1) - self.assertRaises(ValueError, - lambda: hblock.disconnect((hblock, -1), nop1)) - - def test_015_disconnect_input_exceeds(self): - hblock = gr.hier_block2("test_block", - gr.io_signature(1,1,gr.sizeof_int), - gr.io_signature(1,1,gr.sizeof_int)) - nop1 = gr.nop(gr.sizeof_int) - hblock.connect(hblock, nop1) - self.assertRaises(ValueError, - lambda: hblock.disconnect((hblock, 1), nop1)) - - def test_016_disconnect_output(self): - hblock = gr.hier_block2("test_block", - gr.io_signature(1,1,gr.sizeof_int), - gr.io_signature(1,1,gr.sizeof_int)) - nop1 = gr.nop(gr.sizeof_int) - hblock.connect(nop1, hblock) - hblock.disconnect(nop1, hblock) - - def test_017_disconnect_output_not_connected(self): - hblock = gr.hier_block2("test_block", - gr.io_signature(1,1,gr.sizeof_int), - gr.io_signature(1,1,gr.sizeof_int)) - nop1 = gr.nop(gr.sizeof_int) - nop2 = gr.nop(gr.sizeof_int) - hblock.connect(nop1, hblock) - self.assertRaises(ValueError, - lambda: hblock.disconnect(nop2, hblock)) - - def test_018_disconnect_output_neg(self): - hblock = gr.hier_block2("test_block", - gr.io_signature(1,1,gr.sizeof_int), - gr.io_signature(1,1,gr.sizeof_int)) - nop1 = gr.nop(gr.sizeof_int) - hblock.connect(hblock, nop1) - self.assertRaises(ValueError, - lambda: hblock.disconnect(nop1, (hblock, -1))) - - def test_019_disconnect_output_exceeds(self): - hblock = gr.hier_block2("test_block", - gr.io_signature(1,1,gr.sizeof_int), - gr.io_signature(1,1,gr.sizeof_int)) - nop1 = gr.nop(gr.sizeof_int) - hblock.connect(nop1, hblock) - self.assertRaises(ValueError, - lambda: hblock.disconnect(nop1, (hblock, 1))) - - def test_020_run(self): - hblock = gr.top_block("test_block") - data = (1.0, 2.0, 3.0, 4.0) - src = gr.vector_source_f(data, False) - dst = gr.vector_sink_f() - hblock.connect(src, dst) - hblock.run() - self.assertEquals(data, dst.data()) - - def test_021_connect_single(self): - hblock = gr.top_block("test_block") - blk = gr.hier_block2("block", - gr.io_signature(0, 0, 0), - gr.io_signature(0, 0, 0)) - hblock.connect(blk) - - def test_022_connect_single_with_ports(self): - hblock = gr.top_block("test_block") - blk = gr.hier_block2("block", - gr.io_signature(1, 1, 1), - gr.io_signature(1, 1, 1)) - self.assertRaises(ValueError, - lambda: hblock.connect(blk)) - - def test_023_connect_single_twice(self): - hblock = gr.top_block("test_block") - blk = gr.hier_block2("block", - gr.io_signature(0, 0, 0), - gr.io_signature(0, 0, 0)) - hblock.connect(blk) - self.assertRaises(ValueError, - lambda: hblock.connect(blk)) - - def test_024_disconnect_single(self): - hblock = gr.top_block("test_block") - blk = gr.hier_block2("block", - gr.io_signature(0, 0, 0), - gr.io_signature(0, 0, 0)) - hblock.connect(blk) - hblock.disconnect(blk) - - def test_025_disconnect_single_not_connected(self): - hblock = gr.top_block("test_block") - blk = gr.hier_block2("block", - gr.io_signature(0, 0, 0), - gr.io_signature(0, 0, 0)) - self.assertRaises(ValueError, - lambda: hblock.disconnect(blk)) - - def test_026_run_single(self): - expected_data = (1.0,) - tb = gr.top_block("top_block") - hb = gr.hier_block2("block", - gr.io_signature(0, 0, 0), - gr.io_signature(0, 0, 0)) - src = gr.vector_source_f(expected_data) - dst = gr.vector_sink_f() - hb.connect(src, dst) - tb.connect(hb) - tb.run() - self.assertEquals(expected_data, dst.data()) - - def test_027a_internally_unconnected_input(self): - tb = gr.top_block() - hb = gr.hier_block2("block", - gr.io_signature(1, 1, 1), - gr.io_signature(1, 1, 1)) - hsrc = gr.vector_source_b([1,]) - hb.connect(hsrc, hb) # wire output internally - src = gr.vector_source_b([1, ]) - dst = gr.vector_sink_b() - tb.connect(src, hb, dst) # hb's input is not connected internally - self.assertRaises(RuntimeError, - lambda: tb.run()) - - def test_027b_internally_unconnected_output(self): - tb = gr.top_block() - - hb = gr.hier_block2("block", - gr.io_signature(1, 1, 1), - gr.io_signature(1, 1, 1)) - hdst = gr.vector_sink_b() - hb.connect(hb, hdst) # wire input internally - src = gr.vector_source_b([1, ]) - dst = gr.vector_sink_b() - tb.connect(src, hb, dst) # hb's output is not connected internally - self.assertRaises(RuntimeError, - lambda: tb.run()) - - def test_027c_fully_unconnected_output(self): - tb = gr.top_block() - hb = gr.hier_block2("block", - gr.io_signature(1, 1, 1), - gr.io_signature(1, 1, 1)) - hsrc = gr.vector_sink_b() - hb.connect(hb, hsrc) # wire input internally - src = gr.vector_source_b([1, ]) - dst = gr.vector_sink_b() - tb.connect(src, hb) # hb's output is not connected internally or externally - self.assertRaises(RuntimeError, - lambda: tb.run()) - - def test_027d_fully_unconnected_input(self): - tb = gr.top_block() - hb = gr.hier_block2("block", - gr.io_signature(1, 1, 1), - gr.io_signature(1, 1, 1)) - hdst = gr.vector_source_b([1,]) - hb.connect(hdst, hb) # wire output internally - dst = gr.vector_sink_b() - tb.connect(hb, dst) # hb's input is not connected internally or externally - self.assertRaises(RuntimeError, - lambda: tb.run()) - - def test_028_singleton_reconfigure(self): - tb = gr.top_block() - hb = gr.hier_block2("block", - gr.io_signature(0, 0, 0), gr.io_signature(0, 0, 0)) - src = gr.vector_source_b([1, ]) - dst = gr.vector_sink_b() - hb.connect(src, dst) - tb.connect(hb) # Singleton connect - tb.lock() - tb.disconnect_all() - tb.connect(src, dst) - tb.unlock() - - def test_029_singleton_disconnect(self): - tb = gr.top_block() - src = gr.vector_source_b([1, ]) - dst = gr.vector_sink_b() - tb.connect(src, dst) - tb.disconnect(src) # Singleton disconnect - tb.connect(src, dst) - tb.run() - self.assertEquals(dst.data(), (1,)) - - def test_030_nested_input(self): - tb = gr.top_block() - src = gr.vector_source_b([1,]) - hb1 = gr.hier_block2("hb1", - gr.io_signature(1, 1, gr.sizeof_char), - gr.io_signature(0, 0, 0)) - hb2 = gr.hier_block2("hb2", - gr.io_signature(1, 1, gr.sizeof_char), - gr.io_signature(0, 0, 0)) - dst = gr.vector_sink_b() - tb.connect(src, hb1) - hb1.connect(hb1, hb2) - hb2.connect(hb2, gr.kludge_copy(gr.sizeof_char), dst) - tb.run() - self.assertEquals(dst.data(), (1,)) - - def test_031_multiple_internal_inputs(self): - tb = gr.top_block() - src = gr.vector_source_f([1.0,]) - hb = gr.hier_block2("hb", - gr.io_signature(1, 1, gr.sizeof_float), - gr.io_signature(1, 1, gr.sizeof_float)) - m1 = multiply_const_ff(1.0) - m2 = multiply_const_ff(2.0) - add = add_ff() - hb.connect(hb, m1) # m1 is connected to hb external input #0 - hb.connect(hb, m2) # m2 is also connected to hb external input #0 - hb.connect(m1, (add, 0)) - hb.connect(m2, (add, 1)) - hb.connect(add, hb) # add is connected to hb external output #0 - dst = gr.vector_sink_f() - tb.connect(src, hb, dst) - tb.run() - self.assertEquals(dst.data(), (3.0,)) - - def test_032_nested_multiple_internal_inputs(self): - tb = gr.top_block() - src = gr.vector_source_f([1.0,]) - hb = gr.hier_block2("hb", - gr.io_signature(1, 1, gr.sizeof_float), - gr.io_signature(1, 1, gr.sizeof_float)) - hb2 = gr.hier_block2("hb", - gr.io_signature(1, 1, gr.sizeof_float), - gr.io_signature(1, 1, gr.sizeof_float)) - - m1 = multiply_const_ff(1.0) - m2 = multiply_const_ff(2.0) - add = add_ff() - hb2.connect(hb2, m1) # m1 is connected to hb2 external input #0 - hb2.connect(hb2, m2) # m2 is also connected to hb2 external input #0 - hb2.connect(m1, (add, 0)) - hb2.connect(m2, (add, 1)) - hb2.connect(add, hb2) # add is connected to hb2 external output #0 - hb.connect(hb, hb2, hb) # hb as hb2 as nested internal block - dst = gr.vector_sink_f() - tb.connect(src, hb, dst) - tb.run() - self.assertEquals(dst.data(), (3.0,)) - - -if __name__ == "__main__": - gr_unittest.run(test_hier_block2, "test_hier_block2.xml") diff --git a/gnuradio-core/src/python/gnuradio/gr/qa_kludge_copy.py b/gnuradio-core/src/python/gnuradio/gr/qa_kludge_copy.py deleted file mode 100755 index 2a3aa44b1b..0000000000 --- a/gnuradio-core/src/python/gnuradio/gr/qa_kludge_copy.py +++ /dev/null @@ -1,91 +0,0 @@ -#!/usr/bin/env python -# -# Copyright 2006,2010 Free Software Foundation, Inc. -# -# This file is part of GNU Radio -# -# GNU Radio is free software; you can redistribute it and/or modify -# it under the terms of the GNU General Public License as published by -# the Free Software Foundation; either version 3, or (at your option) -# any later version. -# -# GNU Radio is distributed in the hope that it will be useful, -# but WITHOUT ANY WARRANTY; without even the implied warranty of -# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the -# GNU General Public License for more details. -# -# You should have received a copy of the GNU General Public License -# along with GNU Radio; see the file COPYING. If not, write to -# the Free Software Foundation, Inc., 51 Franklin Street, -# Boston, MA 02110-1301, USA. -# - -from gnuradio import gr, gr_unittest -import math -import random - - -class test_kludge_copy(gr_unittest.TestCase): - - def setUp(self): - self.tb = gr.top_block() - self.rng = random.Random() - self.rng.seed(0) - - def tearDown(self): - del self.tb - del self.rng - - def make_random_int_tuple(self, L): - result = [] - for x in range(L): - result.append(self.rng.randint(int(-1e9), int(+1e9))) - return tuple(result) - - - def test_001(self): - # 1 input stream; 1 output stream - src0_data = self.make_random_int_tuple(16000) - src0 = gr.vector_source_i(src0_data) - op = gr.kludge_copy(gr.sizeof_int) - dst0 = gr.vector_sink_i() - self.tb.connect(src0, op, dst0) - self.tb.run() - dst0_data = dst0.data() - self.assertEqual(src0_data, dst0_data) - - def test_002(self): - # 2 input streams; 2 output streams - src0_data = self.make_random_int_tuple(16000) - src1_data = self.make_random_int_tuple(16000) - src0 = gr.vector_source_i(src0_data) - src1 = gr.vector_source_i(src1_data) - op = gr.kludge_copy(gr.sizeof_int) - dst0 = gr.vector_sink_i() - dst1 = gr.vector_sink_i() - self.tb.connect(src0, (op, 0), dst0) - self.tb.connect(src1, (op, 1), dst1) - self.tb.run() - dst0_data = dst0.data() - dst1_data = dst1.data() - self.assertEqual(src0_data, dst0_data) - self.assertEqual(src1_data, dst1_data) - - # Note: this is disabled due to triggering bug in ticket:181 - # It only occurs with new top block code - def xtest_003(self): - # number of input streams != number of output streams - src0_data = self.make_random_int_tuple(16000) - src1_data = self.make_random_int_tuple(16000) - src0 = gr.vector_source_i(src0_data) - src1 = gr.vector_source_i(src1_data) - op = gr.kludge_copy(gr.sizeof_int) - dst0 = gr.vector_sink_i() - dst1 = gr.vector_sink_i() - self.tb.connect(src0, (op, 0), dst0) - self.tb.connect(src1, (op, 1)) - self.assertRaises(ValueError, self.tb.run) - -if __name__ == '__main__': - gr_unittest.run(test_kludge_copy, "test_kludge_copy.xml") - diff --git a/gnuradio-core/src/python/gnuradio/gr/qa_python_message_passing.py b/gnuradio-core/src/python/gnuradio/gr/qa_python_message_passing.py deleted file mode 100644 index 51f8ede4db..0000000000 --- a/gnuradio-core/src/python/gnuradio/gr/qa_python_message_passing.py +++ /dev/null @@ -1,123 +0,0 @@ -#!/usr/bin/env python -# -# Copyright 2013 Free Software Foundation, Inc. -# -# This file is part of GNU Radio -# -# GNU Radio is free software; you can redistribute it and/or modify -# it under the terms of the GNU General Public License as published by -# the Free Software Foundation; either version 3, or (at your option) -# any later version. -# -# GNU Radio is distributed in the hope that it will be useful, -# but WITHOUT ANY WARRANTY; without even the implied warranty of -# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the -# GNU General Public License for more details. -# -# You should have received a copy of the GNU General Public License -# along with GNU Radio; see the file COPYING. If not, write to -# the Free Software Foundation, Inc., 51 Franklin Street, -# Boston, MA 02110-1301, USA. -# - -from gnuradio import gr, gr_unittest -try: import pmt -except: from gruel import pmt -import numpy -import time - -# Simple block to generate messages -class message_generator(gr.sync_block): - def __init__(self, msg_list, msg_interval): - gr.sync_block.__init__( - self, - name = "message generator", - in_sig = [numpy.float32], - out_sig = None - ) - self.msg_list = msg_list - self.msg_interval = msg_interval - self.msg_ctr = 0 - self.message_port_register_out(pmt.intern('out_port')) - - - def work(self, input_items, output_items): - inLen = len(input_items[0]) - while self.msg_ctr < len(self.msg_list) and \ - (self.msg_ctr * self.msg_interval) < \ - (self.nitems_read(0) + inLen): - self.message_port_pub(pmt.intern('out_port'), - self.msg_list[self.msg_ctr]) - self.msg_ctr += 1 - return inLen - -# Simple block to consume messages -class message_consumer(gr.sync_block): - def __init__(self): - gr.sync_block.__init__( - self, - name = "message consumer", - in_sig = None, - out_sig = None - ) - self.msg_list = [] - self.message_port_register_in(pmt.intern('in_port')) - self.set_msg_handler(pmt.intern('in_port'), - self.handle_msg) - - def handle_msg(self, msg): - # Create a new PMT from long value and put in list - self.msg_list.append(pmt.from_long(pmt.to_long(msg))) - -class test_python_message_passing(gr_unittest.TestCase): - - def setUp(self): - self.tb = gr.top_block() - - def tearDown(self): - self.tb = None - - def test_000(self): - num_msgs = 10 - msg_interval = 1000 - msg_list = [] - for i in range(num_msgs): - msg_list.append(pmt.from_long(i)) - - # Create vector source with dummy data to trigger messages - src_data = [] - for i in range(num_msgs*msg_interval): - src_data.append(float(i)) - src = gr.vector_source_f(src_data, False) - msg_gen = message_generator(msg_list, msg_interval) - msg_cons = message_consumer() - - # Connect vector source to message gen - self.tb.connect(src, msg_gen) - - # Connect message generator to message consumer - self.tb.msg_connect(msg_gen, 'out_port', msg_cons, 'in_port') - - # Verify that the messgae port query functions work - self.assertEqual(pmt.symbol_to_string(pmt.vector_ref( - msg_gen.message_ports_out(), 0)), 'out_port') - self.assertEqual(pmt.symbol_to_string(pmt.vector_ref( - msg_cons.message_ports_in(), 0)), 'in_port') - - # Run to verify message passing - self.tb.start() - - # Wait for all messages to be sent - while msg_gen.msg_ctr < num_msgs: - time.sleep(0.5) - self.tb.stop() - self.tb.wait() - - # Verify that the message consumer got all the messages - self.assertEqual(num_msgs, len(msg_cons.msg_list)) - for i in range(num_msgs): - self.assertTrue(pmt.equal(msg_list[i], msg_cons.msg_list[i])) - -if __name__ == '__main__': - gr_unittest.run(test_python_message_passing, - 'test_python_message_passing.xml') diff --git a/gnuradio-core/src/python/gnuradio/gr/qa_skiphead.py b/gnuradio-core/src/python/gnuradio/gr/qa_skiphead.py deleted file mode 100755 index 1e730398c7..0000000000 --- a/gnuradio-core/src/python/gnuradio/gr/qa_skiphead.py +++ /dev/null @@ -1,102 +0,0 @@ -#!/usr/bin/env python -# -# Copyright 2007,2010 Free Software Foundation, Inc. -# -# This file is part of GNU Radio -# -# GNU Radio is free software; you can redistribute it and/or modify -# it under the terms of the GNU General Public License as published by -# the Free Software Foundation; either version 3, or (at your option) -# any later version. -# -# GNU Radio is distributed in the hope that it will be useful, -# but WITHOUT ANY WARRANTY; without even the implied warranty of -# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the -# GNU General Public License for more details. -# -# You should have received a copy of the GNU General Public License -# along with GNU Radio; see the file COPYING. If not, write to -# the Free Software Foundation, Inc., 51 Franklin Street, -# Boston, MA 02110-1301, USA. -# - -from gnuradio import gr, gr_unittest - -class test_skiphead (gr_unittest.TestCase): - - def setUp(self): - self.tb = gr.top_block () - self.src_data = [int(x) for x in range(65536)] - - def tearDown(self): - self.tb = None - - def test_skip_0(self): - skip_cnt = 0 - expected_result = tuple(self.src_data[skip_cnt:]) - src1 = gr.vector_source_i (self.src_data) - op = gr.skiphead (gr.sizeof_int, skip_cnt) - dst1 = gr.vector_sink_i () - self.tb.connect (src1, op, dst1) - self.tb.run () - dst_data = dst1.data () - self.assertEqual (expected_result, dst_data) - - def test_skip_1(self): - skip_cnt = 1 - expected_result = tuple(self.src_data[skip_cnt:]) - src1 = gr.vector_source_i (self.src_data) - op = gr.skiphead (gr.sizeof_int, skip_cnt) - dst1 = gr.vector_sink_i () - self.tb.connect (src1, op, dst1) - self.tb.run () - dst_data = dst1.data () - self.assertEqual (expected_result, dst_data) - - def test_skip_1023(self): - skip_cnt = 1023 - expected_result = tuple(self.src_data[skip_cnt:]) - src1 = gr.vector_source_i (self.src_data) - op = gr.skiphead (gr.sizeof_int, skip_cnt) - dst1 = gr.vector_sink_i () - self.tb.connect (src1, op, dst1) - self.tb.run () - dst_data = dst1.data () - self.assertEqual (expected_result, dst_data) - - def test_skip_6339(self): - skip_cnt = 6339 - expected_result = tuple(self.src_data[skip_cnt:]) - src1 = gr.vector_source_i (self.src_data) - op = gr.skiphead (gr.sizeof_int, skip_cnt) - dst1 = gr.vector_sink_i () - self.tb.connect (src1, op, dst1) - self.tb.run () - dst_data = dst1.data () - self.assertEqual (expected_result, dst_data) - - def test_skip_12678(self): - skip_cnt = 12678 - expected_result = tuple(self.src_data[skip_cnt:]) - src1 = gr.vector_source_i (self.src_data) - op = gr.skiphead (gr.sizeof_int, skip_cnt) - dst1 = gr.vector_sink_i () - self.tb.connect (src1, op, dst1) - self.tb.run () - dst_data = dst1.data () - self.assertEqual (expected_result, dst_data) - - def test_skip_all(self): - skip_cnt = len(self.src_data) - expected_result = tuple(self.src_data[skip_cnt:]) - src1 = gr.vector_source_i (self.src_data) - op = gr.skiphead (gr.sizeof_int, skip_cnt) - dst1 = gr.vector_sink_i () - self.tb.connect (src1, op, dst1) - self.tb.run () - dst_data = dst1.data () - self.assertEqual (expected_result, dst_data) - - -if __name__ == '__main__': - gr_unittest.run(test_skiphead, "test_skiphead.xml") diff --git a/gnuradio-core/src/python/gnuradio/gr/qa_vector_insert.py b/gnuradio-core/src/python/gnuradio/gr/qa_vector_insert.py deleted file mode 100755 index acc06dfded..0000000000 --- a/gnuradio-core/src/python/gnuradio/gr/qa_vector_insert.py +++ /dev/null @@ -1,58 +0,0 @@ -#!/usr/bin/env python -# -# Copyright 2012 Free Software Foundation, Inc. -# -# This file is part of GNU Radio -# -# GNU Radio is free software; you can redistribute it and/or modify -# it under the terms of the GNU General Public License as published by -# the Free Software Foundation; either version 3, or (at your option) -# any later version. -# -# GNU Radio is distributed in the hope that it will be useful, -# but WITHOUT ANY WARRANTY; without even the implied warranty of -# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the -# GNU General Public License for more details. -# -# You should have received a copy of the GNU General Public License -# along with GNU Radio; see the file COPYING. If not, write to -# the Free Software Foundation, Inc., 51 Franklin Street, -# Boston, MA 02110-1301, USA. -# - -from gnuradio import gr, gr_unittest -import math - -class test_vector_insert(gr_unittest.TestCase): - - def setUp (self): - self.tb = gr.top_block () - - def tearDown (self): - self.tb = None - - def test_001(self): - src_data = [float(x) for x in range(16)] - expected_result = tuple(src_data) - - period = 9177; - offset = 0; - - src = gr.null_source(1) - head = gr.head(1, 10000000); - ins = gr.vector_insert_b([1], period, offset); - dst = gr.vector_sink_b() - - self.tb.connect(src, head, ins, dst) - self.tb.run() - result_data = dst.data() - - for i in range(10000): - if(i%period == offset): - self.assertEqual(1, result_data[i]) - else: - self.assertEqual(0, result_data[i]) - -if __name__ == '__main__': - gr_unittest.run(test_vector_insert, "test_vector_insert.xml") - diff --git a/gnuradio-core/src/python/gnuradio/gr/qa_vector_map.py b/gnuradio-core/src/python/gnuradio/gr/qa_vector_map.py deleted file mode 100644 index 12f4be5898..0000000000 --- a/gnuradio-core/src/python/gnuradio/gr/qa_vector_map.py +++ /dev/null @@ -1,105 +0,0 @@ -#!/usr/bin/env python -# -# Copyright 2012 Free Software Foundation, Inc. -# -# This file is part of GNU Radio -# -# GNU Radio is free software; you can redistribute it and/or modify -# it under the terms of the GNU General Public License as published by -# the Free Software Foundation; either version 3, or (at your option) -# any later version. -# -# GNU Radio is distributed in the hope that it will be useful, -# but WITHOUT ANY WARRANTY; without even the implied warranty of -# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the -# GNU General Public License for more details. -# -# You should have received a copy of the GNU General Public License -# along with GNU Radio; see the file COPYING. If not, write to -# the Free Software Foundation, Inc., 51 Franklin Street, -# Boston, MA 02110-1301, USA. -# - -from gnuradio import gr, gr_unittest -import math - -class test_vector_map(gr_unittest.TestCase): - - def setUp (self): - self.tb = gr.top_block() - - def tearDown (self): - self.tb = None - - def test_reversing(self): - # Chunk data in blocks of N and reverse the block contents. - N = 5 - src_data = range(0, 20) - expected_result = [] - for i in range(N-1, len(src_data), N): - for j in range(0, N): - expected_result.append(1.0*(i-j)) - mapping = [list(reversed([(0, i) for i in range(0, N)]))] - src = gr.vector_source_f(src_data, False, N) - vmap = gr.vector_map(gr.sizeof_float, (N, ), mapping) - dst = gr.vector_sink_f(N) - self.tb.connect(src, vmap, dst) - self.tb.run() - result_data = list(dst.data()) - self.assertEqual(expected_result, result_data) - - def test_vector_to_streams(self): - # Split an input vector into N streams. - N = 5 - M = 20 - src_data = range(0, M) - expected_results = [] - for n in range(0, N): - expected_results.append(range(n, M, N)) - mapping = [[(0, n)] for n in range(0, N)] - src = gr.vector_source_f(src_data, False, N) - vmap = gr.vector_map(gr.sizeof_float, (N, ), mapping) - dsts = [gr.vector_sink_f(1) for n in range(0, N)] - self.tb.connect(src, vmap) - for n in range(0, N): - self.tb.connect((vmap, n), dsts[n]) - self.tb.run() - for n in range(0, N): - result_data = list(dsts[n].data()) - self.assertEqual(expected_results[n], result_data) - - def test_interleaving(self): - # Takes 3 streams (a, b and c) - # Outputs 2 streams. - # First (d) is interleaving of a and b. - # Second (e) is interleaving of a and b and c. c is taken in - # chunks of 2 which are reversed. - A = (1, 2, 3, 4, 5) - B = (11, 12, 13, 14, 15) - C = (99, 98, 97, 96, 95, 94, 93, 92, 91, 90) - expected_D = (1, 11, 2, 12, 3, 13, 4, 14, 5, 15) - expected_E = (1, 11, 98, 99, 2, 12, 96, 97, 3, 13, 94, 95, - 4, 14, 92, 93, 5, 15, 90, 91) - mapping = [[(0, 0), (1, 0)], # mapping to produce D - [(0, 0), (1, 0), (2, 1), (2, 0)], # mapping to produce E - ] - srcA = gr.vector_source_f(A, False, 1) - srcB = gr.vector_source_f(B, False, 1) - srcC = gr.vector_source_f(C, False, 2) - vmap = gr.vector_map(gr.sizeof_int, (1, 1, 2), mapping) - dstD = gr.vector_sink_f(2) - dstE = gr.vector_sink_f(4) - self.tb.connect(srcA, (vmap, 0)) - self.tb.connect(srcB, (vmap, 1)) - self.tb.connect(srcC, (vmap, 2)) - self.tb.connect((vmap, 0), dstD) - self.tb.connect((vmap, 1), dstE) - self.tb.run() - self.assertEqual(expected_D, dstD.data()) - self.assertEqual(expected_E, dstE.data()) - - - -if __name__ == '__main__': - gr_unittest.run(test_vector_map, "test_vector_map.xml") - diff --git a/gnuradio-core/src/python/gnuradio/gr/qa_vector_sink_source.py b/gnuradio-core/src/python/gnuradio/gr/qa_vector_sink_source.py deleted file mode 100755 index 64cbbe72a3..0000000000 --- a/gnuradio-core/src/python/gnuradio/gr/qa_vector_sink_source.py +++ /dev/null @@ -1,65 +0,0 @@ -#!/usr/bin/env python -# -# Copyright 2008,2010 Free Software Foundation, Inc. -# -# This file is part of GNU Radio -# -# GNU Radio is free software; you can redistribute it and/or modify -# it under the terms of the GNU General Public License as published by -# the Free Software Foundation; either version 3, or (at your option) -# any later version. -# -# GNU Radio is distributed in the hope that it will be useful, -# but WITHOUT ANY WARRANTY; without even the implied warranty of -# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the -# GNU General Public License for more details. -# -# You should have received a copy of the GNU General Public License -# along with GNU Radio; see the file COPYING. If not, write to -# the Free Software Foundation, Inc., 51 Franklin Street, -# Boston, MA 02110-1301, USA. -# - -from gnuradio import gr, gr_unittest -import math - -class test_vector_sink_source(gr_unittest.TestCase): - - def setUp (self): - self.tb = gr.top_block () - - def tearDown (self): - self.tb = None - - def test_001(self): - src_data = [float(x) for x in range(16)] - expected_result = tuple(src_data) - - src = gr.vector_source_f(src_data) - dst = gr.vector_sink_f() - - self.tb.connect(src, dst) - self.tb.run() - result_data = dst.data() - self.assertEqual(expected_result, result_data) - - def test_002(self): - src_data = [float(x) for x in range(16)] - expected_result = tuple(src_data) - - src = gr.vector_source_f(src_data, False, 2) - dst = gr.vector_sink_f(2) - - self.tb.connect(src, dst) - self.tb.run() - result_data = dst.data() - self.assertEqual(expected_result, result_data) - - def test_003(self): - src_data = [float(x) for x in range(16)] - expected_result = tuple(src_data) - self.assertRaises(ValueError, lambda : gr.vector_source_f(src_data, False, 3)) - -if __name__ == '__main__': - gr_unittest.run(test_vector_sink_source, "test_vector_sink_source.xml") - |