Changeset f264b17 for python/tests/test_sink.py
- Timestamp:
- Jun 22, 2016, 1:00:10 PM (9 years ago)
- Branches:
- feature/autosink, feature/cnn, feature/cnn_org, feature/constantq, feature/crepe, feature/crepe_org, feature/pitchshift, feature/pydocstrings, feature/timestretch, fix/ffmpeg5, master, pitchshift, sampler, timestretch, yinfft+
- Children:
- 4b9443c4
- Parents:
- 60fc05b (diff), 6769586 (diff)
Note: this is a merge changeset, the changes displayed below correspond to the merge itself.
Use the(diff)links above to see all the changes relative to each parent. - File:
-
- 1 edited
-
python/tests/test_sink.py (modified) (4 diffs)
Legend:
- Unmodified
- Added
- Removed
-
python/tests/test_sink.py
r60fc05b rf264b17 1 1 #! /usr/bin/env python 2 2 3 from numpy.testing import TestCase, assert_equal, assert_almost_equal 3 from nose2 import main 4 from nose2.tools import params 5 from numpy.testing import TestCase 4 6 from aubio import fvec, source, sink 5 from numpy import array6 7 from utils import list_all_sounds, get_tmp_sink_path, del_tmp_sink_path 7 8 8 9 list_of_sounds = list_all_sounds('sounds') 10 samplerates = [0, 44100, 8000, 32000] 11 hop_sizes = [512, 1024, 64] 12 9 13 path = None 10 14 11 15 many_files = 300 # 256 opened files is too much 12 16 17 all_params = [] 18 for soundfile in list_of_sounds: 19 for hop_size in hop_sizes: 20 for samplerate in samplerates: 21 all_params.append((hop_size, samplerate, soundfile)) 22 13 23 class aubio_sink_test_case(TestCase): 24 25 def setUp(self): 26 if not len(list_of_sounds): 27 self.skipTest('add some sound files in \'python/tests/sounds\'') 14 28 15 29 def test_many_sinks(self): … … 24 38 sink_list.append(g) 25 39 write = 32 26 for nin range(200):40 for _ in range(200): 27 41 vec = fvec(write) 28 42 g(vec, write) … … 30 44 shutil.rmtree(tmpdir) 31 45 32 def test_many_sinks_not_closed(self): 33 from tempfile import mkdtemp 34 import os.path 35 import shutil 36 tmpdir = mkdtemp() 37 sink_list = [] 46 @params(*all_params) 47 def test_read_and_write(self, hop_size, samplerate, path): 48 38 49 try: 39 for i in range(many_files): 40 path = os.path.join(tmpdir, 'f-' + str(i) + '.wav') 41 g = sink(path, 0) 42 sink_list.append(g) 43 write = 256 44 for n in range(200): 45 vec = fvec(write) 46 g(vec, write) 47 except StandardError: 48 pass 49 else: 50 self.fail("does not fail on too many files open") 51 for g in sink_list: 52 g.close() 53 shutil.rmtree(tmpdir) 50 f = source(path, samplerate, hop_size) 51 except RuntimeError as e: 52 self.skipTest('failed opening with hop_s = {:d}, samplerate = {:d} ({:s})'.format(hop_size, samplerate, str(e))) 53 if samplerate == 0: samplerate = f.samplerate 54 sink_path = get_tmp_sink_path() 55 g = sink(sink_path, samplerate) 56 total_frames = 0 57 while True: 58 vec, read = f() 59 g(vec, read) 60 total_frames += read 61 if read < f.hop_size: break 62 del_tmp_sink_path(sink_path) 54 63 55 def test_read_and_write(self): 56 57 if not len(list_of_sounds): 58 self.skipTest('add some sound files in \'python/tests/sounds\'') 59 60 for path in list_of_sounds: 61 for samplerate, hop_size in zip([0, 44100, 8000, 32000], [512, 1024, 64, 256]): 62 f = source(path, samplerate, hop_size) 63 if samplerate == 0: samplerate = f.samplerate 64 sink_path = get_tmp_sink_path() 65 g = sink(sink_path, samplerate) 66 total_frames = 0 67 while True: 68 vec, read = f() 69 g(vec, read) 70 total_frames += read 71 if read < f.hop_size: break 72 if 0: 73 print "read", "%.2fs" % (total_frames / float(f.samplerate) ), 74 print "(", total_frames, "frames", "in", 75 print total_frames / f.hop_size, "blocks", "at", "%dHz" % f.samplerate, ")", 76 print "from", f.uri, 77 print "to", g.uri 78 del_tmp_sink_path(sink_path) 79 80 def test_read_and_write_multi(self): 81 82 if not len(list_of_sounds): 83 self.skipTest('add some sound files in \'python/tests/sounds\'') 84 85 for path in list_of_sounds: 86 for samplerate, hop_size in zip([0, 44100, 8000, 32000], [512, 1024, 64, 256]): 87 f = source(path, samplerate, hop_size) 88 if samplerate == 0: samplerate = f.samplerate 89 sink_path = get_tmp_sink_path() 90 g = sink(sink_path, samplerate, channels = f.channels) 91 total_frames = 0 92 while True: 93 vec, read = f.do_multi() 94 g.do_multi(vec, read) 95 total_frames += read 96 if read < f.hop_size: break 97 if 0: 98 print "read", "%.2fs" % (total_frames / float(f.samplerate) ), 99 print "(", total_frames, "frames", "in", 100 print f.channels, "channels", "in", 101 print total_frames / f.hop_size, "blocks", "at", "%dHz" % f.samplerate, ")", 102 print "from", f.uri, 103 print "to", g.uri, 104 print "in", g.channels, "channels" 105 del_tmp_sink_path(sink_path) 64 @params(*all_params) 65 def test_read_and_write_multi(self, hop_size, samplerate, path): 66 try: 67 f = source(path, samplerate, hop_size) 68 except RuntimeError as e: 69 self.skipTest('failed opening with hop_s = {:d}, samplerate = {:d} ({:s})'.format(hop_size, samplerate, str(e))) 70 if samplerate == 0: samplerate = f.samplerate 71 sink_path = get_tmp_sink_path() 72 g = sink(sink_path, samplerate, channels = f.channels) 73 total_frames = 0 74 while True: 75 vec, read = f.do_multi() 76 g.do_multi(vec, read) 77 total_frames += read 78 if read < f.hop_size: break 79 del_tmp_sink_path(sink_path) 106 80 107 81 def test_close_file(self): … … 121 95 122 96 if __name__ == '__main__': 123 from unittest import main124 97 main()
Note: See TracChangeset
for help on using the changeset viewer.
