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