diff options
| author | Johnathan Corgan <johnathan@corganlabs.com> | 2017-10-05 16:39:09 -0700 |
|---|---|---|
| committer | Johnathan Corgan <johnathan@corganlabs.com> | 2017-10-05 16:39:09 -0700 |
| commit | 6fa9d33246251f44a0e78682e50e9a1cb0b03171 (patch) | |
| tree | 513fed9b2dfb49f9c3fbb2a7a98526d27927c166 /gr-blocks/python | |
| parent | 07a0ca6c3664669f85faeee00fa69d2af3a7b59f (diff) | |
| parent | 3c63f7334d6de70d655aa97fcccbfb950645f4d4 (diff) | |
Merge branch 'next' into python3
Diffstat (limited to 'gr-blocks/python')
| -rw-r--r-- | gr-blocks/python/blocks/qa_moving_average.py | 68 |
1 files changed, 67 insertions, 1 deletions
diff --git a/gr-blocks/python/blocks/qa_moving_average.py b/gr-blocks/python/blocks/qa_moving_average.py index 513861f007..87f8d3015f 100644 --- a/gr-blocks/python/blocks/qa_moving_average.py +++ b/gr-blocks/python/blocks/qa_moving_average.py @@ -1,6 +1,6 @@ #!/usr/bin/env python # -# Copyright 2013 Free Software Foundation, Inc. +# Copyright 2013,2017 Free Software Foundation, Inc. # # This file is part of GNU Radio # @@ -46,6 +46,10 @@ class test_moving_average(gr_unittest.TestCase): def tearDown(self): self.tb = None + # These tests will always pass and are therefore useless. 100 random numbers [-1,1) are + # getting summed up and scaled with 0.001. Then, an assertion verifies a result near 0, + # which is the case even if the block is malfunctioning. + def test_01(self): tb = self.tb @@ -88,5 +92,67 @@ class test_moving_average(gr_unittest.TestCase): # make sure result is close to zero self.assertComplexTuplesAlmostEqual(expected_result, dst_data, 1) + # This tests implement own moving average to verify correct behaviour of the block + + def test_03(self): + tb = self.tb + + vlen = 5 + N = 10*vlen + seed = 0 + data = make_random_float_tuple(N, 2**10) + data = [int(d*1000) for d in data] + src = blocks.vector_source_i(data, False) + one_to_many = blocks.stream_to_streams(gr.sizeof_int, vlen) + one_to_vector = blocks.stream_to_vector(gr.sizeof_int, vlen) + many_to_vector = blocks.streams_to_vector(gr.sizeof_int, vlen) + isolated = [ blocks.moving_average_ii(100, 1) for i in range(vlen)] + dut = blocks.moving_average_ii(100, 1, vlen=vlen) + dut_dst = blocks.vector_sink_i(vlen=vlen) + ref_dst = blocks.vector_sink_i(vlen=vlen) + + tb.connect(src, one_to_many) + tb.connect(src, one_to_vector, dut, dut_dst) + tb.connect(many_to_vector, ref_dst) + for idx, single in enumerate(isolated): + tb.connect((one_to_many,idx), single, (many_to_vector,idx)) + + tb.run() + + dut_data = dut_dst.data() + ref_data = ref_dst.data() + + # make sure result is close to zero + self.assertTupleEqual(dut_data, ref_data) + + def test_04(self): + tb = self.tb + + N = 10000 # number of samples + history = 100 # num of samples to average + data = make_random_complex_tuple(N, 1) # generate random data + + # pythonic MA filter + data_padded = (history-1)*[0.0+1j*0.0]+list(data) # history + expected_result = [] + moving_sum = sum(data_padded[:history-1]) + for i in range(N): + moving_sum += data_padded[i+history-1] + expected_result.append(moving_sum) + moving_sum -= data_padded[i] + + src = blocks.vector_source_c(data, False) + op = blocks.moving_average_cc(history, 1) + dst = blocks.vector_sink_c() + + tb.connect(src, op) + tb.connect(op, dst) + tb.run() + + dst_data = dst.data() + + # make sure result is close to zero + self.assertComplexTuplesAlmostEqual(expected_result, dst_data, 4) + if __name__ == '__main__': gr_unittest.run(test_moving_average, "test_moving_average.xml") |