source: python/tests/test_sink.py @ ab35262

feature/autosinkfeature/cnnfeature/cnn_orgfeature/constantqfeature/crepefeature/crepe_orgfeature/pitchshiftfeature/pydocstringsfeature/timestretchfix/ffmpeg5pitchshiftsamplertimestretchyinfft+
Last change on this file since ab35262 was ab35262, checked in by Paul Brossier <piem@piem.org>, 10 years ago

python/tests/test_sink.py: add more tests, clean up created files

  • Property mode set to 100755
File size: 3.1 KB
Line 
1#! /usr/bin/env python
2
3from numpy.testing import TestCase, assert_equal, assert_almost_equal
4from aubio import fvec, source, sink
5from numpy import array
6from utils import list_all_sounds, get_tmp_sink_path, del_tmp_sink_path
7
8list_of_sounds = list_all_sounds('sounds')
9path = None
10
11many_files = 300 # 256 opened files is too much
12
13class 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
95if __name__ == '__main__':
96    from unittest import main
97    main()
Note: See TracBrowser for help on using the repository browser.