python/tests/test_sink.py

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