annotate tests/test_io.py @ 162:ee883f8c372b

Add tests for from_file with large_file=True
author Amine Sehili <amine.sehili@gmail.com>
date Sat, 02 Mar 2019 15:01:05 +0100
parents 682bf4477fae
children 3ddaa5eda8d4
rev   line source
amine@106 1 import os
amine@106 2 import sys
amine@106 3 import math
amine@107 4 from array import array
amine@133 5 from tempfile import NamedTemporaryFile, TemporaryDirectory
amine@110 6 import filecmp
amine@108 7 from unittest import TestCase
amine@108 8 from genty import genty, genty_dataset
amine@157 9 from test_util import _sample_generator, _generate_pure_tone, PURE_TONE_DICT
amine@110 10 from auditok.io import (
amine@126 11 DATA_FORMAT,
amine@121 12 AudioIOError,
amine@110 13 AudioParameterError,
amine@126 14 BufferAudioSource,
amine@162 15 RawAudioSource,
amine@162 16 WaveAudioSource,
amine@110 17 check_audio_data,
amine@143 18 _guess_audio_format,
amine@144 19 _normalize_use_channel,
amine@128 20 _get_audio_parameters,
amine@116 21 _array_to_bytes,
amine@118 22 _mix_audio_channels,
amine@119 23 _extract_selected_channel,
amine@126 24 _load_raw,
amine@129 25 _load_wave,
amine@131 26 _load_with_pydub,
amine@120 27 from_file,
amine@111 28 _save_raw,
amine@110 29 _save_wave,
amine@141 30 _save_with_pydub,
amine@135 31 to_file,
amine@110 32 )
amine@106 33
amine@106 34
amine@106 35 if sys.version_info >= (3, 0):
amine@106 36 PYTHON_3 = True
amine@124 37 from unittest.mock import patch, Mock
amine@106 38 else:
amine@106 39 PYTHON_3 = False
amine@124 40 from mock import patch, Mock
amine@120 41
amine@120 42 AUDIO_PARAMS_SHORT = {"sr": 16000, "sw": 2, "ch": 1}
amine@106 43
amine@106 44
amine@108 45 @genty
amine@108 46 class TestIO(TestCase):
amine@108 47 @genty_dataset(
amine@108 48 valid_mono=(b"\0" * 113, 1, 1),
amine@108 49 valid_stereo=(b"\0" * 160, 1, 2),
amine@108 50 invalid_mono_sw_2=(b"\0" * 113, 2, 1, False),
amine@108 51 invalid_stereo_sw_1=(b"\0" * 113, 1, 2, False),
amine@108 52 invalid_stereo_sw_2=(b"\0" * 158, 2, 2, False),
amine@108 53 )
amine@108 54 def test_check_audio_data(self, data, sample_width, channels, valid=True):
amine@108 55
amine@108 56 if not valid:
amine@108 57 with self.assertRaises(AudioParameterError):
amine@108 58 check_audio_data(data, sample_width, channels)
amine@108 59 else:
amine@108 60 self.assertIsNone(check_audio_data(data, sample_width, channels))
amine@110 61
amine@110 62 @genty_dataset(
amine@143 63 extention_and_format_same=("wav", "filename.wav", "wav"),
amine@143 64 extention_and_format_different=("wav", "filename.mp3", "wav"),
amine@143 65 extention_no_format=(None, "filename.wav", "wav"),
amine@143 66 format_no_extension=("wav", "filename", "wav"),
amine@143 67 no_format_no_extension=(None, "filename", None),
amine@143 68 )
amine@143 69 def test_guess_audio_format(self, fmt, filename, expected):
amine@143 70 result = _guess_audio_format(fmt, filename)
amine@143 71 self.assertEqual(result, expected)
amine@143 72
amine@143 73 @genty_dataset(
amine@144 74 none=(None, 0),
amine@144 75 positive_int=(1, 1),
amine@144 76 negative_int=(-1, -1),
amine@144 77 left=("left", 0),
amine@144 78 right=("right", 1),
amine@144 79 mix=("mix", "mix"),
amine@144 80 )
amine@144 81 def test_normalize_use_channel(self, use_channel, expected):
amine@144 82 result = _normalize_use_channel(use_channel)
amine@144 83 self.assertEqual(result, expected)
amine@144 84
amine@144 85 @genty_dataset(
amine@145 86 simple=((8000, 2, 1, 0), (8000, 2, 1, 0)),
amine@145 87 use_channel_left=((8000, 2, 1, "left"), (8000, 2, 1, 0)),
amine@145 88 use_channel_right=((8000, 2, 1, "right"), (8000, 2, 1, 1)),
amine@145 89 use_channel_mix=((8000, 2, 1, "mix"), (8000, 2, 1, "mix")),
amine@145 90 use_channel_None=((8000, 2, 2, None), (8000, 2, 2, 0)),
amine@145 91 no_use_channel=((8000, 2, 2), (8000, 2, 2, 0)),
amine@145 92 )
amine@145 93 def test_get_audio_parameters_short_params(self, values, expected):
amine@145 94 params = {k: v for k, v in zip(("sr", "sw", "ch", "uc"), values)}
amine@145 95 result = _get_audio_parameters(params)
amine@145 96 self.assertEqual(result, expected)
amine@145 97
amine@145 98 @genty_dataset(
amine@145 99 simple=((8000, 2, 1, 0), (8000, 2, 1, 0)),
amine@145 100 use_channel_left=((8000, 2, 1, "left"), (8000, 2, 1, 0)),
amine@145 101 use_channel_right=((8000, 2, 1, "right"), (8000, 2, 1, 1)),
amine@145 102 use_channel_mix=((8000, 2, 1, "mix"), (8000, 2, 1, "mix")),
amine@145 103 use_channel_None=((8000, 2, 2, None), (8000, 2, 2, 0)),
amine@145 104 no_use_channel=((8000, 2, 2), (8000, 2, 2, 0)),
amine@145 105 )
amine@145 106 def test_get_audio_parameters_long_params(self, values, expected):
amine@145 107 params = {
amine@145 108 k: v
amine@145 109 for k, v in zip(
amine@145 110 ("sampling_rate", "sample_width", "channels", "use_channel"),
amine@145 111 values,
amine@145 112 )
amine@145 113 }
amine@145 114 result = _get_audio_parameters(params)
amine@145 115 self.assertEqual(result, expected)
amine@145 116
amine@145 117 @genty_dataset(simple=((8000, 2, 1, 0), (8000, 2, 1, 0)))
amine@145 118 def test_get_audio_parameters_short_and_long_params(
amine@145 119 self, values, expected
amine@145 120 ):
amine@145 121 params = {
amine@145 122 k: v
amine@145 123 for k, v in zip(
amine@145 124 ("sampling_rate", "sample_width", "channels", "use_channel"),
amine@145 125 values,
amine@145 126 )
amine@145 127 }
amine@145 128
amine@145 129 params.update({k: v for k, v in zip(("sr", "sw", "ch", "uc"), "xxxx")})
amine@145 130 result = _get_audio_parameters(params)
amine@145 131 self.assertEqual(result, expected)
amine@145 132
amine@145 133 @genty_dataset(
amine@146 134 str_sampling_rate=(("x", 2, 1, 0),),
amine@146 135 negative_sampling_rate=((-8000, 2, 1, 0),),
amine@146 136 str_sample_width=((8000, "x", 1, 0),),
amine@146 137 negative_sample_width=((8000, -2, 1, 0),),
amine@146 138 str_channels=((8000, 2, "x", 0),),
amine@146 139 negative_channels=((8000, 2, -1, 0),),
amine@146 140 )
amine@146 141 def test_get_audio_parameters_invalid(self, values):
amine@146 142 params = {
amine@146 143 k: v
amine@146 144 for k, v in zip(
amine@146 145 ("sampling_rate", "sample_width", "channels", "use_channel"),
amine@146 146 values,
amine@146 147 )
amine@146 148 }
amine@146 149 with self.assertRaises(AudioParameterError):
amine@146 150 _get_audio_parameters(params)
amine@146 151
amine@146 152 @genty_dataset(
amine@118 153 mono_1byte=([400], 1),
amine@118 154 stereo_1byte=([400, 600], 1),
amine@118 155 three_channel_1byte=([400, 600, 2400], 1),
amine@118 156 mono_2byte=([400], 2),
amine@118 157 stereo_2byte=([400, 600], 2),
amine@118 158 three_channel_2byte=([400, 600, 1150], 2),
amine@118 159 mono_4byte=([400], 4),
amine@118 160 stereo_4byte=([400, 600], 4),
amine@118 161 four_channel_2byte=([400, 600, 1150, 7220], 4),
amine@118 162 )
amine@118 163 def test_mix_audio_channels(self, frequencies, sample_width):
amine@118 164 sampling_rate = 16000
amine@118 165 sample_width = 2
amine@118 166 channels = len(frequencies)
amine@118 167 mono_channels = [
amine@118 168 _generate_pure_tone(
amine@118 169 freq,
amine@118 170 duration_sec=0.1,
amine@118 171 sampling_rate=sampling_rate,
amine@118 172 sample_width=sample_width,
amine@118 173 )
amine@118 174 for freq in frequencies
amine@118 175 ]
amine@118 176 fmt = DATA_FORMAT[sample_width]
amine@118 177 expected = _array_to_bytes(
amine@118 178 array(
amine@118 179 fmt,
amine@118 180 (sum(samples) // channels for samples in zip(*mono_channels)),
amine@118 181 )
amine@118 182 )
amine@118 183 data = _array_to_bytes(array(fmt, _sample_generator(*mono_channels)))
amine@118 184 mixed = _mix_audio_channels(data, channels, sample_width)
amine@118 185 self.assertEqual(mixed, expected)
amine@118 186
amine@118 187 @genty_dataset(
amine@119 188 mono_1byte=([400], 1, 0),
amine@119 189 stereo_1byte_2st_channel=([400, 600], 1, 1),
amine@119 190 mono_2byte=([400], 2, 0),
amine@119 191 stereo_2byte_1st_channel=([400, 600], 2, 0),
amine@119 192 stereo_2byte_2nd_channel=([400, 600], 2, 1),
amine@119 193 three_channel_2byte_last_negative_idx=([400, 600, 1150], 2, -1),
amine@119 194 three_channel_2byte_2nd_negative_idx=([400, 600, 1150], 2, -2),
amine@119 195 three_channel_2byte_1st_negative_idx=([400, 600, 1150], 2, -3),
amine@119 196 three_channel_4byte_1st=([400, 600, 1150], 4, 0),
amine@119 197 three_channel_4byte_last_negative_idx=([400, 600, 1150], 4, -1),
amine@119 198 )
amine@119 199 def test_extract_selected_channel(
amine@119 200 self, frequencies, sample_width, use_channel
amine@119 201 ):
amine@119 202
amine@119 203 mono_channels = [
amine@119 204 _generate_pure_tone(
amine@119 205 freq,
amine@119 206 duration_sec=0.1,
amine@119 207 sampling_rate=16000,
amine@119 208 sample_width=sample_width,
amine@119 209 )
amine@119 210 for freq in frequencies
amine@119 211 ]
amine@119 212 channels = len(frequencies)
amine@119 213 fmt = DATA_FORMAT[sample_width]
amine@119 214 expected = _array_to_bytes(mono_channels[use_channel])
amine@119 215 data = _array_to_bytes(array(fmt, _sample_generator(*mono_channels)))
amine@119 216 selected_channel = _extract_selected_channel(
amine@119 217 data, channels, sample_width, use_channel
amine@119 218 )
amine@119 219 self.assertEqual(selected_channel, expected)
amine@119 220
amine@148 221 @genty_dataset(mono=([400],), three_channel=([600, 1150, 2400],))
amine@148 222 def test_extract_selected_channel_mix(self, frequencies):
amine@148 223
amine@148 224 mono_channels = [PURE_TONE_DICT[freq] for freq in frequencies]
amine@148 225 channels = len(frequencies)
amine@148 226 fmt = DATA_FORMAT[2]
amine@148 227 expected = _array_to_bytes(
amine@148 228 array(
amine@148 229 fmt,
amine@148 230 (sum(samples) // channels for samples in zip(*mono_channels)),
amine@148 231 )
amine@148 232 )
amine@148 233 data = _array_to_bytes(array(fmt, _sample_generator(*mono_channels)))
amine@148 234 selected_channel = _extract_selected_channel(data, channels, 2, "mix")
amine@148 235 self.assertEqual(selected_channel, expected)
amine@148 236
amine@149 237 @genty_dataset(positive=(2,), negative=(-3,))
amine@149 238 def test_extract_selected_channel_invalid_use_channel(self, use_channel):
amine@149 239 with self.assertRaises(AudioParameterError):
amine@149 240 _extract_selected_channel(b"\0\0", 2, 2, use_channel)
amine@149 241
amine@119 242 @genty_dataset(
amine@120 243 raw_with_audio_format=(
amine@120 244 "audio",
amine@120 245 "raw",
amine@120 246 "_load_raw",
amine@120 247 AUDIO_PARAMS_SHORT,
amine@120 248 ),
amine@120 249 raw_with_extension=(
amine@120 250 "audio.raw",
amine@120 251 None,
amine@120 252 "_load_raw",
amine@120 253 AUDIO_PARAMS_SHORT,
amine@120 254 ),
amine@120 255 wave_with_audio_format=("audio", "wave", "_load_wave"),
amine@120 256 wav_with_audio_format=("audio", "wave", "_load_wave"),
amine@120 257 wav_with_extension=("audio.wav", None, "_load_wave"),
amine@120 258 format_and_extension_both_given=("audio.dat", "wav", "_load_wave"),
amine@120 259 format_and_extension_both_given_b=("audio.raw", "wave", "_load_wave"),
amine@120 260 no_format_nor_extension=("audio", None, "_load_with_pydub"),
amine@120 261 other_formats_ogg=("audio.ogg", None, "_load_with_pydub"),
amine@120 262 other_formats_webm=("audio", "webm", "_load_with_pydub"),
amine@120 263 )
amine@120 264 def test_from_file(
amine@120 265 self, filename, audio_format, funtion_name, kwargs=None
amine@120 266 ):
amine@120 267 funtion_name = "auditok.io." + funtion_name
amine@120 268 if kwargs is None:
amine@120 269 kwargs = {}
amine@120 270 with patch(funtion_name) as patch_function:
amine@120 271 from_file(filename, audio_format, **kwargs)
amine@120 272 self.assertTrue(patch_function.called)
amine@120 273
amine@162 274 def test_from_file_large_file_raw(self, ):
amine@162 275 filename = "tests/data/test_16KHZ_mono_400Hz.raw"
amine@162 276 audio_source = from_file(filename, large_file=True,
amine@162 277 sampling_rate=16000,
amine@162 278 sample_width=2,
amine@162 279 channels=1)
amine@162 280 self.assertIsInstance(audio_source, RawAudioSource)
amine@162 281
amine@162 282 def test_from_file_large_file_wave(self, ):
amine@162 283 filename = "tests/data/test_16KHZ_mono_400Hz.wav"
amine@162 284 audio_source = from_file(filename, large_file=True)
amine@162 285 self.assertIsInstance(audio_source, WaveAudioSource)
amine@162 286
amine@137 287 @genty_dataset(
amine@137 288 missing_sampling_rate=("sr",),
amine@137 289 missing_sample_width=("sw",),
amine@137 290 missing_channels=("ch",),
amine@137 291 )
amine@137 292 def test_from_file_missing_audio_param(self, missing_param):
amine@137 293 with self.assertRaises(AudioParameterError):
amine@137 294 params = AUDIO_PARAMS_SHORT.copy()
amine@137 295 del params[missing_param]
amine@137 296 from_file("audio", audio_format="raw", **params)
amine@137 297
amine@121 298 def test_from_file_no_pydub(self):
amine@121 299 with patch("auditok.io._WITH_PYDUB", False):
amine@121 300 with self.assertRaises(AudioIOError):
amine@121 301 from_file("audio", "mp3")
amine@121 302
amine@111 303 @genty_dataset(
amine@122 304 raw_first_channel=("raw", 0, 400),
amine@122 305 raw_second_channel=("raw", 1, 800),
amine@122 306 raw_third_channel=("raw", 2, 1600),
amine@122 307 raw_left_channel=("raw", "left", 400),
amine@122 308 raw_right_channel=("raw", "right", 800),
amine@122 309 wav_first_channel=("wav", 0, 400),
amine@122 310 wav_second_channel=("wav", 1, 800),
amine@122 311 wav_third_channel=("wav", 2, 1600),
amine@122 312 wav_left_channel=("wav", "left", 400),
amine@122 313 wav_right_channel=("wav", "right", 800),
amine@122 314 )
amine@122 315 def test_from_file_multichannel_audio(
amine@122 316 self, audio_format, use_channel, frequency
amine@122 317 ):
amine@122 318 expected = PURE_TONE_DICT[frequency]
amine@122 319 filename = "tests/data/test_16KHZ_3channel_400-800-1600Hz.{}".format(
amine@122 320 audio_format
amine@122 321 )
amine@122 322 sample_width = 2
amine@122 323 audio_source = from_file(
amine@122 324 filename,
amine@122 325 sampling_rate=16000,
amine@122 326 sample_width=sample_width,
amine@122 327 channels=3,
amine@122 328 use_channel=use_channel,
amine@122 329 )
amine@122 330 fmt = DATA_FORMAT[sample_width]
amine@122 331 data = array(fmt, audio_source._buffer)
amine@122 332 self.assertEqual(data, expected)
amine@122 333
amine@122 334 @genty_dataset(
amine@123 335 raw_mono=("raw", "mono_400Hz", (400,)),
amine@123 336 raw_3channel=("raw", "3channel_400-800-1600Hz", (400, 800, 1600)),
amine@123 337 wav_mono=("wav", "mono_400Hz", (400,)),
amine@123 338 wav_3channel=("wav", "3channel_400-800-1600Hz", (400, 800, 1600)),
amine@123 339 )
amine@123 340 def test_from_file_multichannel_audio_mix(
amine@123 341 self, audio_format, filename_suffix, frequencies
amine@123 342 ):
amine@123 343 sampling_rate = 16000
amine@123 344 sample_width = 2
amine@123 345 channels = len(frequencies)
amine@123 346 mono_channels = [PURE_TONE_DICT[freq] for freq in frequencies]
amine@123 347 channels = len(frequencies)
amine@123 348 fmt = DATA_FORMAT[sample_width]
amine@123 349 expected = _array_to_bytes(
amine@123 350 array(
amine@123 351 fmt,
amine@123 352 (sum(samples) // channels for samples in zip(*mono_channels)),
amine@123 353 )
amine@123 354 )
amine@123 355 filename = "tests/data/test_16KHZ_{}.{}".format(
amine@123 356 filename_suffix, audio_format
amine@123 357 )
amine@123 358 audio_source = from_file(
amine@123 359 filename,
amine@123 360 use_channel="mix",
amine@123 361 sampling_rate=sampling_rate,
amine@123 362 sample_width=2,
amine@123 363 channels=channels,
amine@123 364 )
amine@123 365 mixed = audio_source._buffer
amine@123 366 self.assertEqual((mixed), expected)
amine@123 367
amine@124 368 @patch("auditok.io._WITH_PYDUB", True)
amine@124 369 @patch("auditok.io.BufferAudioSource")
amine@124 370 @genty_dataset(
amine@124 371 ogg_first_channel=("ogg", 0, "from_ogg"),
amine@124 372 ogg_second_channel=("ogg", 1, "from_ogg"),
amine@124 373 ogg_mix=("ogg", "mix", "from_ogg"),
amine@124 374 ogg_default=("ogg", None, "from_ogg"),
amine@124 375 mp3_left_channel=("mp3", "left", "from_mp3"),
amine@124 376 mp3_right_channel=("mp3", "right", "from_mp3"),
amine@124 377 flac_first_channel=("flac", 0, "from_file"),
amine@124 378 flac_second_channel=("flac", 1, "from_file"),
amine@124 379 flv_left_channel=("flv", "left", "from_flv"),
amine@124 380 webm_right_channel=("webm", "right", "from_file"),
amine@124 381 )
amine@124 382 def test_from_file_multichannel_audio_compressed(
amine@124 383 self, audio_format, use_channel, function, *mocks
amine@124 384 ):
amine@124 385 filename = "audio.{}".format(audio_format)
amine@124 386 segment_mock = Mock()
amine@124 387 segment_mock.sample_width = 2
amine@124 388 segment_mock.channels = 2
amine@124 389 segment_mock._data = b"abcd"
amine@124 390 with patch("auditok.io._extract_selected_channel") as ext_mock:
amine@124 391 with patch(
amine@124 392 "auditok.io.AudioSegment.{}".format(function)
amine@124 393 ) as open_func:
amine@124 394 open_func.return_value = segment_mock
amine@124 395 from_file(filename, use_channel=use_channel)
amine@124 396 self.assertTrue(open_func.called)
amine@124 397 self.assertTrue(ext_mock.called)
amine@124 398
amine@124 399 use_channel = {"left": 0, "right": 1, None: 0}.get(
amine@124 400 use_channel, use_channel
amine@124 401 )
amine@124 402 ext_mock.assert_called_with(
amine@124 403 segment_mock._data,
amine@124 404 segment_mock.channels,
amine@124 405 segment_mock.sample_width,
amine@124 406 use_channel,
amine@124 407 )
amine@124 408
amine@124 409 with patch("auditok.io._extract_selected_channel") as ext_mock:
amine@124 410 with patch(
amine@124 411 "auditok.io.AudioSegment.{}".format(function)
amine@124 412 ) as open_func:
amine@124 413 segment_mock.channels = 1
amine@124 414 open_func.return_value = segment_mock
amine@124 415 from_file(filename, use_channel=use_channel)
amine@124 416 self.assertTrue(open_func.called)
amine@124 417 self.assertFalse(ext_mock.called)
amine@124 418
amine@125 419 @patch("auditok.io._WITH_PYDUB", True)
amine@125 420 @patch("auditok.io.BufferAudioSource")
amine@125 421 @genty_dataset(
amine@125 422 ogg=("ogg", "from_ogg"),
amine@125 423 mp3=("mp3", "from_mp3"),
amine@125 424 flac=("flac", "from_file"),
amine@125 425 )
amine@125 426 def test_from_file_multichannel_audio_mix_compressed(
amine@125 427 self, audio_format, function, *mocks
amine@125 428 ):
amine@125 429 filename = "audio.{}".format(audio_format)
amine@125 430 segment_mock = Mock()
amine@125 431 segment_mock.sample_width = 2
amine@125 432 segment_mock.channels = 2
amine@125 433 segment_mock._data = b"abcd"
amine@125 434 with patch("auditok.io._mix_audio_channels") as mix_mock:
amine@125 435 with patch(
amine@125 436 "auditok.io.AudioSegment.{}".format(function)
amine@125 437 ) as open_func:
amine@125 438 open_func.return_value = segment_mock
amine@125 439 from_file(filename, use_channel="mix")
amine@125 440 self.assertTrue(open_func.called)
amine@125 441 mix_mock.assert_called_with(
amine@125 442 segment_mock._data,
amine@125 443 segment_mock.channels,
amine@125 444 segment_mock.sample_width,
amine@125 445 )
amine@125 446
amine@123 447 @genty_dataset(
amine@126 448 dafault_first_channel=(None, 400),
amine@126 449 first_channel=(0, 400),
amine@126 450 second_channel=(1, 800),
amine@126 451 third_channel=(2, 1600),
amine@126 452 negative_first_channel=(-3, 400),
amine@126 453 negative_second_channel=(-2, 800),
amine@126 454 negative_third_channel=(-1, 1600),
amine@126 455 )
amine@126 456 def test_load_raw(self, use_channel, frequency):
amine@126 457 filename = "tests/data/test_16KHZ_3channel_400-800-1600Hz.raw"
amine@126 458 if use_channel is not None:
amine@126 459 audio_source = _load_raw(
amine@126 460 filename,
amine@126 461 sampling_rate=16000,
amine@126 462 sample_width=2,
amine@126 463 channels=3,
amine@126 464 use_channel=use_channel,
amine@126 465 )
amine@126 466 else:
amine@126 467 audio_source = _load_raw(
amine@126 468 filename, sampling_rate=16000, sample_width=2, channels=3
amine@126 469 )
amine@126 470 self.assertIsInstance(audio_source, BufferAudioSource)
amine@126 471 self.assertEqual(audio_source.sampling_rate, 16000)
amine@126 472 self.assertEqual(audio_source.sample_width, 2)
amine@126 473 self.assertEqual(audio_source.channels, 1)
amine@126 474 # generate a pure sine wave tone of the given frequency
amine@126 475 expected = PURE_TONE_DICT[frequency]
amine@126 476 # compre with data read from file
amine@126 477 fmt = DATA_FORMAT[2]
amine@126 478 data = array(fmt, audio_source._buffer)
amine@126 479 self.assertEqual(data, expected)
amine@126 480
amine@126 481 @genty_dataset(
amine@127 482 mono=("mono_400Hz", (400,)),
amine@127 483 three_channel=("3channel_400-800-1600Hz", (400, 800, 1600)),
amine@127 484 )
amine@127 485 def test_load_raw_mix(self, filename_suffix, frequencies):
amine@127 486 sampling_rate = 16000
amine@127 487 sample_width = 2
amine@127 488 channels = len(frequencies)
amine@127 489 mono_channels = [PURE_TONE_DICT[freq] for freq in frequencies]
amine@127 490
amine@127 491 fmt = DATA_FORMAT[sample_width]
amine@127 492 expected = _array_to_bytes(
amine@127 493 array(
amine@127 494 fmt,
amine@127 495 (sum(samples) // channels for samples in zip(*mono_channels)),
amine@127 496 )
amine@127 497 )
amine@127 498 filename = "tests/data/test_16KHZ_{}.raw".format(filename_suffix)
amine@127 499 audio_source = _load_raw(
amine@127 500 filename,
amine@127 501 use_channel="mix",
amine@127 502 sampling_rate=sampling_rate,
amine@127 503 sample_width=2,
amine@127 504 channels=channels,
amine@127 505 )
amine@127 506 mixed = audio_source._buffer
amine@127 507 self.assertEqual(mixed, expected)
amine@127 508 self.assertIsInstance(audio_source, BufferAudioSource)
amine@127 509 self.assertEqual(audio_source.sampling_rate, sampling_rate)
amine@127 510 self.assertEqual(audio_source.sample_width, sample_width)
amine@127 511 self.assertEqual(audio_source.channels, 1)
amine@127 512
amine@127 513 @genty_dataset(
amine@128 514 missing_sampling_rate=("sr",),
amine@128 515 missing_sample_width=("sw",),
amine@128 516 missing_channels=("ch",),
amine@128 517 )
amine@128 518 def test_load_raw_missing_audio_param(self, missing_param):
amine@128 519 with self.assertRaises(AudioParameterError):
amine@128 520 params = AUDIO_PARAMS_SHORT.copy()
amine@128 521 del params[missing_param]
amine@128 522 srate, swidth, channels, _ = _get_audio_parameters(params)
amine@128 523 _load_raw("audio", srate, swidth, channels)
amine@128 524
amine@128 525 @genty_dataset(
amine@129 526 dafault_first_channel=(None, 400),
amine@129 527 first_channel=(0, 400),
amine@129 528 second_channel=(1, 800),
amine@129 529 third_channel=(2, 1600),
amine@129 530 negative_first_channel=(-3, 400),
amine@129 531 negative_second_channel=(-2, 800),
amine@129 532 negative_third_channel=(-1, 1600),
amine@129 533 )
amine@129 534 def test_load_wave(self, use_channel, frequency):
amine@129 535 filename = "tests/data/test_16KHZ_3channel_400-800-1600Hz.wav"
amine@129 536 if use_channel is not None:
amine@129 537 audio_source = _load_wave(filename, use_channel=use_channel)
amine@129 538 else:
amine@129 539 audio_source = _load_wave(filename)
amine@129 540 self.assertIsInstance(audio_source, BufferAudioSource)
amine@129 541 self.assertEqual(audio_source.sampling_rate, 16000)
amine@129 542 self.assertEqual(audio_source.sample_width, 2)
amine@129 543 self.assertEqual(audio_source.channels, 1)
amine@129 544 # generate a pure sine wave tone of the given frequency
amine@129 545 expected = PURE_TONE_DICT[frequency]
amine@129 546 # compre with data read from file
amine@129 547 fmt = DATA_FORMAT[2]
amine@129 548 data = array(fmt, audio_source._buffer)
amine@129 549 self.assertEqual(data, expected)
amine@129 550
amine@129 551 @genty_dataset(
amine@130 552 mono=("mono_400Hz", (400,)),
amine@130 553 three_channel=("3channel_400-800-1600Hz", (400, 800, 1600)),
amine@130 554 )
amine@130 555 def test_load_wave_mix(self, filename_suffix, frequencies):
amine@130 556 sampling_rate = 16000
amine@130 557 sample_width = 2
amine@130 558 channels = len(frequencies)
amine@130 559 mono_channels = [PURE_TONE_DICT[freq] for freq in frequencies]
amine@130 560 fmt = DATA_FORMAT[sample_width]
amine@130 561 expected = _array_to_bytes(
amine@130 562 array(
amine@130 563 fmt,
amine@130 564 (sum(samples) // channels for samples in zip(*mono_channels)),
amine@130 565 )
amine@130 566 )
amine@130 567 filename = "tests/data/test_16KHZ_{}.wav".format(filename_suffix)
amine@130 568 audio_source = _load_wave(filename, use_channel="mix")
amine@130 569 mixed = audio_source._buffer
amine@130 570 self.assertEqual(mixed, expected)
amine@130 571 self.assertIsInstance(audio_source, BufferAudioSource)
amine@130 572 self.assertEqual(audio_source.sampling_rate, sampling_rate)
amine@130 573 self.assertEqual(audio_source.sample_width, sample_width)
amine@130 574 self.assertEqual(audio_source.channels, 1)
amine@130 575
amine@131 576 @patch("auditok.io._WITH_PYDUB", True)
amine@131 577 @patch("auditok.io.BufferAudioSource")
amine@131 578 @genty_dataset(
amine@131 579 ogg_default_first_channel=("ogg", 2, None, "from_ogg"),
amine@131 580 ogg_first_channel=("ogg", 1, 0, "from_ogg"),
amine@131 581 ogg_second_channel=("ogg", 2, 1, "from_ogg"),
amine@131 582 ogg_mix_channels=("ogg", 3, "mix", "from_ogg"),
amine@131 583 mp3_left_channel=("mp3", 1, "left", "from_mp3"),
amine@131 584 mp3_right_channel=("mp3", 2, "right", "from_mp3"),
amine@131 585 mp3_mix_channels=("mp3", 3, "mix", "from_mp3"),
amine@131 586 flac_first_channel=("flac", 2, 0, "from_file"),
amine@131 587 flac_second_channel=("flac", 2, 1, "from_file"),
amine@131 588 flv_left_channel=("flv", 1, "left", "from_flv"),
amine@131 589 webm_right_channel=("webm", 2, "right", "from_file"),
amine@131 590 webm_mix_channels=("webm", 4, "mix", "from_file"),
amine@131 591 )
amine@131 592 def test_load_with_pydub(
amine@131 593 self, audio_format, channels, use_channel, function, *mocks
amine@131 594 ):
amine@131 595 filename = "audio.{}".format(audio_format)
amine@131 596 segment_mock = Mock()
amine@131 597 segment_mock.sample_width = 2
amine@131 598 segment_mock.channels = channels
amine@131 599 segment_mock._data = b"abcdefgh"
amine@131 600 with patch("auditok.io._extract_selected_channel") as ext_mock:
amine@131 601 with patch(
amine@131 602 "auditok.io.AudioSegment.{}".format(function)
amine@131 603 ) as open_func:
amine@131 604 open_func.return_value = segment_mock
amine@131 605 use_channel = {"left": 0, "right": 1, None: 0}.get(
amine@131 606 use_channel, use_channel
amine@131 607 )
amine@131 608 _load_with_pydub(filename, audio_format, use_channel)
amine@131 609 self.assertTrue(open_func.called)
amine@131 610 if channels > 1:
amine@131 611 self.assertTrue(ext_mock.called)
amine@131 612 ext_mock.assert_called_with(
amine@131 613 segment_mock._data,
amine@131 614 segment_mock.channels,
amine@131 615 segment_mock.sample_width,
amine@131 616 use_channel,
amine@131 617 )
amine@131 618 else:
amine@131 619 self.assertFalse(ext_mock.called)
amine@131 620
amine@130 621 @genty_dataset(
amine@132 622 mono=("mono_400Hz.raw", (400,)),
amine@132 623 three_channel=("3channel_400-800-1600Hz.raw", (400, 800, 1600)),
amine@132 624 )
amine@132 625 def test_save_raw(self, filename, frequencies):
amine@132 626 filename = "tests/data/test_16KHZ_{}".format(filename)
amine@132 627 sample_width = 2
amine@132 628 fmt = DATA_FORMAT[sample_width]
amine@132 629 mono_channels = [PURE_TONE_DICT[freq] for freq in frequencies]
amine@132 630 data = _array_to_bytes(array(fmt, _sample_generator(*mono_channels)))
amine@132 631 tmpfile = NamedTemporaryFile()
amine@136 632 _save_raw(data, tmpfile.name)
amine@132 633 self.assertTrue(filecmp.cmp(tmpfile.name, filename, shallow=False))
amine@132 634
amine@132 635 @genty_dataset(
amine@110 636 mono=("mono_400Hz.wav", (400,)),
amine@110 637 three_channel=("3channel_400-800-1600Hz.wav", (400, 800, 1600)),
amine@110 638 )
amine@110 639 def test_save_wave(self, filename, frequencies):
amine@110 640 filename = "tests/data/test_16KHZ_{}".format(filename)
amine@110 641 sampling_rate = 16000
amine@110 642 sample_width = 2
amine@110 643 channels = len(frequencies)
amine@110 644 fmt = DATA_FORMAT[sample_width]
amine@110 645 mono_channels = [PURE_TONE_DICT[freq] for freq in frequencies]
amine@110 646 data = _array_to_bytes(array(fmt, _sample_generator(*mono_channels)))
amine@110 647 tmpfile = NamedTemporaryFile()
amine@136 648 _save_wave(data, tmpfile.name, sampling_rate, sample_width, channels)
amine@110 649 self.assertTrue(filecmp.cmp(tmpfile.name, filename, shallow=False))
amine@132 650
amine@132 651 @genty_dataset(
amine@132 652 missing_sampling_rate=("sr",),
amine@132 653 missing_sample_width=("sw",),
amine@132 654 missing_channels=("ch",),
amine@132 655 )
amine@132 656 def test_save_wave_missing_audio_param(self, missing_param):
amine@132 657 with self.assertRaises(AudioParameterError):
amine@132 658 params = AUDIO_PARAMS_SHORT.copy()
amine@132 659 del params[missing_param]
amine@132 660 srate, swidth, channels, _ = _get_audio_parameters(params)
amine@136 661 _save_wave(b"\0\0", "audio", srate, swidth, channels)
amine@133 662
amine@141 663 def test_save_with_pydub(self):
amine@141 664 with patch("auditok.io.AudioSegment.export") as export:
amine@142 665 tmpdir = TemporaryDirectory()
amine@142 666 filename = os.path.join(tmpdir.name, "audio.ogg")
amine@142 667 _save_with_pydub(b"\0\0", filename, "ogg", 16000, 2, 1)
amine@141 668 self.assertTrue(export.called)
amine@142 669 tmpdir.cleanup()
amine@141 670
amine@133 671 @genty_dataset(
amine@133 672 raw_with_audio_format=("audio", "raw"),
amine@133 673 raw_with_extension=("audio.raw", None),
amine@133 674 raw_with_audio_format_and_extension=("audio.mp3", "raw"),
amine@133 675 raw_no_audio_format_nor_extension=("audio", None),
amine@133 676 )
amine@133 677 def test_to_file_raw(self, filename, audio_format):
amine@133 678 exp_filename = "tests/data/test_16KHZ_mono_400Hz.raw"
amine@133 679 tmpdir = TemporaryDirectory()
amine@133 680 filename = os.path.join(tmpdir.name, filename)
amine@133 681 data = _array_to_bytes(PURE_TONE_DICT[400])
amine@135 682 to_file(data, filename, audio_format=audio_format)
amine@133 683 self.assertTrue(filecmp.cmp(filename, exp_filename, shallow=False))
amine@133 684 tmpdir.cleanup()
amine@134 685
amine@134 686 @genty_dataset(
amine@134 687 wav_with_audio_format=("audio", "wav"),
amine@134 688 wav_with_extension=("audio.wav", None),
amine@134 689 wav_with_audio_format_and_extension=("audio.mp3", "wav"),
amine@134 690 wave_with_audio_format=("audio", "wave"),
amine@134 691 wave_with_extension=("audio.wave", None),
amine@134 692 wave_with_audio_format_and_extension=("audio.mp3", "wave"),
amine@134 693 )
amine@135 694 def test_to_file_wave(self, filename, audio_format):
amine@134 695 exp_filename = "tests/data/test_16KHZ_mono_400Hz.wav"
amine@134 696 tmpdir = TemporaryDirectory()
amine@134 697 filename = os.path.join(tmpdir.name, filename)
amine@134 698 data = _array_to_bytes(PURE_TONE_DICT[400])
amine@135 699 to_file(
amine@135 700 data,
amine@135 701 filename,
amine@135 702 audio_format=audio_format,
amine@135 703 sampling_rate=16000,
amine@135 704 sample_width=2,
amine@135 705 channels=1,
amine@134 706 )
amine@134 707 self.assertTrue(filecmp.cmp(filename, exp_filename, shallow=False))
amine@134 708 tmpdir.cleanup()
amine@138 709
amine@138 710 @genty_dataset(
amine@138 711 missing_sampling_rate=("sr",),
amine@138 712 missing_sample_width=("sw",),
amine@138 713 missing_channels=("ch",),
amine@138 714 )
amine@138 715 def test_to_file_missing_audio_param(self, missing_param):
amine@138 716 params = AUDIO_PARAMS_SHORT.copy()
amine@138 717 del params[missing_param]
amine@138 718 with self.assertRaises(AudioParameterError):
amine@138 719 to_file(b"\0\0", "audio", audio_format="wav", **params)
amine@138 720 with self.assertRaises(AudioParameterError):
amine@138 721 to_file(b"\0\0", "audio", audio_format="mp3", **params)
amine@139 722
amine@139 723 def test_to_file_no_pydub(self):
amine@139 724 with patch("auditok.io._WITH_PYDUB", False):
amine@139 725 with self.assertRaises(AudioIOError):
amine@139 726 to_file("audio", b"", "mp3")
amine@140 727
amine@140 728 @patch("auditok.io._WITH_PYDUB", True)
amine@140 729 @genty_dataset(
amine@140 730 ogg_with_extension=("audio.ogg", None),
amine@140 731 ogg_with_audio_format=("audio", "ogg"),
amine@140 732 ogg_format_with_wrong_extension=("audio.wav", "ogg"),
amine@140 733 )
amine@140 734 def test_to_file_compressed(self, filename, audio_format, *mocks):
amine@140 735 with patch("auditok.io.AudioSegment.export") as export:
amine@142 736 tmpdir = TemporaryDirectory()
amine@142 737 filename = os.path.join(tmpdir.name, filename)
amine@140 738 to_file(b"\0\0", filename, audio_format, **AUDIO_PARAMS_SHORT)
amine@140 739 self.assertTrue(export.called)
amine@142 740 tmpdir.cleanup()