python/tests/test_source.py

   1 #! /usr/bin/env python
   2
   3
   4 from numpy.testing import TestCase, assert_equal
   5 from aubio import source
   6 from utils import list_all_sounds
   7 import unittest
   8 from _tools import parametrize, assert_raises, assert_equal, skipTest
   9
  10 list_of_sounds = list_all_sounds('sounds')
  11 samplerates = [0, 44100, 8000, 32000]
  12 hop_sizes = [512, 1024, 64]
  13
  14 default_test_sound = len(list_of_sounds) and list_of_sounds[0] or None
  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 no_sounds_msg = "no test sounds, add some in 'python/tests/sounds/'!"
  23
  24 _debug = False
  25
  26 class Test_aubio_source_test_case(TestCase):
  27
  28     def setUp(self):
  29         if not default_test_sound:
  30             skipTest(no_sounds_msg)
  31
  32     def test_close_file(self):
  33         samplerate = 0 # use native samplerate
  34         hop_size = 256
  35         f = source(default_test_sound, samplerate, hop_size)
  36         f.close()
  37
  38     def test_close_file_twice(self):
  39         samplerate = 0 # use native samplerate
  40         hop_size = 256
  41         f = source(default_test_sound, samplerate, hop_size)
  42         f.close()
  43         f.close()
  44
  45     def test_read_after_close(self):
  46         samplerate = 0 # use native samplerate
  47         hop_size = 256
  48         f = source(default_test_sound, samplerate, hop_size)
  49         read, frames = f()
  50         f.close()
  51         with assert_raises(RuntimeError):
  52             read, frames = f()
  53         with assert_raises(RuntimeError):
  54             read, frames = f.do_multi()
  55
  56
  57 class Test_aubio_source_read(object):
  58
  59     def read_from_source(self, f):
  60         total_frames = 0
  61         while True:
  62             samples , read = f()
  63             total_frames += read
  64             if read < f.hop_size:
  65                 assert_equal(samples[read:], 0)
  66                 break
  67         if _debug:
  68             result_str = "read {:.2f}s ({:d} frames"
  69             result_str += " in {:d} blocks at {:d}Hz) from {:s}"
  70             result_params = total_frames / float(f.samplerate), total_frames, \
  71                     total_frames//f.hop_size, f.samplerate, f.uri
  72             print (result_str.format(*result_params))
  73         return total_frames
  74
  75     @parametrize('hop_size, samplerate, soundfile', all_params)
  76     def test_samplerate_hopsize(self, hop_size, samplerate, soundfile):
  77         try:
  78             f = source(soundfile, samplerate, hop_size)
  79         except RuntimeError as e:
  80             err_msg = 'failed opening with hop_s={:d}, samplerate={:d} ({:s})'
  81             skipTest(err_msg.format(hop_size, samplerate, str(e)))
  82         assert f.samplerate != 0
  83         read_frames = self.read_from_source(f)
  84         if 'f_' in soundfile and samplerate == 0:
  85             import re
  86             f = re.compile(r'.*_\([0:9]*f\)_.*')
  87             match_f = re.findall('([0-9]*)f_', soundfile)
  88             if len(match_f) == 1:
  89                 expected_frames = int(match_f[0])
  90                 assert_equal(expected_frames, read_frames)
  91
  92     @parametrize('p', list_of_sounds)
  93     def test_samplerate_none(self, p):
  94         f = source(p)
  95         assert f.samplerate != 0
  96         self.read_from_source(f)
  97
  98     @parametrize('p', list_of_sounds)
  99     def test_samplerate_0(self, p):
 100         f = source(p, 0)
 101         assert f.samplerate != 0
 102         self.read_from_source(f)
 103
 104     @parametrize('p', list_of_sounds)
 105     def test_zero_hop_size(self, p):
 106         f = source(p, 0, 0)
 107         assert f.samplerate != 0
 108         assert f.hop_size != 0
 109         self.read_from_source(f)
 110
 111     @parametrize('p', list_of_sounds)
 112     def test_seek_to_half(self, p):
 113         from random import randint
 114         f = source(p, 0, 0)
 115         assert f.samplerate != 0
 116         assert f.hop_size != 0
 117         a = self.read_from_source(f)
 118         c = randint(0, a)
 119         f.seek(c)
 120         b = self.read_from_source(f)
 121         assert a == b + c
 122
 123     @parametrize('p', list_of_sounds)
 124     def test_duration(self, p):
 125         total_frames = 0
 126         f = source(p)
 127         duration = f.duration
 128         while True:
 129             _, read = f()
 130             total_frames += read
 131             if read < f.hop_size: break
 132         assert_equal (duration, total_frames)
 133
 134
 135 class Test_aubio_source_wrong_params(object):
 136
 137     def test_wrong_file(self):
 138         with assert_raises(RuntimeError):
 139             source('path_to/unexisting file.mp3')
 140
 141 @unittest.skipIf(default_test_sound is None, no_sounds_msg)
 142 class Test_aubio_source_wrong_params_with_file(TestCase):
 143
 144     def test_wrong_samplerate(self):
 145         with assert_raises(ValueError):
 146             source(default_test_sound, -1)
 147
 148     def test_wrong_hop_size(self):
 149         with assert_raises(ValueError):
 150             source(default_test_sound, 0, -1)
 151
 152     def test_wrong_channels(self):
 153         with assert_raises(ValueError):
 154             source(default_test_sound, 0, 0, -1)
 155
 156     def test_wrong_seek(self):
 157         f = source(default_test_sound)
 158         with assert_raises(ValueError):
 159             f.seek(-1)
 160
 161     def test_wrong_seek_too_large(self):
 162         f = source(default_test_sound)
 163         try:
 164             with assert_raises(ValueError):
 165                 f.seek(f.duration + f.samplerate * 10)
 166         except:
 167             skipTest('seeking after end of stream failed raising ValueError')
 168
 169 class Test_aubio_source_readmulti(Test_aubio_source_read):
 170
 171     def read_from_source(self, f):
 172         total_frames = 0
 173         while True:
 174             samples, read = f.do_multi()
 175             total_frames += read
 176             if read < f.hop_size:
 177                 assert_equal(samples[:,read:], 0)
 178                 break
 179         if _debug:
 180             result_str = "read {:.2f}s ({:d} frames in {:d} channels"
 181             result_str += " and {:d} blocks at {:d}Hz) from {:s}"
 182             result_params = total_frames / float(f.samplerate), total_frames, \
 183                     f.channels, int(total_frames/f.hop_size), \
 184                     f.samplerate, f.uri
 185             print (result_str.format(*result_params))
 186         return total_frames
 187
 188 class Test_aubio_source_with(object):
 189
 190     @parametrize('filename', list_of_sounds)
 191     def test_read_from_mono(self, filename):
 192         total_frames = 0
 193         hop_size = 2048
 194         with source(filename, 0, hop_size) as input_source:
 195             assert_equal(input_source.hop_size, hop_size)
 196             #assert_equal(input_source.samplerate, samplerate)
 197             total_frames = 0
 198             for frames in input_source:
 199                 total_frames += frames.shape[-1]
 200             # check we read as many samples as we expected
 201             assert_equal(total_frames, input_source.duration)
 202
 203 if __name__ == '__main__':
 204     from _tools import run_module_suite
 205     run_module_suite()