source: python/tests/test_sink.py @ 1c6fe64

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

python/tests/test_sink.py: add read_write_multi

  • Property mode set to 100755
File size: 4.3 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 test_many_sinks(self):
16        from tempfile import mkdtemp
17        import os.path
18        import shutil
19        tmpdir = mkdtemp()
20        sink_list = []
21        for i in range(many_files):
22            path = os.path.join(tmpdir, 'f-' + str(i) + '.wav')
23            g = sink(path, 0)
24            sink_list.append(g)
25            write = 32
26            for n in range(200):
27                vec = fvec(write)
28                g(vec, write)
29            g.close()
30        shutil.rmtree(tmpdir)
31
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 = []
38        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)
54
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)
106
107    def test_close_file(self):
108        samplerate = 44100
109        sink_path = get_tmp_sink_path()
110        g = sink(sink_path, samplerate)
111        g.close()
112        del_tmp_sink_path(sink_path)
113
114    def test_close_file_twice(self):
115        samplerate = 44100
116        sink_path = get_tmp_sink_path()
117        g = sink(sink_path, samplerate)
118        g.close()
119        g.close()
120        del_tmp_sink_path(sink_path)
121
122if __name__ == '__main__':
123    from unittest import main
124    main()
Note: See TracBrowser for help on using the repository browser.