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 list_of_sounds = list_all_sounds('sounds')
  10 samplerates = [0, 44100, 8000, 32000]
  11 hop_sizes = [512, 1024, 64]
  12
  13 path = None
  14
  15 many_files = 300 # 256 opened files is too much
  16
  17 all_params = []
  18 for soundfile in list_of_sounds:
  19     for hop_size in hop_sizes:
  20         for samplerate in samplerates:
  21             all_params.append((hop_size, samplerate, soundfile))
  22
  23 class aubio_sink_test_case(TestCase):
  24
  25     def setUp(self):
  26         if not len(list_of_sounds):
  27             self.skipTest('add some sound files in \'python/tests/sounds\'')
  28
  29     def test_many_sinks(self):
  30         from tempfile import mkdtemp
  31         import os.path
  32         import shutil
  33         tmpdir = mkdtemp()
  34         sink_list = []
  35         for i in range(many_files):
  36             path = os.path.join(tmpdir, 'f-' + str(i) + '.wav')
  37             g = sink(path, 0)
  38             sink_list.append(g)
  39             write = 32
  40             for _ in range(200):
  41                 vec = fvec(write)
  42                 g(vec, write)
  43             g.close()
  44         shutil.rmtree(tmpdir)
  45
  46     @params(*all_params)
  47     def test_read_and_write(self, hop_size, samplerate, path):
  48
  49         try:
  50             f = source(path, samplerate, hop_size)
  51         except RuntimeError as e:
  52             self.skipTest('failed opening with hop_s = {:d}, samplerate = {:d} ({:s})'.format(hop_size, samplerate, str(e)))
  53         if samplerate == 0: samplerate = f.samplerate
  54         sink_path = get_tmp_sink_path()
  55         g = sink(sink_path, samplerate)
  56         total_frames = 0
  57         while True:
  58             vec, read = f()
  59             g(vec, read)
  60             total_frames += read
  61             if read < f.hop_size: break
  62         del_tmp_sink_path(sink_path)
  63
  64     @params(*all_params)
  65     def test_read_and_write_multi(self, hop_size, samplerate, path):
  66         try:
  67             f = source(path, samplerate, hop_size)
  68         except RuntimeError as e:
  69             self.skipTest('failed opening with hop_s = {:d}, samplerate = {:d} ({:s})'.format(hop_size, samplerate, str(e)))
  70         if samplerate == 0: samplerate = f.samplerate
  71         sink_path = get_tmp_sink_path()
  72         g = sink(sink_path, samplerate, channels = f.channels)
  73         total_frames = 0
  74         while True:
  75             vec, read = f.do_multi()
  76             g.do_multi(vec, read)
  77             total_frames += read
  78             if read < f.hop_size: break
  79         del_tmp_sink_path(sink_path)
  80
  81     def test_close_file(self):
  82         samplerate = 44100
  83         sink_path = get_tmp_sink_path()
  84         g = sink(sink_path, samplerate)
  85         g.close()
  86         del_tmp_sink_path(sink_path)
  87
  88     def test_close_file_twice(self):
  89         samplerate = 44100
  90         sink_path = get_tmp_sink_path()
  91         g = sink(sink_path, samplerate)
  92         g.close()
  93         g.close()
  94         del_tmp_sink_path(sink_path)
  95
  96 if __name__ == '__main__':
  97     main()