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