diff options
Diffstat (limited to 'gr-blocks/python')
-rw-r--r-- | gr-blocks/python/blocks/qa_file_descriptor_source_sink.py | 70 | ||||
-rw-r--r-- | gr-blocks/python/blocks/qa_file_sink.py | 60 | ||||
-rw-r--r-- | gr-blocks/python/blocks/qa_file_source.py | 153 | ||||
-rw-r--r-- | gr-blocks/python/blocks/qa_file_source_sink.py | 169 | ||||
-rw-r--r-- | gr-blocks/python/blocks/qa_hier_block2.py | 26 | ||||
-rw-r--r-- | gr-blocks/python/blocks/qa_wavfile.py | 2 |
6 files changed, 297 insertions, 183 deletions
diff --git a/gr-blocks/python/blocks/qa_file_descriptor_source_sink.py b/gr-blocks/python/blocks/qa_file_descriptor_source_sink.py new file mode 100644 index 0000000000..c84a82d59f --- /dev/null +++ b/gr-blocks/python/blocks/qa_file_descriptor_source_sink.py @@ -0,0 +1,70 @@ +#!/usr/bin/env python +# +# Copyright 2018 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, blocks +import os +import tempfile +import pmt + +class test_file_descriptor_source_sink(gr_unittest.TestCase): + + def setUp (self): + os.environ['GR_CONF_CONTROLPORT_ON'] = 'False' + self.tb = gr.top_block () + + def tearDown (self): + self.tb = None + + def test_file_descriptor(self): + src_data = range(1000) + expected_result = range(1000) + + snk2 = blocks.vector_sink_f() + + with tempfile.NamedTemporaryFile() as temp: + fhandle0 = open(temp.name, "wb") + fd0 = fhandle0.fileno() + + src = blocks.vector_source_f(src_data) + snk = blocks.file_descriptor_sink(gr.sizeof_float, fd0) + + self.tb.connect(src, snk) + self.tb.run() + os.fsync(fd0) + fhandle0.close() + + fhandle1 = open(temp.name, "rb") + fd1 = fhandle1.fileno() + src2 = blocks.file_descriptor_source(gr.sizeof_float, fd1, False) + + self.tb.disconnect(src, snk) + self.tb.connect(src2, snk2) + self.tb.run() + os.fsync(fd1) + fhandle1.close() + + result_data = snk2.data() + self.assertFloatTuplesAlmostEqual(expected_result, result_data) + self.assertEqual(len(snk2.tags()), 0) + +if __name__ == '__main__': + gr_unittest.run(test_file_descriptor_source_sink, "test_file_descriptor_source_sink.xml") diff --git a/gr-blocks/python/blocks/qa_file_sink.py b/gr-blocks/python/blocks/qa_file_sink.py new file mode 100644 index 0000000000..a7296183ef --- /dev/null +++ b/gr-blocks/python/blocks/qa_file_sink.py @@ -0,0 +1,60 @@ +#!/usr/bin/env python +# +# Copyright 2018 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, blocks +import os +import tempfile +import pmt +import array + +class test_file_sink(gr_unittest.TestCase): + + def setUp (self): + os.environ['GR_CONF_CONTROLPORT_ON'] = 'False' + self.tb = gr.top_block () + + def tearDown (self): + self.tb = None + + def test_file_sink(self): + data = range(1000) + expected_result = data + + with tempfile.NamedTemporaryFile() as temp: + src = blocks.vector_source_f(data) + snk = blocks.file_sink(gr.sizeof_float, temp.name) + snk.set_unbuffered(True) + self.tb.connect(src, snk) + self.tb.run() + + # Check file length (float: 4 * nsamples) + st = os.stat(temp.name) + self.assertEqual(st.st_size, 4 * len(data)) + + # Check file contents + datafile = open(temp.name, 'r') + result_data = array.array('f') + result_data.fromfile(datafile, len(data)) + self.assertFloatTuplesAlmostEqual(expected_result, result_data) + +if __name__ == '__main__': + gr_unittest.run(test_file_sink, "test_file_sink.xml") diff --git a/gr-blocks/python/blocks/qa_file_source.py b/gr-blocks/python/blocks/qa_file_source.py new file mode 100644 index 0000000000..f8ca75d79c --- /dev/null +++ b/gr-blocks/python/blocks/qa_file_source.py @@ -0,0 +1,153 @@ +#!/usr/bin/env python +# +# Copyright 2018 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, blocks +import os +import tempfile +import pmt +import array + +class test_file_source(gr_unittest.TestCase): + + @classmethod + def setUpClass(cls): + os.environ['GR_CONF_CONTROLPORT_ON'] = 'False' + cls._datafile = tempfile.NamedTemporaryFile() + cls._datafilename = cls._datafile.name + cls._vector = range(1000) + with open(cls._datafilename, 'w') as f: + array.array('f', cls._vector).tofile(f) + + @classmethod + def tearDownClass(cls): + del cls._vector + del cls._datafilename + del cls._datafile + + def setUp (self): + self.tb = gr.top_block () + + def tearDown (self): + self.tb = None + + def test_file_source(self): + expected_result = self._vector + + src = blocks.file_source(gr.sizeof_float, self._datafilename) + snk = blocks.vector_sink_f() + self.tb.connect(src, snk) + self.tb.run() + + result_data = snk.data() + self.assertFloatTuplesAlmostEqual(self._vector, result_data) + self.assertEqual(len(snk.tags()), 0) + + def test_file_source_no_such_file(self): + + try: + src = blocks.file_source(gr.sizeof_float, "___no_such_file___") + self.assertTrue(False) + except RuntimeError, e: + self.assertTrue(True) + + def test_file_source_with_offset(self): + expected_result = self._vector[100:] + + src = blocks.file_source(gr.sizeof_float, self._datafilename, offset=100) + snk = blocks.vector_sink_f() + + self.tb.connect(src, snk) + self.tb.run() + + result_data = snk.data() + self.assertFloatTuplesAlmostEqual(expected_result, result_data) + self.assertEqual(len(snk.tags()), 0) + + def test_source_with_offset_and_len(self): + expected_result = self._vector[100:100+600] + + src = blocks.file_source(gr.sizeof_float, self._datafilename, offset=100, len=600) + snk = blocks.vector_sink_f() + self.tb.connect(src, snk) + self.tb.run() + + result_data = snk.data() + self.assertFloatTuplesAlmostEqual(expected_result, result_data) + self.assertEqual(len(snk.tags()), 0) + + def test_file_source_can_seek_after_open(self): + + src = blocks.file_source(gr.sizeof_float, self._datafilename) + self.assertTrue(src.seek(0, os.SEEK_SET)) + self.assertTrue(src.seek(len(self._vector)-1, os.SEEK_SET)) + # Seek past end of file - this will also log a warning + self.assertFalse(src.seek(len(self._vector), os.SEEK_SET)) + # Negative seek - this will also log a warning + self.assertFalse(src.seek(-1, os.SEEK_SET)) + + self.assertTrue(src.seek(1, os.SEEK_END)) + self.assertTrue(src.seek(len(self._vector), os.SEEK_END)) + # Seek past end of file - this will also log a warning + self.assertFalse(src.seek(0, os.SEEK_END)) + + self.assertTrue(src.seek(0, os.SEEK_SET)) + self.assertTrue(src.seek(1, os.SEEK_CUR)) + # Seek past end of file - this will also log a warning + self.assertFalse(src.seek(len(self._vector), os.SEEK_CUR)); + + + def test_begin_tag(self): + expected_result = self._vector + + src = blocks.file_source(gr.sizeof_float, self._datafilename) + src.set_begin_tag(pmt.string_to_symbol("file_begin")) + snk = blocks.vector_sink_f() + self.tb.connect(src, snk) + self.tb.run() + + result_data = snk.data() + self.assertFloatTuplesAlmostEqual(expected_result, result_data) + self.assertEqual(len(snk.tags()), 1) + + def test_begin_tag_repeat(self): + expected_result = self._vector + self._vector + + src = blocks.file_source(gr.sizeof_float, self._datafilename, True) + src.set_begin_tag(pmt.string_to_symbol("file_begin")) + hd = blocks.head(gr.sizeof_float, 2 * len(self._vector)) + snk = blocks.vector_sink_f() + self.tb.connect(src, hd, snk) + self.tb.run() + + result_data = snk.data() + self.assertFloatTuplesAlmostEqual(expected_result, result_data) + tags = snk.tags() + self.assertEqual(len(tags), 2) + self.assertEqual(str(tags[0].key), "file_begin") + self.assertEqual(str(tags[0].value), "0") + self.assertEqual(tags[0].offset, 0) + self.assertEqual(str(tags[1].key), "file_begin") + self.assertEqual(str(tags[1].value), "1") + self.assertEqual(tags[1].offset, 1000) + +if __name__ == '__main__': + gr_unittest.run(test_file_source, "test_file_source.xml") diff --git a/gr-blocks/python/blocks/qa_file_source_sink.py b/gr-blocks/python/blocks/qa_file_source_sink.py deleted file mode 100644 index e39a34ad82..0000000000 --- a/gr-blocks/python/blocks/qa_file_source_sink.py +++ /dev/null @@ -1,169 +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, blocks -import os -import tempfile -import pmt - -class test_file_source_sink(gr_unittest.TestCase): - - def setUp (self): - os.environ['GR_CONF_CONTROLPORT_ON'] = 'False' - self.tb = gr.top_block () - - def tearDown (self): - self.tb = None - - def test_001(self): - src_data = list(range(1000)) - expected_result = list(range(1000)) - - snk2 = blocks.vector_sink_f() - - with tempfile.NamedTemporaryFile() as temp: - src = blocks.vector_source_f(src_data) - snk = blocks.file_sink(gr.sizeof_float, temp.name) - snk.set_unbuffered(True) - - src2 = blocks.file_source(gr.sizeof_float, temp.name) - - self.tb.connect(src, snk) - self.tb.run() - - self.tb.disconnect(src, snk) - self.tb.connect(src2, snk2) - self.tb.run() - - result_data = snk2.data() - self.assertFloatTuplesAlmostEqual(expected_result, result_data) - self.assertEqual(len(snk2.tags()), 0) - - def test_descriptor_001(self): - src_data = list(range(1000)) - expected_result = list(range(1000)) - - snk2 = blocks.vector_sink_f() - - with tempfile.NamedTemporaryFile() as temp: - fhandle0 = open(temp.name, "wb") - fd0 = fhandle0.fileno() - - src = blocks.vector_source_f(src_data) - snk = blocks.file_descriptor_sink(gr.sizeof_float, fd0) - - self.tb.connect(src, snk) - self.tb.run() - os.fsync(fd0) - fhandle0.close() - - fhandle1 = open(temp.name, "rb") - fd1 = fhandle1.fileno() - src2 = blocks.file_descriptor_source(gr.sizeof_float, fd1, False) - - self.tb.disconnect(src, snk) - self.tb.connect(src2, snk2) - self.tb.run() - os.fsync(fd1) - fhandle1.close() - - result_data = snk2.data() - self.assertFloatTuplesAlmostEqual(expected_result, result_data) - self.assertEqual(len(snk2.tags()), 0) - - def test_file_source_can_seek_after_open(self): - src_data = list(range(1000)) - - with tempfile.NamedTemporaryFile() as temp: - src = blocks.vector_source_f(src_data) - snk = blocks.file_sink(gr.sizeof_float, temp.name) - snk.set_unbuffered(True) - - self.tb.connect(src, snk) - self.tb.run() - - source = blocks.file_source(gr.sizeof_float, temp.name) - self.assertTrue(source.seek(0, os.SEEK_SET)) - - def test_begin_tag(self): - src_data = range(1000) - expected_result = range(1000) - - snk2 = blocks.vector_sink_f() - - with tempfile.NamedTemporaryFile() as temp: - src = blocks.vector_source_f(src_data) - snk = blocks.file_sink(gr.sizeof_float, temp.name) - snk.set_unbuffered(True) - - src2 = blocks.file_source(gr.sizeof_float, temp.name) - src2.set_begin_tag(pmt.string_to_symbol("file_begin")) - - self.tb.connect(src, snk) - self.tb.run() - - self.tb.disconnect(src, snk) - self.tb.connect(src2, snk2) - self.tb.run() - - result_data = snk2.data() - self.assertFloatTuplesAlmostEqual(expected_result, result_data) - self.assertEqual(len(snk2.tags()), 1) - - def test_begin_tag_repeat(self): - src_data = range(1000) - expected_result = range(1000) - expected_result.extend(range(1000)) - - snk2 = blocks.vector_sink_f() - - with tempfile.NamedTemporaryFile() as temp: - src = blocks.vector_source_f(src_data) - snk = blocks.file_sink(gr.sizeof_float, temp.name) - snk.set_unbuffered(True) - - src2 = blocks.file_source(gr.sizeof_float, temp.name, True) - src2.set_begin_tag(pmt.string_to_symbol("file_begin")) - hd = blocks.head(gr.sizeof_float, 2000) - - self.tb.connect(src, snk) - self.tb.run() - - self.tb.disconnect(src, snk) - self.tb.connect(src2, hd, snk2) - self.tb.run() - - result_data = snk2.data() - self.assertFloatTuplesAlmostEqual(expected_result, result_data) - tags = snk2.tags() - self.assertEqual(len(tags), 2) - self.assertEqual(str(tags[0].key), "file_begin") - self.assertEqual(str(tags[0].value), "0") - self.assertEqual(tags[0].offset, 0) - self.assertEqual(str(tags[1].key), "file_begin") - self.assertEqual(str(tags[1].value), "1") - self.assertEqual(tags[1].offset, 1000) - -if __name__ == '__main__': - gr_unittest.run(test_file_source_sink, "test_file_source_sink.xml") - diff --git a/gr-blocks/python/blocks/qa_hier_block2.py b/gr-blocks/python/blocks/qa_hier_block2.py index 20c6a7c180..fefe0c3b49 100644 --- a/gr-blocks/python/blocks/qa_hier_block2.py +++ b/gr-blocks/python/blocks/qa_hier_block2.py @@ -74,7 +74,7 @@ class test_hier_block2(gr_unittest.TestCase): nop1 = blocks.nop(gr.sizeof_int) nop2 = blocks.nop(gr.sizeof_int) hblock.connect(nop1, hblock) - self.assertRaises(ValueError, + self.assertRaises(RuntimeError, lambda: hblock.connect(nop2, hblock)) def test_006_connect_invalid_src_port_neg(self): @@ -82,7 +82,7 @@ class test_hier_block2(gr_unittest.TestCase): gr.io_signature(1,1,gr.sizeof_int), gr.io_signature(1,1,gr.sizeof_int)) nop1 = blocks.nop(gr.sizeof_int) - self.assertRaises(ValueError, + self.assertRaises(RuntimeError, lambda: hblock.connect((hblock, -1), nop1)) def test_005_connect_invalid_src_port_exceeds(self): @@ -90,7 +90,7 @@ class test_hier_block2(gr_unittest.TestCase): gr.io_signature(1,1,gr.sizeof_int), gr.io_signature(1,1,gr.sizeof_int)) nop1 = blocks.nop(gr.sizeof_int) - self.assertRaises(ValueError, + self.assertRaises(RuntimeError, lambda: hblock.connect((hblock, 1), nop1)) def test_007_connect_invalid_dst_port_neg(self): @@ -99,7 +99,7 @@ class test_hier_block2(gr_unittest.TestCase): gr.io_signature(1,1,gr.sizeof_int)) nop1 = blocks.nop(gr.sizeof_int) nop2 = blocks.nop(gr.sizeof_int) - self.assertRaises(ValueError, + self.assertRaises(RuntimeError, lambda: hblock.connect(nop1, (nop2, -1))) def test_008_connect_invalid_dst_port_exceeds(self): @@ -108,7 +108,7 @@ class test_hier_block2(gr_unittest.TestCase): gr.io_signature(1,1,gr.sizeof_int)) nop1 = blocks.null_sink(gr.sizeof_int) nop2 = blocks.null_sink(gr.sizeof_int) - self.assertRaises(ValueError, + self.assertRaises(RuntimeError, lambda: hblock.connect(nop1, (nop2, 1))) def test_009_check_topology(self): @@ -144,7 +144,7 @@ class test_hier_block2(gr_unittest.TestCase): nop1 = blocks.nop(gr.sizeof_int) nop2 = blocks.nop(gr.sizeof_int) hblock.connect(hblock, nop1) - self.assertRaises(ValueError, + self.assertRaises(RuntimeError, lambda: hblock.disconnect(hblock, nop2)) def test_014_disconnect_input_neg(self): @@ -162,7 +162,7 @@ class test_hier_block2(gr_unittest.TestCase): gr.io_signature(1,1,gr.sizeof_int)) nop1 = blocks.nop(gr.sizeof_int) hblock.connect(hblock, nop1) - self.assertRaises(ValueError, + self.assertRaises(RuntimeError, lambda: hblock.disconnect((hblock, 1), nop1)) def test_016_disconnect_output(self): @@ -180,7 +180,7 @@ class test_hier_block2(gr_unittest.TestCase): nop1 = blocks.nop(gr.sizeof_int) nop2 = blocks.nop(gr.sizeof_int) hblock.connect(nop1, hblock) - self.assertRaises(ValueError, + self.assertRaises(RuntimeError, lambda: hblock.disconnect(nop2, hblock)) def test_018_disconnect_output_neg(self): @@ -189,7 +189,7 @@ class test_hier_block2(gr_unittest.TestCase): gr.io_signature(1,1,gr.sizeof_int)) nop1 = blocks.nop(gr.sizeof_int) hblock.connect(hblock, nop1) - self.assertRaises(ValueError, + self.assertRaises(RuntimeError, lambda: hblock.disconnect(nop1, (hblock, -1))) def test_019_disconnect_output_exceeds(self): @@ -198,7 +198,7 @@ class test_hier_block2(gr_unittest.TestCase): gr.io_signature(1,1,gr.sizeof_int)) nop1 = blocks.nop(gr.sizeof_int) hblock.connect(nop1, hblock) - self.assertRaises(ValueError, + self.assertRaises(RuntimeError, lambda: hblock.disconnect(nop1, (hblock, 1))) def test_020_run(self): @@ -222,7 +222,7 @@ class test_hier_block2(gr_unittest.TestCase): blk = gr.hier_block2("block", gr.io_signature(1, 1, 1), gr.io_signature(1, 1, 1)) - self.assertRaises(ValueError, + self.assertRaises(RuntimeError, lambda: hblock.connect(blk)) def test_023_connect_single_twice(self): @@ -231,7 +231,7 @@ class test_hier_block2(gr_unittest.TestCase): gr.io_signature(0, 0, 0), gr.io_signature(0, 0, 0)) hblock.connect(blk) - self.assertRaises(ValueError, + self.assertRaises(RuntimeError, lambda: hblock.connect(blk)) def test_024_disconnect_single(self): @@ -247,7 +247,7 @@ class test_hier_block2(gr_unittest.TestCase): blk = gr.hier_block2("block", gr.io_signature(0, 0, 0), gr.io_signature(0, 0, 0)) - self.assertRaises(ValueError, + self.assertRaises(RuntimeError, lambda: hblock.disconnect(blk)) def test_026_run_single(self): diff --git a/gr-blocks/python/blocks/qa_wavfile.py b/gr-blocks/python/blocks/qa_wavfile.py index 438da000ef..761b0be9c5 100644 --- a/gr-blocks/python/blocks/qa_wavfile.py +++ b/gr-blocks/python/blocks/qa_wavfile.py @@ -55,7 +55,7 @@ class test_wavefile(gr_unittest.TestCase): self.tb.run() wf_out.close() - # we're loosing all extra header chunks + # we're losing all extra header chunks self.assertEqual(getsize(infile) - g_extra_header_len, getsize(outfile)) in_f = open(infile, 'rb') |