python/tests/test_sink.py

   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(object):
  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     from _tools import run_module_suite
 122     run_module_suite()