amine@173: """
amine@172: @author: Amine Sehili <amine.sehili@gmail.com>
amine@172: September 2015
amine@172: 
amine@173: """
amine@172: 
amine@172: import unittest
amine@172: from functools import partial
amine@172: import sys
amine@173: from auditok import (
amine@173:     dataset,
amine@173:     ADSFactory,
amine@173:     AudioDataSource,
amine@173:     BufferAudioSource,
amine@173:     WaveAudioSource,
amine@173:     DuplicateArgument,
amine@173: )
amine@172: import wave
amine@172: 
amine@172: 
amine@172: try:
amine@172:     from builtins import range
amine@172: except ImportError:
amine@172:     if sys.version_info < (3, 0):
amine@172:         range = xrange
amine@172: 
amine@173: 
amine@172: class TestADSFactoryFileAudioSource(unittest.TestCase):
amine@172:     def setUp(self):
amine@173:         self.audio_source = WaveAudioSource(
amine@173:             filename=dataset.one_to_six_arabic_16000_mono_bc_noise
amine@173:         )
amine@173: 
amine@172:     def test_ADS_type(self):
amine@173: 
amine@172:         ads = ADSFactory.ads(audio_source=self.audio_source)
amine@173: 
amine@173:         self.assertIsInstance(
amine@173:             ads,
amine@173:             AudioDataSource,
amine@173:             msg="wrong type for ads object, expected: 'AudioDataSource', found: {0}".format(
amine@173:                 type(ads)
amine@173:             ),
amine@173:         )
amine@173: 
amine@172:     def test_default_block_size(self):
amine@172:         ads = ADSFactory.ads(audio_source=self.audio_source)
amine@173:         size = ads.block_size
amine@173:         self.assertEqual(
amine@173:             size,
amine@173:             160,
amine@173:             "Wrong default block_size, expected: 160, found: {0}".format(size),
amine@173:         )
amine@173: 
amine@172:     def test_block_size(self):
amine@172:         ads = ADSFactory.ads(audio_source=self.audio_source, block_size=512)
amine@173:         size = ads.block_size
amine@173:         self.assertEqual(
amine@173:             size,
amine@173:             512,
amine@173:             "Wrong block_size, expected: 512, found: {0}".format(size),
amine@173:         )
amine@173: 
amine@172:         # with alias keyword
amine@172:         ads = ADSFactory.ads(audio_source=self.audio_source, bs=160)
amine@173:         size = ads.block_size
amine@173:         self.assertEqual(
amine@173:             size,
amine@173:             160,
amine@173:             "Wrong block_size, expected: 160, found: {0}".format(size),
amine@173:         )
amine@173: 
amine@172:     def test_block_duration(self):
amine@173: 
amine@173:         ads = ADSFactory.ads(
amine@173:             audio_source=self.audio_source, block_dur=0.01
amine@173:         )  # 10 ms
amine@173:         size = ads.block_size
amine@173:         self.assertEqual(
amine@173:             size,
amine@173:             160,
amine@173:             "Wrong block_size, expected: 160, found: {0}".format(size),
amine@173:         )
amine@173: 
amine@172:         # with alias keyword
amine@173:         ads = ADSFactory.ads(audio_source=self.audio_source, bd=0.025)  # 25 ms
amine@173:         size = ads.block_size
amine@173:         self.assertEqual(
amine@173:             size,
amine@173:             400,
amine@173:             "Wrong block_size, expected: 400, found: {0}".format(size),
amine@173:         )
amine@173: 
amine@172:     def test_hop_duration(self):
amine@173: 
amine@173:         ads = ADSFactory.ads(
amine@173:             audio_source=self.audio_source, block_dur=0.02, hop_dur=0.01
amine@173:         )  # 10 ms
amine@172:         size = ads.hop_size
amine@173:         self.assertEqual(
amine@173:             size, 160, "Wrong hop_size, expected: 160, found: {0}".format(size)
amine@173:         )
amine@173: 
amine@172:         # with alias keyword
amine@173:         ads = ADSFactory.ads(
amine@173:             audio_source=self.audio_source, bd=0.025, hop_dur=0.015
amine@173:         )  # 15 ms
amine@172:         size = ads.hop_size
amine@173:         self.assertEqual(
amine@173:             size,
amine@173:             240,
amine@173:             "Wrong block_size, expected: 240, found: {0}".format(size),
amine@173:         )
amine@173: 
amine@172:     def test_sampling_rate(self):
amine@172:         ads = ADSFactory.ads(audio_source=self.audio_source)
amine@173: 
amine@173:         srate = ads.sampling_rate
amine@173:         self.assertEqual(
amine@173:             srate,
amine@173:             16000,
amine@173:             "Wrong sampling rate, expected: 16000, found: {0}".format(srate),
amine@173:         )
amine@173: 
amine@172:     def test_sample_width(self):
amine@172:         ads = ADSFactory.ads(audio_source=self.audio_source)
amine@173: 
amine@173:         swidth = ads.sample_width
amine@173:         self.assertEqual(
amine@173:             swidth,
amine@173:             2,
amine@173:             "Wrong sample width, expected: 2, found: {0}".format(swidth),
amine@173:         )
amine@173: 
amine@172:     def test_channels(self):
amine@172:         ads = ADSFactory.ads(audio_source=self.audio_source)
amine@173: 
amine@173:         channels = ads.channels
amine@173:         self.assertEqual(
amine@173:             channels,
amine@173:             1,
amine@173:             "Wrong number of channels, expected: 1, found: {0}".format(
amine@173:                 channels
amine@173:             ),
amine@173:         )
amine@173: 
amine@172:     def test_read(self):
amine@173:         ads = ADSFactory.ads(audio_source=self.audio_source, block_size=256)
amine@173: 
amine@172:         ads.open()
amine@172:         ads_data = ads.read()
amine@172:         ads.close()
amine@173: 
amine@173:         audio_source = WaveAudioSource(
amine@173:             filename=dataset.one_to_six_arabic_16000_mono_bc_noise
amine@173:         )
amine@172:         audio_source.open()
amine@172:         audio_source_data = audio_source.read(256)
amine@172:         audio_source.close()
amine@173: 
amine@173:         self.assertEqual(
amine@173:             ads_data, audio_source_data, "Unexpected data read from ads"
amine@173:         )
amine@173: 
amine@172:     def test_Limiter_Deco_read(self):
amine@172:         # read a maximum of 0.75 seconds from audio source
amine@172:         ads = ADSFactory.ads(audio_source=self.audio_source, max_time=0.75)
amine@173: 
amine@172:         ads_data = []
amine@172:         ads.open()
amine@172:         while True:
amine@172:             block = ads.read()
amine@172:             if block is None:
amine@172:                 break
amine@172:             ads_data.append(block)
amine@172:         ads.close()
amine@173:         ads_data = b"".join(ads_data)
amine@173: 
amine@173:         audio_source = WaveAudioSource(
amine@173:             filename=dataset.one_to_six_arabic_16000_mono_bc_noise
amine@173:         )
amine@172:         audio_source.open()
amine@172:         audio_source_data = audio_source.read(int(16000 * 0.75))
amine@172:         audio_source.close()
amine@173: 
amine@173:         self.assertEqual(
amine@173:             ads_data, audio_source_data, "Unexpected data read from LimiterADS"
amine@173:         )
amine@173: 
amine@172:     def test_Limiter_Deco_read_limit(self):
amine@173:         # read a maximum of 1.191 seconds from audio source
amine@172:         ads = ADSFactory.ads(audio_source=self.audio_source, max_time=1.191)
amine@173:         total_samples = round(ads.sampling_rate * 1.191)
amine@173:         nb_full_blocks, last_block_size = divmod(total_samples, ads.block_size)
amine@173:         total_samples_with_overlap = (
amine@173:             nb_full_blocks * ads.block_size + last_block_size
amine@173:         )
amine@173:         expected_read_bytes = (
amine@173:             total_samples_with_overlap * ads.sw * ads.channels
amine@173:         )
amine@173: 
amine@172:         total_read = 0
amine@172:         ads.open()
amine@172:         i = 0
amine@172:         while True:
amine@172:             block = ads.read()
amine@172:             if block is None:
amine@172:                 break
amine@172:             i += 1
amine@172:             total_read += len(block)
amine@173: 
amine@172:         ads.close()
amine@173: 
amine@173:         self.assertEqual(
amine@173:             total_read,
amine@173:             expected_read_bytes,
amine@173:             "Wrong data length read from LimiterADS, expected: {0}, found: {1}".format(
amine@173:                 expected_read_bytes, total_read
amine@173:             ),
amine@173:         )
amine@173: 
amine@172:     def test_Recorder_Deco_read(self):
amine@173:         ads = ADSFactory.ads(
amine@173:             audio_source=self.audio_source, record=True, block_size=500
amine@173:         )
amine@173: 
amine@172:         ads_data = []
amine@172:         ads.open()
amine@172:         for i in range(10):
amine@172:             block = ads.read()
amine@172:             if block is None:
amine@172:                 break
amine@172:             ads_data.append(block)
amine@172:         ads.close()
amine@173:         ads_data = b"".join(ads_data)
amine@173: 
amine@173:         audio_source = WaveAudioSource(
amine@173:             filename=dataset.one_to_six_arabic_16000_mono_bc_noise
amine@173:         )
amine@172:         audio_source.open()
amine@172:         audio_source_data = audio_source.read(500 * 10)
amine@172:         audio_source.close()
amine@173: 
amine@173:         self.assertEqual(
amine@173:             ads_data,
amine@173:             audio_source_data,
amine@173:             "Unexpected data read from RecorderADS",
amine@173:         )
amine@173: 
amine@172:     def test_Recorder_Deco_is_rewindable(self):
amine@172:         ads = ADSFactory.ads(audio_source=self.audio_source, record=True)
amine@173: 
amine@173:         self.assertTrue(
amine@173:             ads.rewindable, "RecorderADS.is_rewindable should return True"
amine@173:         )
amine@173: 
amine@172:     def test_Recorder_Deco_rewind_and_read(self):
amine@173:         ads = ADSFactory.ads(
amine@173:             audio_source=self.audio_source, record=True, block_size=320
amine@173:         )
amine@173: 
amine@172:         ads.open()
amine@172:         for i in range(10):
amine@172:             ads.read()
amine@173: 
amine@172:         ads.rewind()
amine@173: 
amine@172:         # read all available data after rewind
amine@172:         ads_data = []
amine@172:         while True:
amine@172:             block = ads.read()
amine@172:             if block is None:
amine@172:                 break
amine@172:             ads_data.append(block)
amine@172:         ads.close()
amine@173:         ads_data = b"".join(ads_data)
amine@173: 
amine@173:         audio_source = WaveAudioSource(
amine@173:             filename=dataset.one_to_six_arabic_16000_mono_bc_noise
amine@173:         )
amine@172:         audio_source.open()
amine@172:         audio_source_data = audio_source.read(320 * 10)
amine@172:         audio_source.close()
amine@173: 
amine@173:         self.assertEqual(
amine@173:             ads_data,
amine@173:             audio_source_data,
amine@173:             "Unexpected data read from RecorderADS",
amine@173:         )
amine@173: 
amine@172:     def test_Overlap_Deco_read(self):
amine@173: 
amine@172:         # Use arbitrary valid block_size and hop_size
amine@172:         block_size = 1714
amine@172:         hop_size = 313
amine@173: 
amine@173:         ads = ADSFactory.ads(
amine@173:             audio_source=self.audio_source,
amine@173:             block_size=block_size,
amine@173:             hop_size=hop_size,
amine@173:         )
amine@173: 
amine@172:         # Read all available data overlapping blocks
amine@172:         ads.open()
amine@172:         ads_data = []
amine@172:         while True:
amine@172:             block = ads.read()
amine@172:             if block is None:
amine@172:                 break
amine@172:             ads_data.append(block)
amine@172:         ads.close()
amine@173: 
amine@172:         # Read all data from file and build a BufferAudioSource
amine@172:         fp = wave.open(dataset.one_to_six_arabic_16000_mono_bc_noise, "r")
amine@172:         wave_data = fp.readframes(fp.getnframes())
amine@172:         fp.close()
amine@173:         audio_source = BufferAudioSource(
amine@173:             wave_data, ads.sampling_rate, ads.sample_width, ads.channels
amine@173:         )
amine@172:         audio_source.open()
amine@173: 
amine@172:         # Compare all blocks read from OverlapADS to those read
amine@172:         # from an audio source with a manual set_position
amine@173:         for i, block in enumerate(ads_data):
amine@173: 
amine@172:             tmp = audio_source.read(block_size)
amine@173: 
amine@173:             self.assertEqual(
amine@173:                 block,
amine@173:                 tmp,
amine@173:                 "Unexpected block (N={0}) read from OverlapADS".format(i),
amine@173:             )
amine@173: 
amine@173:             audio_source.set_position((i + 1) * hop_size)
amine@173: 
amine@172:         audio_source.close()
amine@173: 
amine@173:     def test_Limiter_Overlap_Deco_read(self):
amine@173: 
amine@172:         block_size = 256
amine@172:         hop_size = 200
amine@173: 
amine@173:         ads = ADSFactory.ads(
amine@173:             audio_source=self.audio_source,
amine@173:             max_time=0.50,
amine@173:             block_size=block_size,
amine@173:             hop_size=hop_size,
amine@173:         )
amine@173: 
amine@172:         # Read all available data overlapping blocks
amine@172:         ads.open()
amine@172:         ads_data = []
amine@172:         while True:
amine@172:             block = ads.read()
amine@172:             if block is None:
amine@172:                 break
amine@172:             ads_data.append(block)
amine@172:         ads.close()
amine@173: 
amine@172:         # Read all data from file and build a BufferAudioSource
amine@172:         fp = wave.open(dataset.one_to_six_arabic_16000_mono_bc_noise, "r")
amine@172:         wave_data = fp.readframes(fp.getnframes())
amine@172:         fp.close()
amine@173:         audio_source = BufferAudioSource(
amine@173:             wave_data, ads.sampling_rate, ads.sample_width, ads.channels
amine@173:         )
amine@172:         audio_source.open()
amine@173: 
amine@172:         # Compare all blocks read from OverlapADS to those read
amine@172:         # from an audio source with a manual set_position
amine@173:         for i, block in enumerate(ads_data):
amine@173:             tmp = audio_source.read(len(block) // (ads.sw * ads.ch))
amine@173:             self.assertEqual(
amine@173:                 len(block),
amine@173:                 len(tmp),
amine@173:                 "Unexpected block (N={0}) read from OverlapADS".format(i),
amine@173:             )
amine@173:             audio_source.set_position((i + 1) * hop_size)
amine@173: 
amine@172:         audio_source.close()
amine@173: 
amine@172:     def test_Limiter_Overlap_Deco_read_limit(self):
amine@173: 
amine@172:         block_size = 313
amine@172:         hop_size = 207
amine@173:         ads = ADSFactory.ads(
amine@173:             audio_source=self.audio_source,
amine@173:             max_time=1.932,
amine@173:             block_size=block_size,
amine@173:             hop_size=hop_size,
amine@173:         )
amine@173: 
amine@173:         total_samples = round(ads.sampling_rate * 1.932)
amine@173:         first_read_size = block_size
amine@173:         next_read_size = block_size - hop_size
amine@173:         nb_next_blocks, last_block_size = divmod(
amine@173:             (total_samples - first_read_size), next_read_size
amine@173:         )
amine@173:         total_samples_with_overlap = (
amine@173:             first_read_size + next_read_size * nb_next_blocks + last_block_size
amine@173:         )
amine@173:         expected_read_bytes = (
amine@173:             total_samples_with_overlap * ads.sw * ads.channels
amine@173:         )
amine@173: 
amine@173:         cache_size = (block_size - hop_size) * ads.sample_width * ads.channels
amine@172:         total_read = cache_size
amine@173: 
amine@172:         ads.open()
amine@172:         i = 0
amine@172:         while True:
amine@172:             block = ads.read()
amine@172:             if block is None:
amine@172:                 break
amine@172:             i += 1
amine@172:             total_read += len(block) - cache_size
amine@173: 
amine@172:         ads.close()
amine@173:         self.assertEqual(
amine@173:             total_read,
amine@173:             expected_read_bytes,
amine@173:             "Wrong data length read from LimiterADS, expected: {0}, found: {1}".format(
amine@173:                 expected_read_bytes, total_read
amine@173:             ),
amine@173:         )
amine@173: 
amine@172:     def test_Recorder_Overlap_Deco_is_rewindable(self):
amine@173:         ads = ADSFactory.ads(
amine@173:             audio_source=self.audio_source,
amine@173:             block_size=320,
amine@173:             hop_size=160,
amine@173:             record=True,
amine@173:         )
amine@173:         self.assertTrue(
amine@173:             ads.rewindable, "RecorderADS.is_rewindable should return True"
amine@173:         )
amine@172: 
amine@172:     def test_Recorder_Overlap_Deco_rewind_and_read(self):
amine@173: 
amine@172:         # Use arbitrary valid block_size and hop_size
amine@172:         block_size = 1600
amine@172:         hop_size = 400
amine@173: 
amine@173:         ads = ADSFactory.ads(
amine@173:             audio_source=self.audio_source,
amine@173:             block_size=block_size,
amine@173:             hop_size=hop_size,
amine@173:             record=True,
amine@173:         )
amine@173: 
amine@172:         # Read all available data overlapping blocks
amine@172:         ads.open()
amine@172:         i = 0
amine@172:         while True:
amine@172:             block = ads.read()
amine@172:             if block is None:
amine@172:                 break
amine@172:             i += 1
amine@173: 
amine@172:         ads.rewind()
amine@173: 
amine@172:         # Read all data from file and build a BufferAudioSource
amine@172:         fp = wave.open(dataset.one_to_six_arabic_16000_mono_bc_noise, "r")
amine@172:         wave_data = fp.readframes(fp.getnframes())
amine@172:         fp.close()
amine@173:         audio_source = BufferAudioSource(
amine@173:             wave_data, ads.sampling_rate, ads.sample_width, ads.channels
amine@173:         )
amine@172:         audio_source.open()
amine@173: 
amine@172:         # Compare all blocks read from OverlapADS to those read
amine@172:         # from an audio source with a manual set_position
amine@172:         for j in range(i):
amine@173: 
amine@172:             tmp = audio_source.read(block_size)
amine@173: 
amine@173:             self.assertEqual(
amine@173:                 ads.read(),
amine@173:                 tmp,
amine@173:                 "Unexpected block (N={0}) read from OverlapADS".format(i),
amine@173:             )
amine@173:             audio_source.set_position((j + 1) * hop_size)
amine@173: 
amine@172:         ads.close()
amine@172:         audio_source.close()
amine@173: 
amine@172:     def test_Limiter_Recorder_Overlap_Deco_rewind_and_read(self):
amine@173: 
amine@172:         # Use arbitrary valid block_size and hop_size
amine@172:         block_size = 1600
amine@172:         hop_size = 400
amine@173: 
amine@173:         ads = ADSFactory.ads(
amine@173:             audio_source=self.audio_source,
amine@173:             max_time=1.50,
amine@173:             block_size=block_size,
amine@173:             hop_size=hop_size,
amine@173:             record=True,
amine@173:         )
amine@173: 
amine@172:         # Read all available data overlapping blocks
amine@172:         ads.open()
amine@172:         i = 0
amine@172:         while True:
amine@172:             block = ads.read()
amine@172:             if block is None:
amine@172:                 break
amine@172:             i += 1
amine@173: 
amine@172:         ads.rewind()
amine@173: 
amine@172:         # Read all data from file and build a BufferAudioSource
amine@172:         fp = wave.open(dataset.one_to_six_arabic_16000_mono_bc_noise, "r")
amine@172:         wave_data = fp.readframes(fp.getnframes())
amine@172:         fp.close()
amine@173:         audio_source = BufferAudioSource(
amine@173:             wave_data, ads.sampling_rate, ads.sample_width, ads.channels
amine@173:         )
amine@172:         audio_source.open()
amine@173: 
amine@172:         # Compare all blocks read from OverlapADS to those read
amine@172:         # from an audio source with a manual set_position
amine@172:         for j in range(i):
amine@173: 
amine@172:             tmp = audio_source.read(block_size)
amine@173: 
amine@173:             self.assertEqual(
amine@173:                 ads.read(),
amine@173:                 tmp,
amine@173:                 "Unexpected block (N={0}) read from OverlapADS".format(i),
amine@173:             )
amine@173:             audio_source.set_position((j + 1) * hop_size)
amine@173: 
amine@172:         ads.close()
amine@172:         audio_source.close()
amine@173: 
amine@172:     def test_Limiter_Recorder_Overlap_Deco_rewind_and_read_limit(self):
amine@173: 
amine@172:         # Use arbitrary valid block_size and hop_size
amine@172:         block_size = 1000
amine@172:         hop_size = 200
amine@173: 
amine@173:         ads = ADSFactory.ads(
amine@173:             audio_source=self.audio_source,
amine@173:             max_time=1.317,
amine@173:             block_size=block_size,
amine@173:             hop_size=hop_size,
amine@173:             record=True,
amine@173:         )
amine@173:         total_samples = round(ads.sampling_rate * 1.317)
amine@173:         first_read_size = block_size
amine@173:         next_read_size = block_size - hop_size
amine@173:         nb_next_blocks, last_block_size = divmod(
amine@173:             (total_samples - first_read_size), next_read_size
amine@173:         )
amine@173:         total_samples_with_overlap = (
amine@173:             first_read_size + next_read_size * nb_next_blocks + last_block_size
amine@173:         )
amine@173:         expected_read_bytes = (
amine@173:             total_samples_with_overlap * ads.sw * ads.channels
amine@173:         )
amine@173: 
amine@173:         cache_size = (block_size - hop_size) * ads.sample_width * ads.channels
amine@172:         total_read = cache_size
amine@173: 
amine@172:         ads.open()
amine@172:         i = 0
amine@172:         while True:
amine@172:             block = ads.read()
amine@172:             if block is None:
amine@172:                 break
amine@172:             i += 1
amine@172:             total_read += len(block) - cache_size
amine@173: 
amine@172:         ads.close()
amine@173:         self.assertEqual(
amine@173:             total_read,
amine@173:             expected_read_bytes,
amine@173:             "Wrong data length read from LimiterADS, expected: {0}, found: {1}".format(
amine@173:                 expected_read_bytes, total_read
amine@173:             ),
amine@173:         )
amine@173: 
amine@173: 
amine@172: class TestADSFactoryBufferAudioSource(unittest.TestCase):
amine@172:     def setUp(self):
amine@173:         self.signal = b"ABCDEFGHIJKLMNOPQRSTUVWXYZ012345"
amine@173:         self.ads = ADSFactory.ads(
amine@173:             data_buffer=self.signal,
amine@173:             sampling_rate=16,
amine@173:             sample_width=2,
amine@173:             channels=1,
amine@173:         )
amine@173: 
amine@172:     def test_ADS_BAS_sampling_rate(self):
amine@173:         srate = self.ads.sampling_rate
amine@173:         self.assertEqual(
amine@173:             srate,
amine@173:             16,
amine@173:             "Wrong sampling rate, expected: 16000, found: {0}".format(srate),
amine@173:         )
amine@173: 
amine@172:     def test_ADS_BAS_get_sample_width(self):
amine@173:         swidth = self.ads.sample_width
amine@173:         self.assertEqual(
amine@173:             swidth,
amine@173:             2,
amine@173:             "Wrong sample width, expected: 2, found: {0}".format(swidth),
amine@173:         )
amine@173: 
amine@172:     def test_ADS_BAS_get_channels(self):
amine@173:         channels = self.ads.channels
amine@173:         self.assertEqual(
amine@173:             channels,
amine@173:             1,
amine@173:             "Wrong number of channels, expected: 1, found: {0}".format(
amine@173:                 channels
amine@173:             ),
amine@173:         )
amine@173: 
amine@172:     def test_Limiter_Recorder_Overlap_Deco_rewind_and_read(self):
amine@173: 
amine@172:         # Use arbitrary valid block_size and hop_size
amine@172:         block_size = 5
amine@172:         hop_size = 4
amine@173: 
amine@173:         ads = ADSFactory.ads(
amine@173:             data_buffer=self.signal,
amine@173:             sampling_rate=16,
amine@173:             sample_width=2,
amine@173:             channels=1,
amine@173:             max_time=0.80,
amine@173:             block_size=block_size,
amine@173:             hop_size=hop_size,
amine@173:             record=True,
amine@173:         )
amine@173: 
amine@172:         # Read all available data overlapping blocks
amine@172:         ads.open()
amine@172:         i = 0
amine@172:         while True:
amine@172:             block = ads.read()
amine@172:             if block is None:
amine@172:                 break
amine@172:             i += 1
amine@173: 
amine@172:         ads.rewind()
amine@173: 
amine@172:         # Build a BufferAudioSource
amine@173:         audio_source = BufferAudioSource(
amine@173:             self.signal, ads.sampling_rate, ads.sample_width, ads.channels
amine@173:         )
amine@172:         audio_source.open()
amine@173: 
amine@172:         # Compare all blocks read from OverlapADS to those read
amine@172:         # from an audio source with a manual set_position
amine@172:         for j in range(i):
amine@173: 
amine@172:             tmp = audio_source.read(block_size)
amine@173: 
amine@172:             block = ads.read()
amine@173: 
amine@173:             self.assertEqual(
amine@173:                 block,
amine@173:                 tmp,
amine@173:                 "Unexpected block '{}' (N={}) read from OverlapADS".format(
amine@173:                     block, i
amine@173:                 ),
amine@173:             )
amine@173:             audio_source.set_position((j + 1) * hop_size)
amine@173: 
amine@172:         ads.close()
amine@172:         audio_source.close()
amine@172: 
amine@172: 
amine@172: class TestADSFactoryAlias(unittest.TestCase):
amine@172:     def setUp(self):
amine@172:         self.signal = "ABCDEFGHIJKLMNOPQRSTUVWXYZ012345"
amine@173: 
amine@172:     def test_sampling_rate_alias(self):
amine@173:         ads = ADSFactory.ads(
amine@173:             data_buffer=self.signal, sr=16, sample_width=2, channels=1
amine@173:         )
amine@173:         srate = ads.sampling_rate
amine@173:         self.assertEqual(
amine@173:             srate,
amine@173:             16,
amine@173:             "Wrong sampling rate, expected: 16000, found: {0}".format(srate),
amine@173:         )
amine@173: 
amine@172:     def test_sampling_rate_duplicate(self):
amine@173:         func = partial(
amine@173:             ADSFactory.ads,
amine@173:             data_buffer=self.signal,
amine@173:             sr=16,
amine@173:             sampling_rate=16,
amine@173:             sample_width=2,
amine@173:             channels=1,
amine@173:         )
amine@172:         self.assertRaises(DuplicateArgument, func)
amine@173: 
amine@172:     def test_sample_width_alias(self):
amine@173:         ads = ADSFactory.ads(
amine@173:             data_buffer=self.signal, sampling_rate=16, sw=2, channels=1
amine@173:         )
amine@173:         swidth = ads.sample_width
amine@173:         self.assertEqual(
amine@173:             swidth,
amine@173:             2,
amine@173:             "Wrong sample width, expected: 2, found: {0}".format(swidth),
amine@173:         )
amine@173: 
amine@172:     def test_sample_width_duplicate(self):
amine@173:         func = partial(
amine@173:             ADSFactory.ads,
amine@173:             data_buffer=self.signal,
amine@173:             sampling_rate=16,
amine@173:             sw=2,
amine@173:             sample_width=2,
amine@173:             channels=1,
amine@173:         )
amine@172:         self.assertRaises(DuplicateArgument, func)
amine@173: 
amine@172:     def test_channels_alias(self):
amine@173:         ads = ADSFactory.ads(
amine@173:             data_buffer=self.signal, sampling_rate=16, sample_width=2, ch=1
amine@173:         )
amine@173:         channels = ads.channels
amine@173:         self.assertEqual(
amine@173:             channels,
amine@173:             1,
amine@173:             "Wrong number of channels, expected: 1, found: {0}".format(
amine@173:                 channels
amine@173:             ),
amine@173:         )
amine@173: 
amine@172:     def test_channels_duplicate(self):
amine@173:         func = partial(
amine@173:             ADSFactory.ads,
amine@173:             data_buffer=self.signal,
amine@173:             sampling_rate=16,
amine@173:             sample_width=2,
amine@173:             ch=1,
amine@173:             channels=1,
amine@173:         )
amine@172:         self.assertRaises(DuplicateArgument, func)
amine@173: 
amine@172:     def test_block_size_alias(self):
amine@173:         ads = ADSFactory.ads(
amine@173:             data_buffer=self.signal,
amine@173:             sampling_rate=16,
amine@173:             sample_width=2,
amine@173:             channels=1,
amine@173:             bs=8,
amine@173:         )
amine@173:         size = ads.block_size
amine@173:         self.assertEqual(
amine@173:             size,
amine@173:             8,
amine@173:             "Wrong block_size using bs alias, expected: 8, found: {0}".format(
amine@173:                 size
amine@173:             ),
amine@173:         )
amine@173: 
amine@172:     def test_block_size_duplicate(self):
amine@173:         func = partial(
amine@173:             ADSFactory.ads,
amine@173:             data_buffer=self.signal,
amine@173:             sampling_rate=16,
amine@173:             sample_width=2,
amine@173:             channels=1,
amine@173:             bs=4,
amine@173:             block_size=4,
amine@173:         )
amine@172:         self.assertRaises(DuplicateArgument, func)
amine@173: 
amine@172:     def test_block_duration_alias(self):
amine@173:         ads = ADSFactory.ads(
amine@173:             data_buffer=self.signal,
amine@173:             sampling_rate=16,
amine@173:             sample_width=2,
amine@173:             channels=1,
amine@173:             bd=0.75,
amine@173:         )
amine@172:         # 0.75 ms = 0.75 * 16 = 12
amine@173:         size = ads.block_size
amine@173:         self.assertEqual(
amine@173:             size,
amine@173:             12,
amine@173:             "Wrong block_size set with a block_dur alias 'bd', expected: 8, found: {0}".format(
amine@173:                 size
amine@173:             ),
amine@173:         )
amine@173: 
amine@172:     def test_block_duration_duplicate(self):
amine@173:         func = partial(
amine@173:             ADSFactory.ads,
amine@173:             data_buffer=self.signal,
amine@173:             sampling_rate=16,
amine@173:             sample_width=2,
amine@173:             channels=1,
amine@173:             bd=4,
amine@173:             block_dur=4,
amine@173:         )
amine@172:         self.assertRaises(DuplicateArgument, func)
amine@173: 
amine@172:     def test_block_size_duration_duplicate(self):
amine@173:         func = partial(
amine@173:             ADSFactory.ads,
amine@173:             data_buffer=self.signal,
amine@173:             sampling_rate=16,
amine@173:             sample_width=2,
amine@173:             channels=1,
amine@173:             bd=4,
amine@173:             bs=12,
amine@173:         )
amine@172:         self.assertRaises(DuplicateArgument, func)
amine@173: 
amine@172:     def test_hop_duration_alias(self):
amine@173: 
amine@173:         ads = ADSFactory.ads(
amine@173:             data_buffer=self.signal,
amine@173:             sampling_rate=16,
amine@173:             sample_width=2,
amine@173:             channels=1,
amine@173:             bd=0.75,
amine@173:             hd=0.5,
amine@173:         )
amine@172:         size = ads.hop_size
amine@173:         self.assertEqual(
amine@173:             size,
amine@173:             8,
amine@173:             "Wrong block_size using bs alias, expected: 8, found: {0}".format(
amine@173:                 size
amine@173:             ),
amine@173:         )
amine@173: 
amine@172:     def test_hop_duration_duplicate(self):
amine@173: 
amine@173:         func = partial(
amine@173:             ADSFactory.ads,
amine@173:             data_buffer=self.signal,
amine@173:             sampling_rate=16,
amine@173:             sample_width=2,
amine@173:             channels=1,
amine@173:             bd=0.75,
amine@173:             hd=0.5,
amine@173:             hop_dur=0.5,
amine@173:         )
amine@172:         self.assertRaises(DuplicateArgument, func)
amine@173: 
amine@172:     def test_hop_size_duration_duplicate(self):
amine@173:         func = partial(
amine@173:             ADSFactory.ads,
amine@173:             data_buffer=self.signal,
amine@173:             sampling_rate=16,
amine@173:             sample_width=2,
amine@173:             channels=1,
amine@173:             bs=8,
amine@173:             hs=4,
amine@173:             hd=1,
amine@173:         )
amine@172:         self.assertRaises(DuplicateArgument, func)
amine@173: 
amine@172:     def test_hop_size_greater_than_block_size(self):
amine@173:         func = partial(
amine@173:             ADSFactory.ads,
amine@173:             data_buffer=self.signal,
amine@173:             sampling_rate=16,
amine@173:             sample_width=2,
amine@173:             channels=1,
amine@173:             bs=4,
amine@173:             hs=8,
amine@173:         )
amine@172:         self.assertRaises(ValueError, func)
amine@173: 
amine@172:     def test_filename_alias(self):
amine@172:         ads = ADSFactory.ads(fn=dataset.one_to_six_arabic_16000_mono_bc_noise)
amine@173: 
amine@172:     def test_filename_duplicate(self):
amine@173: 
amine@173:         func = partial(
amine@173:             ADSFactory.ads,
amine@173:             fn=dataset.one_to_six_arabic_16000_mono_bc_noise,
amine@173:             filename=dataset.one_to_six_arabic_16000_mono_bc_noise,
amine@173:         )
amine@172:         self.assertRaises(DuplicateArgument, func)
amine@173: 
amine@172:     def test_data_buffer_duplicate(self):
amine@173:         func = partial(
amine@173:             ADSFactory.ads,
amine@173:             data_buffer=self.signal,
amine@173:             db=self.signal,
amine@173:             sampling_rate=16,
amine@173:             sample_width=2,
amine@173:             channels=1,
amine@173:         )
amine@172:         self.assertRaises(DuplicateArgument, func)
amine@173: 
amine@172:     def test_max_time_alias(self):
amine@173:         ads = ADSFactory.ads(
amine@173:             data_buffer=self.signal,
amine@173:             sampling_rate=16,
amine@173:             sample_width=2,
amine@173:             channels=1,
amine@173:             mt=10,
amine@173:         )
amine@173:         self.assertEqual(
amine@173:             ads.max_read,
amine@173:             10,
amine@173:             "Wrong AudioDataSource.max_read, expected: 10, found: {}".format(
amine@173:                 ads.max_read
amine@173:             ),
amine@173:         )
amine@173: 
amine@172:     def test_max_time_duplicate(self):
amine@173:         func = partial(
amine@173:             ADSFactory.ads,
amine@173:             data_buffer=self.signal,
amine@173:             sampling_rate=16,
amine@173:             sample_width=2,
amine@173:             channels=1,
amine@173:             mt=True,
amine@173:             max_time=True,
amine@173:         )
amine@173: 
amine@172:         self.assertRaises(DuplicateArgument, func)
amine@173: 
amine@172:     def test_record_alias(self):
amine@173:         ads = ADSFactory.ads(
amine@173:             data_buffer=self.signal,
amine@173:             sampling_rate=16,
amine@173:             sample_width=2,
amine@173:             channels=1,
amine@173:             rec=True,
amine@173:         )
amine@173:         self.assertTrue(
amine@173:             ads.rewindable, "AudioDataSource.rewindable expected to be True"
amine@173:         )
amine@173: 
amine@172:     def test_record_duplicate(self):
amine@173:         func = partial(
amine@173:             ADSFactory.ads,
amine@173:             data_buffer=self.signal,
amine@173:             sampling_rate=16,
amine@173:             sample_width=2,
amine@173:             channels=1,
amine@173:             rec=True,
amine@173:             record=True,
amine@173:         )
amine@172:         self.assertRaises(DuplicateArgument, func)
amine@173: 
amine@172:     def test_Limiter_Recorder_Overlap_Deco_rewind_and_read_alias(self):
amine@173: 
amine@172:         # Use arbitrary valid block_size and hop_size
amine@172:         block_size = 5
amine@172:         hop_size = 4
amine@173: 
amine@173:         ads = ADSFactory.ads(
amine@173:             db=self.signal,
amine@173:             sr=16,
amine@173:             sw=2,
amine@173:             ch=1,
amine@173:             mt=0.80,
amine@173:             bs=block_size,
amine@173:             hs=hop_size,
amine@173:             rec=True,
amine@173:         )
amine@173: 
amine@172:         # Read all available data overlapping blocks
amine@172:         ads.open()
amine@172:         i = 0
amine@172:         while True:
amine@172:             block = ads.read()
amine@172:             if block is None:
amine@172:                 break
amine@172:             i += 1
amine@173: 
amine@172:         ads.rewind()
amine@173: 
amine@172:         # Build a BufferAudioSource
amine@173:         audio_source = BufferAudioSource(
amine@173:             self.signal, ads.sampling_rate, ads.sample_width, ads.channels
amine@173:         )
amine@172:         audio_source.open()
amine@173: 
amine@173:         # Compare all blocks read from AudioDataSource to those read
amine@173:         # from an audio source with manual position definition
amine@172:         for j in range(i):
amine@172:             tmp = audio_source.read(block_size)
amine@172:             block = ads.read()
amine@173:             self.assertEqual(
amine@173:                 block,
amine@173:                 tmp,
amine@173:                 "Unexpected block (N={0}) read from OverlapADS".format(i),
amine@173:             )
amine@173:             audio_source.set_position((j + 1) * hop_size)
amine@172:         ads.close()
amine@172:         audio_source.close()
amine@173: 
amine@172: 
amine@172: if __name__ == "__main__":
amine@173:     # import sys;sys.argv = ['', 'Test.testName']
amine@172:     unittest.main()