annotate tests/test_AudioDataSource.py @ 208:29472a5a798a

Harmonize calls to _extract_selected_channel
author Amine Sehili <amine.sehili@gmail.com>
date Fri, 14 Jun 2019 22:40:04 +0100
parents f35c0a18f4cb
children 3708f2eb8a40
rev   line source
amine@173 1 """
amine@172 2 @author: Amine Sehili <amine.sehili@gmail.com>
amine@172 3 September 2015
amine@172 4
amine@173 5 """
amine@172 6
amine@172 7 import unittest
amine@172 8 from functools import partial
amine@172 9 import sys
amine@173 10 from auditok import (
amine@173 11 dataset,
amine@173 12 ADSFactory,
amine@173 13 AudioDataSource,
amine@173 14 BufferAudioSource,
amine@173 15 WaveAudioSource,
amine@173 16 DuplicateArgument,
amine@173 17 )
amine@172 18 import wave
amine@172 19
amine@172 20
amine@172 21 try:
amine@172 22 from builtins import range
amine@172 23 except ImportError:
amine@172 24 if sys.version_info < (3, 0):
amine@172 25 range = xrange
amine@172 26
amine@173 27
amine@172 28 class TestADSFactoryFileAudioSource(unittest.TestCase):
amine@172 29 def setUp(self):
amine@173 30 self.audio_source = WaveAudioSource(
amine@173 31 filename=dataset.one_to_six_arabic_16000_mono_bc_noise
amine@173 32 )
amine@173 33
amine@172 34 def test_ADS_type(self):
amine@173 35
amine@172 36 ads = ADSFactory.ads(audio_source=self.audio_source)
amine@173 37
amine@173 38 self.assertIsInstance(
amine@173 39 ads,
amine@173 40 AudioDataSource,
amine@173 41 msg="wrong type for ads object, expected: 'AudioDataSource', found: {0}".format(
amine@173 42 type(ads)
amine@173 43 ),
amine@173 44 )
amine@173 45
amine@172 46 def test_default_block_size(self):
amine@172 47 ads = ADSFactory.ads(audio_source=self.audio_source)
amine@173 48 size = ads.block_size
amine@173 49 self.assertEqual(
amine@173 50 size,
amine@173 51 160,
amine@173 52 "Wrong default block_size, expected: 160, found: {0}".format(size),
amine@173 53 )
amine@173 54
amine@172 55 def test_block_size(self):
amine@172 56 ads = ADSFactory.ads(audio_source=self.audio_source, block_size=512)
amine@173 57 size = ads.block_size
amine@173 58 self.assertEqual(
amine@173 59 size,
amine@173 60 512,
amine@173 61 "Wrong block_size, expected: 512, found: {0}".format(size),
amine@173 62 )
amine@173 63
amine@172 64 # with alias keyword
amine@172 65 ads = ADSFactory.ads(audio_source=self.audio_source, bs=160)
amine@173 66 size = ads.block_size
amine@173 67 self.assertEqual(
amine@173 68 size,
amine@173 69 160,
amine@173 70 "Wrong block_size, expected: 160, found: {0}".format(size),
amine@173 71 )
amine@173 72
amine@172 73 def test_block_duration(self):
amine@173 74
amine@173 75 ads = ADSFactory.ads(
amine@173 76 audio_source=self.audio_source, block_dur=0.01
amine@173 77 ) # 10 ms
amine@173 78 size = ads.block_size
amine@173 79 self.assertEqual(
amine@173 80 size,
amine@173 81 160,
amine@173 82 "Wrong block_size, expected: 160, found: {0}".format(size),
amine@173 83 )
amine@173 84
amine@172 85 # with alias keyword
amine@173 86 ads = ADSFactory.ads(audio_source=self.audio_source, bd=0.025) # 25 ms
amine@173 87 size = ads.block_size
amine@173 88 self.assertEqual(
amine@173 89 size,
amine@173 90 400,
amine@173 91 "Wrong block_size, expected: 400, found: {0}".format(size),
amine@173 92 )
amine@173 93
amine@172 94 def test_hop_duration(self):
amine@173 95
amine@173 96 ads = ADSFactory.ads(
amine@173 97 audio_source=self.audio_source, block_dur=0.02, hop_dur=0.01
amine@173 98 ) # 10 ms
amine@172 99 size = ads.hop_size
amine@173 100 self.assertEqual(
amine@173 101 size, 160, "Wrong hop_size, expected: 160, found: {0}".format(size)
amine@173 102 )
amine@173 103
amine@172 104 # with alias keyword
amine@173 105 ads = ADSFactory.ads(
amine@173 106 audio_source=self.audio_source, bd=0.025, hop_dur=0.015
amine@173 107 ) # 15 ms
amine@172 108 size = ads.hop_size
amine@173 109 self.assertEqual(
amine@173 110 size,
amine@173 111 240,
amine@173 112 "Wrong block_size, expected: 240, found: {0}".format(size),
amine@173 113 )
amine@173 114
amine@172 115 def test_sampling_rate(self):
amine@172 116 ads = ADSFactory.ads(audio_source=self.audio_source)
amine@173 117
amine@173 118 srate = ads.sampling_rate
amine@173 119 self.assertEqual(
amine@173 120 srate,
amine@173 121 16000,
amine@173 122 "Wrong sampling rate, expected: 16000, found: {0}".format(srate),
amine@173 123 )
amine@173 124
amine@172 125 def test_sample_width(self):
amine@172 126 ads = ADSFactory.ads(audio_source=self.audio_source)
amine@173 127
amine@173 128 swidth = ads.sample_width
amine@173 129 self.assertEqual(
amine@173 130 swidth,
amine@173 131 2,
amine@173 132 "Wrong sample width, expected: 2, found: {0}".format(swidth),
amine@173 133 )
amine@173 134
amine@172 135 def test_channels(self):
amine@172 136 ads = ADSFactory.ads(audio_source=self.audio_source)
amine@173 137
amine@173 138 channels = ads.channels
amine@173 139 self.assertEqual(
amine@173 140 channels,
amine@173 141 1,
amine@173 142 "Wrong number of channels, expected: 1, found: {0}".format(
amine@173 143 channels
amine@173 144 ),
amine@173 145 )
amine@173 146
amine@172 147 def test_read(self):
amine@173 148 ads = ADSFactory.ads(audio_source=self.audio_source, block_size=256)
amine@173 149
amine@172 150 ads.open()
amine@172 151 ads_data = ads.read()
amine@172 152 ads.close()
amine@173 153
amine@173 154 audio_source = WaveAudioSource(
amine@173 155 filename=dataset.one_to_six_arabic_16000_mono_bc_noise
amine@173 156 )
amine@172 157 audio_source.open()
amine@172 158 audio_source_data = audio_source.read(256)
amine@172 159 audio_source.close()
amine@173 160
amine@173 161 self.assertEqual(
amine@173 162 ads_data, audio_source_data, "Unexpected data read from ads"
amine@173 163 )
amine@173 164
amine@172 165 def test_Limiter_Deco_read(self):
amine@172 166 # read a maximum of 0.75 seconds from audio source
amine@172 167 ads = ADSFactory.ads(audio_source=self.audio_source, max_time=0.75)
amine@173 168
amine@172 169 ads_data = []
amine@172 170 ads.open()
amine@172 171 while True:
amine@172 172 block = ads.read()
amine@172 173 if block is None:
amine@172 174 break
amine@172 175 ads_data.append(block)
amine@172 176 ads.close()
amine@173 177 ads_data = b"".join(ads_data)
amine@173 178
amine@173 179 audio_source = WaveAudioSource(
amine@173 180 filename=dataset.one_to_six_arabic_16000_mono_bc_noise
amine@173 181 )
amine@172 182 audio_source.open()
amine@172 183 audio_source_data = audio_source.read(int(16000 * 0.75))
amine@172 184 audio_source.close()
amine@173 185
amine@173 186 self.assertEqual(
amine@173 187 ads_data, audio_source_data, "Unexpected data read from LimiterADS"
amine@173 188 )
amine@173 189
amine@172 190 def test_Limiter_Deco_read_limit(self):
amine@173 191 # read a maximum of 1.191 seconds from audio source
amine@172 192 ads = ADSFactory.ads(audio_source=self.audio_source, max_time=1.191)
amine@173 193 total_samples = round(ads.sampling_rate * 1.191)
amine@173 194 nb_full_blocks, last_block_size = divmod(total_samples, ads.block_size)
amine@173 195 total_samples_with_overlap = (
amine@173 196 nb_full_blocks * ads.block_size + last_block_size
amine@173 197 )
amine@173 198 expected_read_bytes = (
amine@173 199 total_samples_with_overlap * ads.sw * ads.channels
amine@173 200 )
amine@173 201
amine@172 202 total_read = 0
amine@172 203 ads.open()
amine@172 204 i = 0
amine@172 205 while True:
amine@172 206 block = ads.read()
amine@172 207 if block is None:
amine@172 208 break
amine@172 209 i += 1
amine@172 210 total_read += len(block)
amine@173 211
amine@172 212 ads.close()
amine@173 213
amine@173 214 self.assertEqual(
amine@173 215 total_read,
amine@173 216 expected_read_bytes,
amine@173 217 "Wrong data length read from LimiterADS, expected: {0}, found: {1}".format(
amine@173 218 expected_read_bytes, total_read
amine@173 219 ),
amine@173 220 )
amine@173 221
amine@172 222 def test_Recorder_Deco_read(self):
amine@173 223 ads = ADSFactory.ads(
amine@173 224 audio_source=self.audio_source, record=True, block_size=500
amine@173 225 )
amine@173 226
amine@172 227 ads_data = []
amine@172 228 ads.open()
amine@172 229 for i in range(10):
amine@172 230 block = ads.read()
amine@172 231 if block is None:
amine@172 232 break
amine@172 233 ads_data.append(block)
amine@172 234 ads.close()
amine@173 235 ads_data = b"".join(ads_data)
amine@173 236
amine@173 237 audio_source = WaveAudioSource(
amine@173 238 filename=dataset.one_to_six_arabic_16000_mono_bc_noise
amine@173 239 )
amine@172 240 audio_source.open()
amine@172 241 audio_source_data = audio_source.read(500 * 10)
amine@172 242 audio_source.close()
amine@173 243
amine@173 244 self.assertEqual(
amine@173 245 ads_data,
amine@173 246 audio_source_data,
amine@173 247 "Unexpected data read from RecorderADS",
amine@173 248 )
amine@173 249
amine@172 250 def test_Recorder_Deco_is_rewindable(self):
amine@172 251 ads = ADSFactory.ads(audio_source=self.audio_source, record=True)
amine@173 252
amine@173 253 self.assertTrue(
amine@173 254 ads.rewindable, "RecorderADS.is_rewindable should return True"
amine@173 255 )
amine@173 256
amine@172 257 def test_Recorder_Deco_rewind_and_read(self):
amine@173 258 ads = ADSFactory.ads(
amine@173 259 audio_source=self.audio_source, record=True, block_size=320
amine@173 260 )
amine@173 261
amine@172 262 ads.open()
amine@172 263 for i in range(10):
amine@172 264 ads.read()
amine@173 265
amine@172 266 ads.rewind()
amine@173 267
amine@172 268 # read all available data after rewind
amine@172 269 ads_data = []
amine@172 270 while True:
amine@172 271 block = ads.read()
amine@172 272 if block is None:
amine@172 273 break
amine@172 274 ads_data.append(block)
amine@172 275 ads.close()
amine@173 276 ads_data = b"".join(ads_data)
amine@173 277
amine@173 278 audio_source = WaveAudioSource(
amine@173 279 filename=dataset.one_to_six_arabic_16000_mono_bc_noise
amine@173 280 )
amine@172 281 audio_source.open()
amine@172 282 audio_source_data = audio_source.read(320 * 10)
amine@172 283 audio_source.close()
amine@173 284
amine@173 285 self.assertEqual(
amine@173 286 ads_data,
amine@173 287 audio_source_data,
amine@173 288 "Unexpected data read from RecorderADS",
amine@173 289 )
amine@173 290
amine@172 291 def test_Overlap_Deco_read(self):
amine@173 292
amine@172 293 # Use arbitrary valid block_size and hop_size
amine@172 294 block_size = 1714
amine@172 295 hop_size = 313
amine@173 296
amine@173 297 ads = ADSFactory.ads(
amine@173 298 audio_source=self.audio_source,
amine@173 299 block_size=block_size,
amine@173 300 hop_size=hop_size,
amine@173 301 )
amine@173 302
amine@172 303 # Read all available data overlapping blocks
amine@172 304 ads.open()
amine@172 305 ads_data = []
amine@172 306 while True:
amine@172 307 block = ads.read()
amine@172 308 if block is None:
amine@172 309 break
amine@172 310 ads_data.append(block)
amine@172 311 ads.close()
amine@173 312
amine@172 313 # Read all data from file and build a BufferAudioSource
amine@172 314 fp = wave.open(dataset.one_to_six_arabic_16000_mono_bc_noise, "r")
amine@172 315 wave_data = fp.readframes(fp.getnframes())
amine@172 316 fp.close()
amine@173 317 audio_source = BufferAudioSource(
amine@173 318 wave_data, ads.sampling_rate, ads.sample_width, ads.channels
amine@173 319 )
amine@172 320 audio_source.open()
amine@173 321
amine@172 322 # Compare all blocks read from OverlapADS to those read
amine@172 323 # from an audio source with a manual set_position
amine@173 324 for i, block in enumerate(ads_data):
amine@173 325
amine@172 326 tmp = audio_source.read(block_size)
amine@173 327
amine@173 328 self.assertEqual(
amine@173 329 block,
amine@173 330 tmp,
amine@173 331 "Unexpected block (N={0}) read from OverlapADS".format(i),
amine@173 332 )
amine@173 333
amine@173 334 audio_source.set_position((i + 1) * hop_size)
amine@173 335
amine@172 336 audio_source.close()
amine@173 337
amine@173 338 def test_Limiter_Overlap_Deco_read(self):
amine@173 339
amine@172 340 block_size = 256
amine@172 341 hop_size = 200
amine@173 342
amine@173 343 ads = ADSFactory.ads(
amine@173 344 audio_source=self.audio_source,
amine@173 345 max_time=0.50,
amine@173 346 block_size=block_size,
amine@173 347 hop_size=hop_size,
amine@173 348 )
amine@173 349
amine@172 350 # Read all available data overlapping blocks
amine@172 351 ads.open()
amine@172 352 ads_data = []
amine@172 353 while True:
amine@172 354 block = ads.read()
amine@172 355 if block is None:
amine@172 356 break
amine@172 357 ads_data.append(block)
amine@172 358 ads.close()
amine@173 359
amine@172 360 # Read all data from file and build a BufferAudioSource
amine@172 361 fp = wave.open(dataset.one_to_six_arabic_16000_mono_bc_noise, "r")
amine@172 362 wave_data = fp.readframes(fp.getnframes())
amine@172 363 fp.close()
amine@173 364 audio_source = BufferAudioSource(
amine@173 365 wave_data, ads.sampling_rate, ads.sample_width, ads.channels
amine@173 366 )
amine@172 367 audio_source.open()
amine@173 368
amine@172 369 # Compare all blocks read from OverlapADS to those read
amine@172 370 # from an audio source with a manual set_position
amine@173 371 for i, block in enumerate(ads_data):
amine@173 372 tmp = audio_source.read(len(block) // (ads.sw * ads.ch))
amine@173 373 self.assertEqual(
amine@173 374 len(block),
amine@173 375 len(tmp),
amine@173 376 "Unexpected block (N={0}) read from OverlapADS".format(i),
amine@173 377 )
amine@173 378 audio_source.set_position((i + 1) * hop_size)
amine@173 379
amine@172 380 audio_source.close()
amine@173 381
amine@172 382 def test_Limiter_Overlap_Deco_read_limit(self):
amine@173 383
amine@172 384 block_size = 313
amine@172 385 hop_size = 207
amine@173 386 ads = ADSFactory.ads(
amine@173 387 audio_source=self.audio_source,
amine@173 388 max_time=1.932,
amine@173 389 block_size=block_size,
amine@173 390 hop_size=hop_size,
amine@173 391 )
amine@173 392
amine@173 393 total_samples = round(ads.sampling_rate * 1.932)
amine@173 394 first_read_size = block_size
amine@173 395 next_read_size = block_size - hop_size
amine@173 396 nb_next_blocks, last_block_size = divmod(
amine@173 397 (total_samples - first_read_size), next_read_size
amine@173 398 )
amine@173 399 total_samples_with_overlap = (
amine@173 400 first_read_size + next_read_size * nb_next_blocks + last_block_size
amine@173 401 )
amine@173 402 expected_read_bytes = (
amine@173 403 total_samples_with_overlap * ads.sw * ads.channels
amine@173 404 )
amine@173 405
amine@173 406 cache_size = (block_size - hop_size) * ads.sample_width * ads.channels
amine@172 407 total_read = cache_size
amine@173 408
amine@172 409 ads.open()
amine@172 410 i = 0
amine@172 411 while True:
amine@172 412 block = ads.read()
amine@172 413 if block is None:
amine@172 414 break
amine@172 415 i += 1
amine@172 416 total_read += len(block) - cache_size
amine@173 417
amine@172 418 ads.close()
amine@173 419 self.assertEqual(
amine@173 420 total_read,
amine@173 421 expected_read_bytes,
amine@173 422 "Wrong data length read from LimiterADS, expected: {0}, found: {1}".format(
amine@173 423 expected_read_bytes, total_read
amine@173 424 ),
amine@173 425 )
amine@173 426
amine@172 427 def test_Recorder_Overlap_Deco_is_rewindable(self):
amine@173 428 ads = ADSFactory.ads(
amine@173 429 audio_source=self.audio_source,
amine@173 430 block_size=320,
amine@173 431 hop_size=160,
amine@173 432 record=True,
amine@173 433 )
amine@173 434 self.assertTrue(
amine@173 435 ads.rewindable, "RecorderADS.is_rewindable should return True"
amine@173 436 )
amine@172 437
amine@172 438 def test_Recorder_Overlap_Deco_rewind_and_read(self):
amine@173 439
amine@172 440 # Use arbitrary valid block_size and hop_size
amine@172 441 block_size = 1600
amine@172 442 hop_size = 400
amine@173 443
amine@173 444 ads = ADSFactory.ads(
amine@173 445 audio_source=self.audio_source,
amine@173 446 block_size=block_size,
amine@173 447 hop_size=hop_size,
amine@173 448 record=True,
amine@173 449 )
amine@173 450
amine@172 451 # Read all available data overlapping blocks
amine@172 452 ads.open()
amine@172 453 i = 0
amine@172 454 while True:
amine@172 455 block = ads.read()
amine@172 456 if block is None:
amine@172 457 break
amine@172 458 i += 1
amine@173 459
amine@172 460 ads.rewind()
amine@173 461
amine@172 462 # Read all data from file and build a BufferAudioSource
amine@172 463 fp = wave.open(dataset.one_to_six_arabic_16000_mono_bc_noise, "r")
amine@172 464 wave_data = fp.readframes(fp.getnframes())
amine@172 465 fp.close()
amine@173 466 audio_source = BufferAudioSource(
amine@173 467 wave_data, ads.sampling_rate, ads.sample_width, ads.channels
amine@173 468 )
amine@172 469 audio_source.open()
amine@173 470
amine@172 471 # Compare all blocks read from OverlapADS to those read
amine@172 472 # from an audio source with a manual set_position
amine@172 473 for j in range(i):
amine@173 474
amine@172 475 tmp = audio_source.read(block_size)
amine@173 476
amine@173 477 self.assertEqual(
amine@173 478 ads.read(),
amine@173 479 tmp,
amine@173 480 "Unexpected block (N={0}) read from OverlapADS".format(i),
amine@173 481 )
amine@173 482 audio_source.set_position((j + 1) * hop_size)
amine@173 483
amine@172 484 ads.close()
amine@172 485 audio_source.close()
amine@173 486
amine@172 487 def test_Limiter_Recorder_Overlap_Deco_rewind_and_read(self):
amine@173 488
amine@172 489 # Use arbitrary valid block_size and hop_size
amine@172 490 block_size = 1600
amine@172 491 hop_size = 400
amine@173 492
amine@173 493 ads = ADSFactory.ads(
amine@173 494 audio_source=self.audio_source,
amine@173 495 max_time=1.50,
amine@173 496 block_size=block_size,
amine@173 497 hop_size=hop_size,
amine@173 498 record=True,
amine@173 499 )
amine@173 500
amine@172 501 # Read all available data overlapping blocks
amine@172 502 ads.open()
amine@172 503 i = 0
amine@172 504 while True:
amine@172 505 block = ads.read()
amine@172 506 if block is None:
amine@172 507 break
amine@172 508 i += 1
amine@173 509
amine@172 510 ads.rewind()
amine@173 511
amine@172 512 # Read all data from file and build a BufferAudioSource
amine@172 513 fp = wave.open(dataset.one_to_six_arabic_16000_mono_bc_noise, "r")
amine@172 514 wave_data = fp.readframes(fp.getnframes())
amine@172 515 fp.close()
amine@173 516 audio_source = BufferAudioSource(
amine@173 517 wave_data, ads.sampling_rate, ads.sample_width, ads.channels
amine@173 518 )
amine@172 519 audio_source.open()
amine@173 520
amine@172 521 # Compare all blocks read from OverlapADS to those read
amine@172 522 # from an audio source with a manual set_position
amine@172 523 for j in range(i):
amine@173 524
amine@172 525 tmp = audio_source.read(block_size)
amine@173 526
amine@173 527 self.assertEqual(
amine@173 528 ads.read(),
amine@173 529 tmp,
amine@173 530 "Unexpected block (N={0}) read from OverlapADS".format(i),
amine@173 531 )
amine@173 532 audio_source.set_position((j + 1) * hop_size)
amine@173 533
amine@172 534 ads.close()
amine@172 535 audio_source.close()
amine@173 536
amine@172 537 def test_Limiter_Recorder_Overlap_Deco_rewind_and_read_limit(self):
amine@173 538
amine@172 539 # Use arbitrary valid block_size and hop_size
amine@172 540 block_size = 1000
amine@172 541 hop_size = 200
amine@173 542
amine@173 543 ads = ADSFactory.ads(
amine@173 544 audio_source=self.audio_source,
amine@173 545 max_time=1.317,
amine@173 546 block_size=block_size,
amine@173 547 hop_size=hop_size,
amine@173 548 record=True,
amine@173 549 )
amine@173 550 total_samples = round(ads.sampling_rate * 1.317)
amine@173 551 first_read_size = block_size
amine@173 552 next_read_size = block_size - hop_size
amine@173 553 nb_next_blocks, last_block_size = divmod(
amine@173 554 (total_samples - first_read_size), next_read_size
amine@173 555 )
amine@173 556 total_samples_with_overlap = (
amine@173 557 first_read_size + next_read_size * nb_next_blocks + last_block_size
amine@173 558 )
amine@173 559 expected_read_bytes = (
amine@173 560 total_samples_with_overlap * ads.sw * ads.channels
amine@173 561 )
amine@173 562
amine@173 563 cache_size = (block_size - hop_size) * ads.sample_width * ads.channels
amine@172 564 total_read = cache_size
amine@173 565
amine@172 566 ads.open()
amine@172 567 i = 0
amine@172 568 while True:
amine@172 569 block = ads.read()
amine@172 570 if block is None:
amine@172 571 break
amine@172 572 i += 1
amine@172 573 total_read += len(block) - cache_size
amine@173 574
amine@172 575 ads.close()
amine@173 576 self.assertEqual(
amine@173 577 total_read,
amine@173 578 expected_read_bytes,
amine@173 579 "Wrong data length read from LimiterADS, expected: {0}, found: {1}".format(
amine@173 580 expected_read_bytes, total_read
amine@173 581 ),
amine@173 582 )
amine@173 583
amine@173 584
amine@172 585 class TestADSFactoryBufferAudioSource(unittest.TestCase):
amine@172 586 def setUp(self):
amine@173 587 self.signal = b"ABCDEFGHIJKLMNOPQRSTUVWXYZ012345"
amine@173 588 self.ads = ADSFactory.ads(
amine@173 589 data_buffer=self.signal,
amine@173 590 sampling_rate=16,
amine@173 591 sample_width=2,
amine@173 592 channels=1,
amine@173 593 )
amine@173 594
amine@172 595 def test_ADS_BAS_sampling_rate(self):
amine@173 596 srate = self.ads.sampling_rate
amine@173 597 self.assertEqual(
amine@173 598 srate,
amine@173 599 16,
amine@173 600 "Wrong sampling rate, expected: 16000, found: {0}".format(srate),
amine@173 601 )
amine@173 602
amine@172 603 def test_ADS_BAS_get_sample_width(self):
amine@173 604 swidth = self.ads.sample_width
amine@173 605 self.assertEqual(
amine@173 606 swidth,
amine@173 607 2,
amine@173 608 "Wrong sample width, expected: 2, found: {0}".format(swidth),
amine@173 609 )
amine@173 610
amine@172 611 def test_ADS_BAS_get_channels(self):
amine@173 612 channels = self.ads.channels
amine@173 613 self.assertEqual(
amine@173 614 channels,
amine@173 615 1,
amine@173 616 "Wrong number of channels, expected: 1, found: {0}".format(
amine@173 617 channels
amine@173 618 ),
amine@173 619 )
amine@173 620
amine@172 621 def test_Limiter_Recorder_Overlap_Deco_rewind_and_read(self):
amine@173 622
amine@172 623 # Use arbitrary valid block_size and hop_size
amine@172 624 block_size = 5
amine@172 625 hop_size = 4
amine@173 626
amine@173 627 ads = ADSFactory.ads(
amine@173 628 data_buffer=self.signal,
amine@173 629 sampling_rate=16,
amine@173 630 sample_width=2,
amine@173 631 channels=1,
amine@173 632 max_time=0.80,
amine@173 633 block_size=block_size,
amine@173 634 hop_size=hop_size,
amine@173 635 record=True,
amine@173 636 )
amine@173 637
amine@172 638 # Read all available data overlapping blocks
amine@172 639 ads.open()
amine@172 640 i = 0
amine@172 641 while True:
amine@172 642 block = ads.read()
amine@172 643 if block is None:
amine@172 644 break
amine@172 645 i += 1
amine@173 646
amine@172 647 ads.rewind()
amine@173 648
amine@172 649 # Build a BufferAudioSource
amine@173 650 audio_source = BufferAudioSource(
amine@173 651 self.signal, ads.sampling_rate, ads.sample_width, ads.channels
amine@173 652 )
amine@172 653 audio_source.open()
amine@173 654
amine@172 655 # Compare all blocks read from OverlapADS to those read
amine@172 656 # from an audio source with a manual set_position
amine@172 657 for j in range(i):
amine@173 658
amine@172 659 tmp = audio_source.read(block_size)
amine@173 660
amine@172 661 block = ads.read()
amine@173 662
amine@173 663 self.assertEqual(
amine@173 664 block,
amine@173 665 tmp,
amine@173 666 "Unexpected block '{}' (N={}) read from OverlapADS".format(
amine@173 667 block, i
amine@173 668 ),
amine@173 669 )
amine@173 670 audio_source.set_position((j + 1) * hop_size)
amine@173 671
amine@172 672 ads.close()
amine@172 673 audio_source.close()
amine@172 674
amine@172 675
amine@172 676 class TestADSFactoryAlias(unittest.TestCase):
amine@172 677 def setUp(self):
amine@172 678 self.signal = "ABCDEFGHIJKLMNOPQRSTUVWXYZ012345"
amine@173 679
amine@172 680 def test_sampling_rate_alias(self):
amine@173 681 ads = ADSFactory.ads(
amine@173 682 data_buffer=self.signal, sr=16, sample_width=2, channels=1
amine@173 683 )
amine@173 684 srate = ads.sampling_rate
amine@173 685 self.assertEqual(
amine@173 686 srate,
amine@173 687 16,
amine@173 688 "Wrong sampling rate, expected: 16000, found: {0}".format(srate),
amine@173 689 )
amine@173 690
amine@172 691 def test_sampling_rate_duplicate(self):
amine@173 692 func = partial(
amine@173 693 ADSFactory.ads,
amine@173 694 data_buffer=self.signal,
amine@173 695 sr=16,
amine@173 696 sampling_rate=16,
amine@173 697 sample_width=2,
amine@173 698 channels=1,
amine@173 699 )
amine@172 700 self.assertRaises(DuplicateArgument, func)
amine@173 701
amine@172 702 def test_sample_width_alias(self):
amine@173 703 ads = ADSFactory.ads(
amine@173 704 data_buffer=self.signal, sampling_rate=16, sw=2, channels=1
amine@173 705 )
amine@173 706 swidth = ads.sample_width
amine@173 707 self.assertEqual(
amine@173 708 swidth,
amine@173 709 2,
amine@173 710 "Wrong sample width, expected: 2, found: {0}".format(swidth),
amine@173 711 )
amine@173 712
amine@172 713 def test_sample_width_duplicate(self):
amine@173 714 func = partial(
amine@173 715 ADSFactory.ads,
amine@173 716 data_buffer=self.signal,
amine@173 717 sampling_rate=16,
amine@173 718 sw=2,
amine@173 719 sample_width=2,
amine@173 720 channels=1,
amine@173 721 )
amine@172 722 self.assertRaises(DuplicateArgument, func)
amine@173 723
amine@172 724 def test_channels_alias(self):
amine@173 725 ads = ADSFactory.ads(
amine@173 726 data_buffer=self.signal, sampling_rate=16, sample_width=2, ch=1
amine@173 727 )
amine@173 728 channels = ads.channels
amine@173 729 self.assertEqual(
amine@173 730 channels,
amine@173 731 1,
amine@173 732 "Wrong number of channels, expected: 1, found: {0}".format(
amine@173 733 channels
amine@173 734 ),
amine@173 735 )
amine@173 736
amine@172 737 def test_channels_duplicate(self):
amine@173 738 func = partial(
amine@173 739 ADSFactory.ads,
amine@173 740 data_buffer=self.signal,
amine@173 741 sampling_rate=16,
amine@173 742 sample_width=2,
amine@173 743 ch=1,
amine@173 744 channels=1,
amine@173 745 )
amine@172 746 self.assertRaises(DuplicateArgument, func)
amine@173 747
amine@172 748 def test_block_size_alias(self):
amine@173 749 ads = ADSFactory.ads(
amine@173 750 data_buffer=self.signal,
amine@173 751 sampling_rate=16,
amine@173 752 sample_width=2,
amine@173 753 channels=1,
amine@173 754 bs=8,
amine@173 755 )
amine@173 756 size = ads.block_size
amine@173 757 self.assertEqual(
amine@173 758 size,
amine@173 759 8,
amine@173 760 "Wrong block_size using bs alias, expected: 8, found: {0}".format(
amine@173 761 size
amine@173 762 ),
amine@173 763 )
amine@173 764
amine@172 765 def test_block_size_duplicate(self):
amine@173 766 func = partial(
amine@173 767 ADSFactory.ads,
amine@173 768 data_buffer=self.signal,
amine@173 769 sampling_rate=16,
amine@173 770 sample_width=2,
amine@173 771 channels=1,
amine@173 772 bs=4,
amine@173 773 block_size=4,
amine@173 774 )
amine@172 775 self.assertRaises(DuplicateArgument, func)
amine@173 776
amine@172 777 def test_block_duration_alias(self):
amine@173 778 ads = ADSFactory.ads(
amine@173 779 data_buffer=self.signal,
amine@173 780 sampling_rate=16,
amine@173 781 sample_width=2,
amine@173 782 channels=1,
amine@173 783 bd=0.75,
amine@173 784 )
amine@172 785 # 0.75 ms = 0.75 * 16 = 12
amine@173 786 size = ads.block_size
amine@173 787 self.assertEqual(
amine@173 788 size,
amine@173 789 12,
amine@173 790 "Wrong block_size set with a block_dur alias 'bd', expected: 8, found: {0}".format(
amine@173 791 size
amine@173 792 ),
amine@173 793 )
amine@173 794
amine@172 795 def test_block_duration_duplicate(self):
amine@173 796 func = partial(
amine@173 797 ADSFactory.ads,
amine@173 798 data_buffer=self.signal,
amine@173 799 sampling_rate=16,
amine@173 800 sample_width=2,
amine@173 801 channels=1,
amine@173 802 bd=4,
amine@173 803 block_dur=4,
amine@173 804 )
amine@172 805 self.assertRaises(DuplicateArgument, func)
amine@173 806
amine@172 807 def test_block_size_duration_duplicate(self):
amine@173 808 func = partial(
amine@173 809 ADSFactory.ads,
amine@173 810 data_buffer=self.signal,
amine@173 811 sampling_rate=16,
amine@173 812 sample_width=2,
amine@173 813 channels=1,
amine@173 814 bd=4,
amine@173 815 bs=12,
amine@173 816 )
amine@172 817 self.assertRaises(DuplicateArgument, func)
amine@173 818
amine@172 819 def test_hop_duration_alias(self):
amine@173 820
amine@173 821 ads = ADSFactory.ads(
amine@173 822 data_buffer=self.signal,
amine@173 823 sampling_rate=16,
amine@173 824 sample_width=2,
amine@173 825 channels=1,
amine@173 826 bd=0.75,
amine@173 827 hd=0.5,
amine@173 828 )
amine@172 829 size = ads.hop_size
amine@173 830 self.assertEqual(
amine@173 831 size,
amine@173 832 8,
amine@173 833 "Wrong block_size using bs alias, expected: 8, found: {0}".format(
amine@173 834 size
amine@173 835 ),
amine@173 836 )
amine@173 837
amine@172 838 def test_hop_duration_duplicate(self):
amine@173 839
amine@173 840 func = partial(
amine@173 841 ADSFactory.ads,
amine@173 842 data_buffer=self.signal,
amine@173 843 sampling_rate=16,
amine@173 844 sample_width=2,
amine@173 845 channels=1,
amine@173 846 bd=0.75,
amine@173 847 hd=0.5,
amine@173 848 hop_dur=0.5,
amine@173 849 )
amine@172 850 self.assertRaises(DuplicateArgument, func)
amine@173 851
amine@172 852 def test_hop_size_duration_duplicate(self):
amine@173 853 func = partial(
amine@173 854 ADSFactory.ads,
amine@173 855 data_buffer=self.signal,
amine@173 856 sampling_rate=16,
amine@173 857 sample_width=2,
amine@173 858 channels=1,
amine@173 859 bs=8,
amine@173 860 hs=4,
amine@173 861 hd=1,
amine@173 862 )
amine@172 863 self.assertRaises(DuplicateArgument, func)
amine@173 864
amine@172 865 def test_hop_size_greater_than_block_size(self):
amine@173 866 func = partial(
amine@173 867 ADSFactory.ads,
amine@173 868 data_buffer=self.signal,
amine@173 869 sampling_rate=16,
amine@173 870 sample_width=2,
amine@173 871 channels=1,
amine@173 872 bs=4,
amine@173 873 hs=8,
amine@173 874 )
amine@172 875 self.assertRaises(ValueError, func)
amine@173 876
amine@172 877 def test_filename_alias(self):
amine@172 878 ads = ADSFactory.ads(fn=dataset.one_to_six_arabic_16000_mono_bc_noise)
amine@173 879
amine@172 880 def test_filename_duplicate(self):
amine@173 881
amine@173 882 func = partial(
amine@173 883 ADSFactory.ads,
amine@173 884 fn=dataset.one_to_six_arabic_16000_mono_bc_noise,
amine@173 885 filename=dataset.one_to_six_arabic_16000_mono_bc_noise,
amine@173 886 )
amine@172 887 self.assertRaises(DuplicateArgument, func)
amine@173 888
amine@172 889 def test_data_buffer_duplicate(self):
amine@173 890 func = partial(
amine@173 891 ADSFactory.ads,
amine@173 892 data_buffer=self.signal,
amine@173 893 db=self.signal,
amine@173 894 sampling_rate=16,
amine@173 895 sample_width=2,
amine@173 896 channels=1,
amine@173 897 )
amine@172 898 self.assertRaises(DuplicateArgument, func)
amine@173 899
amine@172 900 def test_max_time_alias(self):
amine@173 901 ads = ADSFactory.ads(
amine@173 902 data_buffer=self.signal,
amine@173 903 sampling_rate=16,
amine@173 904 sample_width=2,
amine@173 905 channels=1,
amine@173 906 mt=10,
amine@173 907 )
amine@173 908 self.assertEqual(
amine@173 909 ads.max_read,
amine@173 910 10,
amine@173 911 "Wrong AudioDataSource.max_read, expected: 10, found: {}".format(
amine@173 912 ads.max_read
amine@173 913 ),
amine@173 914 )
amine@173 915
amine@172 916 def test_max_time_duplicate(self):
amine@173 917 func = partial(
amine@173 918 ADSFactory.ads,
amine@173 919 data_buffer=self.signal,
amine@173 920 sampling_rate=16,
amine@173 921 sample_width=2,
amine@173 922 channels=1,
amine@173 923 mt=True,
amine@173 924 max_time=True,
amine@173 925 )
amine@173 926
amine@172 927 self.assertRaises(DuplicateArgument, func)
amine@173 928
amine@172 929 def test_record_alias(self):
amine@173 930 ads = ADSFactory.ads(
amine@173 931 data_buffer=self.signal,
amine@173 932 sampling_rate=16,
amine@173 933 sample_width=2,
amine@173 934 channels=1,
amine@173 935 rec=True,
amine@173 936 )
amine@173 937 self.assertTrue(
amine@173 938 ads.rewindable, "AudioDataSource.rewindable expected to be True"
amine@173 939 )
amine@173 940
amine@172 941 def test_record_duplicate(self):
amine@173 942 func = partial(
amine@173 943 ADSFactory.ads,
amine@173 944 data_buffer=self.signal,
amine@173 945 sampling_rate=16,
amine@173 946 sample_width=2,
amine@173 947 channels=1,
amine@173 948 rec=True,
amine@173 949 record=True,
amine@173 950 )
amine@172 951 self.assertRaises(DuplicateArgument, func)
amine@173 952
amine@172 953 def test_Limiter_Recorder_Overlap_Deco_rewind_and_read_alias(self):
amine@173 954
amine@172 955 # Use arbitrary valid block_size and hop_size
amine@172 956 block_size = 5
amine@172 957 hop_size = 4
amine@173 958
amine@173 959 ads = ADSFactory.ads(
amine@173 960 db=self.signal,
amine@173 961 sr=16,
amine@173 962 sw=2,
amine@173 963 ch=1,
amine@173 964 mt=0.80,
amine@173 965 bs=block_size,
amine@173 966 hs=hop_size,
amine@173 967 rec=True,
amine@173 968 )
amine@173 969
amine@172 970 # Read all available data overlapping blocks
amine@172 971 ads.open()
amine@172 972 i = 0
amine@172 973 while True:
amine@172 974 block = ads.read()
amine@172 975 if block is None:
amine@172 976 break
amine@172 977 i += 1
amine@173 978
amine@172 979 ads.rewind()
amine@173 980
amine@172 981 # Build a BufferAudioSource
amine@173 982 audio_source = BufferAudioSource(
amine@173 983 self.signal, ads.sampling_rate, ads.sample_width, ads.channels
amine@173 984 )
amine@172 985 audio_source.open()
amine@173 986
amine@173 987 # Compare all blocks read from AudioDataSource to those read
amine@173 988 # from an audio source with manual position definition
amine@172 989 for j in range(i):
amine@172 990 tmp = audio_source.read(block_size)
amine@172 991 block = ads.read()
amine@173 992 self.assertEqual(
amine@173 993 block,
amine@173 994 tmp,
amine@173 995 "Unexpected block (N={0}) read from OverlapADS".format(i),
amine@173 996 )
amine@173 997 audio_source.set_position((j + 1) * hop_size)
amine@172 998 ads.close()
amine@172 999 audio_source.close()
amine@173 1000
amine@172 1001
amine@172 1002 if __name__ == "__main__":
amine@173 1003 # import sys;sys.argv = ['', 'Test.testName']
amine@172 1004 unittest.main()