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