6defbcf0296734da0d02b633453a21f736e7502c
[aubio.git] / python / tests / test_sink.py
1 #! /usr/bin/env python
2
3 from nose2 import main
4 from nose2.tools import params
5 from numpy.testing import TestCase
6 from aubio import fvec, source, sink
7 from .utils import list_all_sounds, get_tmp_sink_path, del_tmp_sink_path
8
9 import warnings
10 warnings.filterwarnings('ignore', category=UserWarning, append=True)
11
12 list_of_sounds = list_all_sounds('sounds')
13 samplerates = [0, 44100, 8000, 32000]
14 hop_sizes = [512, 1024, 64]
15
16 path = None
17
18 many_files = 300 # 256 opened files is too much
19
20 all_params = []
21 for soundfile in list_of_sounds:
22     for hop_size in hop_sizes:
23         for samplerate in samplerates:
24             all_params.append((hop_size, samplerate, soundfile))
25
26 class aubio_sink_test_case(TestCase):
27
28     def setUp(self):
29         if not len(list_of_sounds):
30             self.skipTest('add some sound files in \'python/tests/sounds\'')
31
32     def test_wrong_filename(self):
33         with self.assertRaises(RuntimeError):
34             sink('')
35
36     def test_wrong_samplerate(self):
37         with self.assertRaises(RuntimeError):
38             sink(get_tmp_sink_path(), -1)
39
40     def test_wrong_samplerate_too_large(self):
41         with self.assertRaises(RuntimeError):
42             sink(get_tmp_sink_path(), 1536001, 2)
43
44     def test_wrong_channels(self):
45         with self.assertRaises(RuntimeError):
46             sink(get_tmp_sink_path(), 44100, -1)
47
48     def test_wrong_channels_too_large(self):
49         with self.assertRaises(RuntimeError):
50             sink(get_tmp_sink_path(), 44100, 202020)
51
52     def test_many_sinks(self):
53         from tempfile import mkdtemp
54         import os.path
55         import shutil
56         tmpdir = mkdtemp()
57         sink_list = []
58         for i in range(many_files):
59             path = os.path.join(tmpdir, 'f-' + str(i) + '.wav')
60             g = sink(path, 0)
61             sink_list.append(g)
62             write = 32
63             for _ in range(200):
64                 vec = fvec(write)
65                 g(vec, write)
66             g.close()
67         shutil.rmtree(tmpdir)
68
69     @params(*all_params)
70     def test_read_and_write(self, hop_size, samplerate, path):
71
72         try:
73             f = source(path, samplerate, hop_size)
74         except RuntimeError as e:
75             self.skipTest('failed opening with hop_s = {:d}, samplerate = {:d} ({:s})'.format(hop_size, samplerate, str(e)))
76         if samplerate == 0: samplerate = f.samplerate
77         sink_path = get_tmp_sink_path()
78         g = sink(sink_path, samplerate)
79         total_frames = 0
80         while True:
81             vec, read = f()
82             g(vec, read)
83             total_frames += read
84             if read < f.hop_size: break
85         del_tmp_sink_path(sink_path)
86
87     @params(*all_params)
88     def test_read_and_write_multi(self, hop_size, samplerate, path):
89         try:
90             f = source(path, samplerate, hop_size)
91         except RuntimeError as e:
92             self.skipTest('failed opening with hop_s = {:d}, samplerate = {:d} ({:s})'.format(hop_size, samplerate, str(e)))
93         if samplerate == 0: samplerate = f.samplerate
94         sink_path = get_tmp_sink_path()
95         g = sink(sink_path, samplerate, channels = f.channels)
96         total_frames = 0
97         while True:
98             vec, read = f.do_multi()
99             g.do_multi(vec, read)
100             total_frames += read
101             if read < f.hop_size: break
102         del_tmp_sink_path(sink_path)
103
104     def test_close_file(self):
105         samplerate = 44100
106         sink_path = get_tmp_sink_path()
107         g = sink(sink_path, samplerate)
108         g.close()
109         del_tmp_sink_path(sink_path)
110
111     def test_close_file_twice(self):
112         samplerate = 44100
113         sink_path = get_tmp_sink_path()
114         g = sink(sink_path, samplerate)
115         g.close()
116         g.close()
117         del_tmp_sink_path(sink_path)
118
119 if __name__ == '__main__':
120     main()