1 | #! /usr/bin/env python |
---|
2 | |
---|
3 | from numpy.testing import TestCase, assert_equal, assert_almost_equal |
---|
4 | from aubio import fvec, source, sink |
---|
5 | from numpy import array |
---|
6 | from utils import list_all_sounds, get_tmp_sink_path, del_tmp_sink_path |
---|
7 | |
---|
8 | list_of_sounds = list_all_sounds('sounds') |
---|
9 | path = None |
---|
10 | |
---|
11 | many_files = 300 # 256 opened files is too much |
---|
12 | |
---|
13 | class aubio_sink_test_case(TestCase): |
---|
14 | |
---|
15 | def setUp(self): |
---|
16 | if not len(list_of_sounds): |
---|
17 | self.skipTest('add some sound files in \'python/tests/sounds\'') |
---|
18 | |
---|
19 | def test_many_sinks(self): |
---|
20 | from tempfile import mkdtemp |
---|
21 | import os.path |
---|
22 | import shutil |
---|
23 | tmpdir = mkdtemp() |
---|
24 | sink_list = [] |
---|
25 | for i in range(many_files): |
---|
26 | path = os.path.join(tmpdir, 'f-' + str(i) + '.wav') |
---|
27 | g = sink(path, 0) |
---|
28 | sink_list.append(g) |
---|
29 | write = 32 |
---|
30 | for n in range(200): |
---|
31 | vec = fvec(write) |
---|
32 | g(vec, write) |
---|
33 | g.close() |
---|
34 | shutil.rmtree(tmpdir) |
---|
35 | |
---|
36 | def test_many_sinks_not_closed(self): |
---|
37 | from tempfile import mkdtemp |
---|
38 | import os.path |
---|
39 | import shutil |
---|
40 | tmpdir = mkdtemp() |
---|
41 | sink_list = [] |
---|
42 | try: |
---|
43 | for i in range(many_files): |
---|
44 | path = os.path.join(tmpdir, 'f-' + str(i) + '.wav') |
---|
45 | g = sink(path, 0) |
---|
46 | sink_list.append(g) |
---|
47 | write = 256 |
---|
48 | for n in range(200): |
---|
49 | vec = fvec(write) |
---|
50 | g(vec, write) |
---|
51 | except StandardError: |
---|
52 | pass |
---|
53 | else: |
---|
54 | self.fail("does not fail on too many files open") |
---|
55 | for g in sink_list: |
---|
56 | g.close() |
---|
57 | shutil.rmtree(tmpdir) |
---|
58 | |
---|
59 | def test_read(self): |
---|
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_close_file(self): |
---|
81 | samplerate = 44100 |
---|
82 | sink_path = get_tmp_sink_path() |
---|
83 | g = sink(sink_path, samplerate) |
---|
84 | g.close() |
---|
85 | del_tmp_sink_path(sink_path) |
---|
86 | |
---|
87 | def test_close_file_twice(self): |
---|
88 | samplerate = 44100 |
---|
89 | sink_path = get_tmp_sink_path() |
---|
90 | g = sink(sink_path, samplerate) |
---|
91 | g.close() |
---|
92 | g.close() |
---|
93 | del_tmp_sink_path(sink_path) |
---|
94 | |
---|
95 | if __name__ == '__main__': |
---|
96 | from unittest import main |
---|
97 | main() |
---|