diff options
| author | japm48 <japm48@users.noreply.github.com> | 2020-04-01 03:26:52 +0200 |
|---|---|---|
| committer | Michael Dickens <michael.dickens@ettus.com> | 2020-04-14 15:47:29 -0400 |
| commit | 36d811dab3832f682334b56257e9ed8f5792cc39 (patch) | |
| tree | bb6473015c35af7eddccc7c130f17a8dca622492 /gr-blocks/python | |
| parent | 7975fb0f63bd90954960658be80790b7a518e64f (diff) | |
blocks: wavfile_sink: allow appending to files
Fixes issue #1569.
Implement an option in wavfile_sink that allows appending
sound samples at the end of an existing file.
For this, a target WAV file must meet these requirements:
- the file must exist (will not be created if it does not).
- the last chunk in the file must be DATA.
- file params must match the configured ones (i.e. sample_rate, nchan
and bits_per_sample).
Diffstat (limited to 'gr-blocks/python')
| -rw-r--r-- | gr-blocks/python/blocks/qa_wavfile.py | 77 |
1 files changed, 77 insertions, 0 deletions
diff --git a/gr-blocks/python/blocks/qa_wavfile.py b/gr-blocks/python/blocks/qa_wavfile.py index ba2d80dc66..f22934d71b 100644 --- a/gr-blocks/python/blocks/qa_wavfile.py +++ b/gr-blocks/python/blocks/qa_wavfile.py @@ -71,5 +71,82 @@ class test_wavefile(gr_unittest.TestCase): self.assertEqual(in_data[:g_extra_header_offset] + \ in_data[g_extra_header_offset + g_extra_header_len:], out_data) + def test_003_checkwav_append_copy(self): + infile = g_in_file + outfile = "test_out_append.wav" + + # 1. Copy input to output + from shutil import copyfile + copyfile(infile, outfile) + + # 2. append copy + wf_in = blocks.wavfile_source(infile) + wf_out = blocks.wavfile_sink(outfile, + wf_in.channels(), + wf_in.sample_rate(), + wf_in.bits_per_sample(), + True) + self.tb.connect(wf_in, wf_out) + self.tb.run() + wf_out.close() + + # 3. append halved copy + wf_in = blocks.wavfile_source(infile) + halver = blocks.multiply_const_ff(0.5) + wf_out = blocks.wavfile_sink(outfile, + wf_in.channels(), + wf_in.sample_rate(), + wf_in.bits_per_sample(), + True) + self.tb.connect(wf_in, halver, wf_out) + self.tb.run() + wf_out.close() + + # Test file validity and read data. + import wave + try: + # In + with wave.open(infile, 'rb') as w_in: + in_params = w_in.getparams() + data_in = wav_read_frames(w_in) + # Out + with wave.open(outfile, 'rb') as w_out: + out_params = w_out.getparams() + data_out = wav_read_frames(w_out) + except: + raise AssertionError('Invalid WAV file') + + # Params must be equal except in size: + expected_params = in_params._replace(nframes=3*in_params.nframes) + self.assertEqual(out_params, expected_params) + + # Part 1 + self.assertEqual(data_in, data_out[:len(data_in)]) + + # Part 2 + self.assertEqual(data_in, data_out[len(data_in):2*len(data_in)]) + + # Part 3 + data_in_halved = [int(round(d/2)) for d in data_in] + self.assertEqual(data_in_halved, data_out[2*len(data_in):]) + + def test_003_checkwav_append_non_existent_should_error(self): + outfile = "no_file.wav" + + with self.assertRaisesRegex(RuntimeError, 'WAV append mode requires target file to exist'): + blocks.wavfile_sink(outfile, 1, 44100, 16, True) + + +def wav_read_frames(w): + import struct + # grouper from itertools recipes. + grouper = lambda iterable, n: list(zip(* ([iter(iterable)] * n) )) + assert w.getsampwidth() == 2 # Assume 16 bits + return [ + struct.unpack('<h', bytes(frame_g))[0] + for frame_g in grouper(w.readframes(w.getnframes()), 2) + ] + + if __name__ == '__main__': gr_unittest.run(test_wavefile) |