amine@2: ''' amine@2: @author: Amine Sehili amine@2: September 2015 amine@2: amine@2: ''' amine@2: amine@2: import unittest amine@10: from functools import partial amine@10: import sys amine@10: from auditok import dataset, ADSFactory, BufferAudioSource, WaveAudioSource, DuplicateArgument amine@2: import wave amine@2: amine@2: amine@10: try: amine@10: from builtins import range amine@10: except ImportError: amine@10: if sys.version_info < (3, 0): amine@10: range = xrange amine@10: amine@2: class TestADSFactoryFileAudioSource(unittest.TestCase): amine@2: amine@2: def setUp(self): amine@2: self.audio_source = WaveAudioSource(filename=dataset.one_to_six_arabic_16000_mono_bc_noise) amine@2: amine@2: amine@2: def test_ADS_type(self): amine@2: amine@2: ads = ADSFactory.ads(audio_source=self.audio_source) amine@2: amine@2: self.assertIsInstance(ads, ADSFactory.AudioDataSource, amine@2: msg="wrong type for ads object, expected: 'ADSFactory.AudioDataSource', found: {0}".format(type(ads))) amine@2: amine@2: amine@2: def test_default_block_size(self): amine@2: ads = ADSFactory.ads(audio_source=self.audio_source) amine@2: size = ads.get_block_size() amine@2: self.assertEqual(size, 160, "Wrong default block_size, expected: 160, found: {0}".format(size)) amine@2: amine@2: amine@2: def test_block_size(self): amine@2: ads = ADSFactory.ads(audio_source=self.audio_source, block_size=512) amine@2: size = ads.get_block_size() amine@2: self.assertEqual(size, 512, "Wrong block_size, expected: 512, found: {0}".format(size)) amine@10: amine@10: # with alias keyword amine@10: ads = ADSFactory.ads(audio_source=self.audio_source, bs=160) amine@10: size = ads.get_block_size() amine@10: self.assertEqual(size, 160, "Wrong block_size, expected: 160, found: {0}".format(size)) amine@10: amine@10: def test_block_duration(self): amine@10: amine@10: ads = ADSFactory.ads(audio_source=self.audio_source, block_dur=0.01) # 10 ms amine@10: size = ads.get_block_size() amine@10: self.assertEqual(size, 160, "Wrong block_size, expected: 160, found: {0}".format(size)) amine@10: amine@10: # with alias keyword amine@10: ads = ADSFactory.ads(audio_source=self.audio_source, bd=0.025) # 25 ms amine@10: size = ads.get_block_size() amine@10: self.assertEqual(size, 400, "Wrong block_size, expected: 400, found: {0}".format(size)) amine@10: amine@10: def test_hop_duration(self): amine@10: amine@10: ads = ADSFactory.ads(audio_source=self.audio_source, block_dur=0.02, hop_dur=0.01) # 10 ms amine@10: size = ads.hop_size amine@10: self.assertEqual(size, 160, "Wrong hop_size, expected: 160, found: {0}".format(size)) amine@10: amine@10: # with alias keyword amine@10: ads = ADSFactory.ads(audio_source=self.audio_source, bd=0.025, hop_dur=0.015) # 15 ms amine@10: size = ads.hop_size amine@10: self.assertEqual(size, 240, "Wrong block_size, expected: 240, found: {0}".format(size)) amine@10: amine@2: amine@2: def test_sampling_rate(self): amine@2: ads = ADSFactory.ads(audio_source=self.audio_source) amine@2: amine@2: srate = ads.get_sampling_rate() amine@2: self.assertEqual(srate, 16000, "Wrong sampling rate, expected: 16000, found: {0}".format(srate)) amine@2: amine@2: def test_sample_width(self): amine@2: ads = ADSFactory.ads(audio_source=self.audio_source) amine@2: amine@2: swidth = ads.get_sample_width() amine@2: self.assertEqual(swidth, 2, "Wrong sample width, expected: 2, found: {0}".format(swidth)) amine@2: amine@2: def test_channels(self): amine@2: ads = ADSFactory.ads(audio_source=self.audio_source) amine@2: amine@2: channels = ads.get_channels() amine@2: self.assertEqual(channels, 1, "Wrong number of channels, expected: 1, found: {0}".format(channels)) amine@2: amine@2: def test_read(self): amine@2: ads = ADSFactory.ads(audio_source=self.audio_source, block_size = 256) amine@2: amine@2: ads.open() amine@2: ads_data = ads.read() amine@2: ads.close() amine@2: amine@2: audio_source = WaveAudioSource(filename=dataset.one_to_six_arabic_16000_mono_bc_noise) amine@2: audio_source.open() amine@2: audio_source_data = audio_source.read(256) amine@2: audio_source.close() amine@2: amine@2: self.assertEqual(ads_data, audio_source_data, "Unexpected data read from ads") amine@2: amine@2: def test_Limiter_Deco_type(self): amine@2: ads = ADSFactory.ads(audio_source=self.audio_source, max_time=1) amine@2: amine@2: self.assertIsInstance(ads, ADSFactory.LimiterADS, amine@2: msg="wrong type for ads object, expected: 'ADSFactory.LimiterADS', found: {0}".format(type(ads))) amine@2: amine@2: amine@2: def test_Limiter_Deco_read(self): amine@2: # read a maximum of 0.75 seconds from audio source amine@2: ads = ADSFactory.ads(audio_source=self.audio_source, max_time=0.75) amine@2: amine@2: ads_data = [] amine@2: ads.open() amine@2: while True: amine@2: block = ads.read() amine@2: if block is None: amine@2: break amine@2: ads_data.append(block) amine@2: ads.close() amine@10: ads_data = b''.join(ads_data) amine@2: amine@2: audio_source = WaveAudioSource(filename=dataset.one_to_six_arabic_16000_mono_bc_noise) amine@2: audio_source.open() amine@2: audio_source_data = audio_source.read(int(16000 * 0.75)) amine@2: audio_source.close() amine@2: amine@2: self.assertEqual(ads_data, audio_source_data, "Unexpected data read from LimiterADS") amine@2: amine@2: amine@2: def test_Limiter_Deco_read_limit(self): amine@2: # read a maximum of 1.25 seconds from audio source amine@2: ads = ADSFactory.ads(audio_source=self.audio_source, max_time=1.191) amine@2: amine@2: # desired duration into bytes is obtained by: amine@2: # max_time * sampling_rate * sample_width * nb_channels amine@2: # Limiter deco tries to a total quantity of data as amine@2: # possible to the desired duration in bytes. amine@2: # It reads N block of size block_size where: amine@2: # (N - 1) * block_size < desired duration, AND amine@2: # N * block_size >= desired duration amine@2: amine@2: # theoretical size to reach amine@2: expected_size = int(ads.get_sampling_rate() * 1.191) * \ amine@2: ads.get_sample_width() * ads.get_channels() amine@2: amine@2: amine@2: # how much data are required to get N blocks of size block_size amine@2: block_size_bytes = ads.get_block_size() * ads.get_sample_width() * ads.get_channels() amine@2: r = expected_size % block_size_bytes amine@2: if r > 0: amine@2: expected_size += block_size_bytes - r amine@2: amine@2: total_read = 0 amine@2: ads.open() amine@2: i = 0 amine@2: while True: amine@2: block = ads.read() amine@2: if block is None: amine@2: break amine@2: i += 1 amine@2: total_read += len(block) amine@2: amine@2: ads.close() amine@2: amine@2: self.assertEqual(total_read, expected_size, "Wrong data length read from LimiterADS, expected: {0}, found: {1}".format(expected_size, total_read)) amine@2: amine@2: amine@2: amine@2: def test_Recorder_Deco_type(self): amine@2: ads = ADSFactory.ads(audio_source=self.audio_source, record=True) amine@2: amine@2: self.assertIsInstance(ads, ADSFactory.RecorderADS, amine@2: msg="wrong type for ads object, expected: 'ADSFactory.RecorderADS', found: {0}".format(type(ads))) amine@2: amine@2: amine@2: def test_Recorder_Deco_read(self): amine@2: ads = ADSFactory.ads(audio_source=self.audio_source, record=True, block_size=500) amine@2: amine@2: ads_data = [] amine@2: ads.open() amine@10: for i in range(10): amine@2: block = ads.read() amine@2: if block is None: amine@2: break amine@2: ads_data.append(block) amine@2: ads.close() amine@10: ads_data = b''.join(ads_data) amine@2: amine@2: audio_source = WaveAudioSource(filename=dataset.one_to_six_arabic_16000_mono_bc_noise) amine@2: audio_source.open() amine@2: audio_source_data = audio_source.read(500 * 10) amine@2: audio_source.close() amine@2: amine@2: self.assertEqual(ads_data, audio_source_data, "Unexpected data read from RecorderADS") amine@2: amine@2: def test_Recorder_Deco_is_rewindable(self): amine@2: ads = ADSFactory.ads(audio_source=self.audio_source, record=True) amine@2: amine@2: self.assertTrue(ads.is_rewindable(), "RecorderADS.is_rewindable should return True") amine@2: amine@2: amine@2: def test_Recorder_Deco_rewind(self): amine@2: ads = ADSFactory.ads(audio_source=self.audio_source, record=True, block_size = 320) amine@2: amine@2: ads.open() amine@2: ads.read() amine@2: ads.rewind() amine@2: amine@2: amine@2: self.assertIsInstance(ads.get_audio_source(), amine@2: BufferAudioSource, "After rewind RecorderADS.get_audio_source should \ amine@2: be an instance of BufferAudioSource") amine@2: ads.close() amine@2: amine@2: amine@2: def test_Recorder_Deco_rewind_and_read(self): amine@2: ads = ADSFactory.ads(audio_source=self.audio_source, record=True, block_size = 320) amine@2: amine@2: ads.open() amine@10: for i in range(10): amine@2: ads.read() amine@2: amine@2: ads.rewind() amine@2: amine@2: # read all available data after rewind amine@2: ads_data = [] amine@2: while True: amine@2: block = ads.read() amine@2: if block is None: amine@2: break amine@2: ads_data.append(block) amine@2: ads.close() amine@10: ads_data = b''.join(ads_data) amine@2: amine@2: audio_source = WaveAudioSource(filename=dataset.one_to_six_arabic_16000_mono_bc_noise) amine@2: audio_source.open() amine@2: audio_source_data = audio_source.read(320 * 10) amine@2: audio_source.close() amine@2: amine@2: self.assertEqual(ads_data, audio_source_data, "Unexpected data read from RecorderADS") amine@2: amine@2: def test_Overlap_Deco_type(self): amine@2: # an OverlapADS is obtained if a valid hop_size is given amine@2: ads = ADSFactory.ads(audio_source=self.audio_source, block_size = 256, hop_size = 128) amine@2: amine@2: self.assertIsInstance(ads, ADSFactory.OverlapADS, amine@2: msg="wrong type for ads object, expected: 'ADSFactory.OverlapADS', found: {0}".format(type(ads))) amine@2: amine@2: amine@2: amine@2: amine@2: def test_Overlap_Deco_read(self): amine@2: amine@2: # Use arbitrary valid block_size and hop_size amine@2: block_size = 1714 amine@2: hop_size = 313 amine@2: amine@2: ads = ADSFactory.ads(audio_source=self.audio_source, block_size=block_size, hop_size=hop_size) amine@2: amine@2: # Read all available data overlapping blocks amine@2: ads.open() amine@2: ads_data = [] amine@2: while True: amine@2: block = ads.read() amine@2: if block is None: amine@2: break amine@2: ads_data.append(block) amine@2: ads.close() amine@2: amine@2: # Read all data from file and build a BufferAudioSource amine@2: fp = wave.open(dataset.one_to_six_arabic_16000_mono_bc_noise, "r") amine@2: wave_data = fp.readframes(fp.getnframes()) amine@2: fp.close() amine@2: audio_source = BufferAudioSource(wave_data, ads.get_sampling_rate(), amine@2: ads.get_sample_width(), ads.get_channels()) amine@2: audio_source.open() amine@2: amine@2: # Compare all blocks read from OverlapADS to those read amine@2: # from an audio source with a manual set_position amine@2: for i,block in enumerate(ads_data): amine@2: amine@2: tmp = audio_source.read(block_size) amine@2: amine@2: self.assertEqual(block, tmp, "Unexpected block (N={0}) read from OverlapADS".format(i)) amine@2: amine@2: audio_source.set_position((i+1) * hop_size) amine@2: amine@2: audio_source.close() amine@2: amine@2: amine@2: amine@2: amine@2: def test_Limiter_Overlap_Deco_type(self): amine@2: ads = ADSFactory.ads(audio_source=self.audio_source, max_time=1, block_size = 256, hop_size = 128) amine@2: amine@2: self.assertIsInstance(ads, ADSFactory.OverlapADS, amine@2: msg="wrong type for ads object, expected: 'ADSFactory.OverlapADS', found: {0}".format(type(ads))) amine@2: amine@2: amine@2: self.assertIsInstance(ads.ads, ADSFactory.LimiterADS, amine@2: msg="wrong type for ads object, expected: 'ADSFactory.LimiterADS', found: {0}".format(type(ads))) amine@2: amine@2: amine@2: amine@2: def test_Limiter_Overlap_Deco_read(self): amine@2: amine@2: block_size = 256 amine@2: hop_size = 200 amine@2: amine@2: ads = ADSFactory.ads(audio_source=self.audio_source, max_time=0.50, block_size=block_size, hop_size=hop_size) amine@2: amine@2: # Read all available data overlapping blocks amine@2: ads.open() amine@2: ads_data = [] amine@2: while True: amine@2: block = ads.read() amine@2: if block is None: amine@2: break amine@2: ads_data.append(block) amine@2: ads.close() amine@2: amine@2: # Read all data from file and build a BufferAudioSource amine@2: fp = wave.open(dataset.one_to_six_arabic_16000_mono_bc_noise, "r") amine@2: wave_data = fp.readframes(fp.getnframes()) amine@2: fp.close() amine@2: audio_source = BufferAudioSource(wave_data, ads.get_sampling_rate(), amine@2: ads.get_sample_width(), ads.get_channels()) amine@2: audio_source.open() amine@2: amine@2: # Compare all blocks read from OverlapADS to those read amine@2: # from an audio source with a manual set_position amine@2: for i,block in enumerate(ads_data): amine@2: tmp = audio_source.read(block_size) amine@2: amine@2: self.assertEqual(block, tmp, "Unexpected block (N={0}) read from OverlapADS".format(i)) amine@2: amine@2: audio_source.set_position((i+1) * hop_size) amine@2: amine@2: audio_source.close() amine@2: amine@2: amine@2: amine@2: def test_Limiter_Overlap_Deco_read_limit(self): amine@2: amine@2: block_size = 313 amine@2: hop_size = 207 amine@2: ads = ADSFactory.ads(audio_source=self.audio_source, amine@2: max_time=1.932, block_size=block_size, amine@2: hop_size=hop_size) amine@2: amine@2: # Limiter + Overlap decos => read N block of actual data amine@2: # one block of size block_size amine@2: # N - 1 blocks of size hop_size amine@2: # the total size of read data might be a slightly greater amine@2: # than the required size calculated from max_time amine@2: amine@2: # theoretical size to reach amine@2: expected_size = int(ads.get_sampling_rate() * 1.932) * \ amine@2: ads.get_sample_width() * ads.get_channels() amine@2: amine@2: # minus block_size amine@2: expected_size -= (block_size * ads.get_sample_width() * ads.get_channels()) amine@2: amine@2: # how much data are required to get N - 1 blocks of size hop_size amine@2: hop_size_bytes = hop_size * ads.get_sample_width() * ads.get_channels() amine@2: r = expected_size % hop_size_bytes amine@2: if r > 0: amine@2: expected_size += hop_size_bytes - r amine@2: amine@2: expected_size += block_size * ads.get_sample_width() * ads.get_channels() amine@2: amine@2: cache_size = (block_size - hop_size) * ads.get_sample_width() * ads.get_channels() amine@2: total_read = cache_size amine@2: amine@2: ads.open() amine@2: i = 0 amine@2: while True: amine@2: block = ads.read() amine@2: if block is None: amine@2: break amine@2: i += 1 amine@2: total_read += len(block) - cache_size amine@2: amine@2: ads.close() amine@2: self.assertEqual(total_read, expected_size, "Wrong data length read from LimiterADS, expected: {0}, found: {1}".format(expected_size, total_read)) amine@2: amine@2: amine@2: amine@2: def test_Recorder_Overlap_Deco_type(self): amine@2: ads = ADSFactory.ads(audio_source=self.audio_source, block_size=256, hop_size=128, record=True) amine@2: amine@2: self.assertIsInstance(ads, ADSFactory.OverlapADS, amine@2: msg="wrong type for ads object, expected: 'ADSFactory.OverlapADS', found: {0}".format(type(ads))) amine@2: amine@2: amine@2: self.assertIsInstance(ads.ads, ADSFactory.RecorderADS, amine@2: msg="wrong type for ads object, expected: 'ADSFactory.RecorderADS', found: {0}".format(type(ads))) amine@2: amine@2: amine@2: amine@2: def test_Recorder_Overlap_Deco_is_rewindable(self): amine@2: ads = ADSFactory.ads(audio_source=self.audio_source, block_size=320, hop_size=160, record=True) amine@2: self.assertTrue(ads.is_rewindable(), "RecorderADS.is_rewindable should return True") amine@2: amine@2: amine@2: def test_Recorder_Overlap_Deco_rewind_and_read(self): amine@2: amine@2: # Use arbitrary valid block_size and hop_size amine@2: block_size = 1600 amine@2: hop_size = 400 amine@2: amine@2: ads = ADSFactory.ads(audio_source=self.audio_source, block_size=block_size, hop_size=hop_size, record=True) amine@2: amine@2: # Read all available data overlapping blocks amine@2: ads.open() amine@2: i = 0 amine@2: while True: amine@2: block = ads.read() amine@2: if block is None: amine@2: break amine@2: i += 1 amine@2: amine@2: ads.rewind() amine@2: amine@2: # Read all data from file and build a BufferAudioSource amine@2: fp = wave.open(dataset.one_to_six_arabic_16000_mono_bc_noise, "r") amine@2: wave_data = fp.readframes(fp.getnframes()) amine@2: fp.close() amine@2: audio_source = BufferAudioSource(wave_data, ads.get_sampling_rate(), amine@2: ads.get_sample_width(), ads.get_channels()) amine@2: audio_source.open() amine@2: amine@2: # Compare all blocks read from OverlapADS to those read amine@2: # from an audio source with a manual set_position amine@10: for j in range(i): amine@2: amine@2: tmp = audio_source.read(block_size) amine@2: amine@2: self.assertEqual(ads.read(), tmp, "Unexpected block (N={0}) read from OverlapADS".format(i)) amine@2: audio_source.set_position((j+1) * hop_size) amine@2: amine@2: ads.close() amine@2: audio_source.close() amine@2: amine@2: amine@2: def test_Limiter_Recorder_Overlap_Deco_rewind_and_read(self): amine@2: amine@2: # Use arbitrary valid block_size and hop_size amine@2: block_size = 1600 amine@2: hop_size = 400 amine@2: amine@2: ads = ADSFactory.ads(audio_source=self.audio_source, max_time = 1.50, block_size=block_size, hop_size=hop_size, record=True) amine@2: amine@2: # Read all available data overlapping blocks amine@2: ads.open() amine@2: i = 0 amine@2: while True: amine@2: block = ads.read() amine@2: if block is None: amine@2: break amine@2: i += 1 amine@2: amine@2: ads.rewind() amine@2: amine@2: # Read all data from file and build a BufferAudioSource amine@2: fp = wave.open(dataset.one_to_six_arabic_16000_mono_bc_noise, "r") amine@2: wave_data = fp.readframes(fp.getnframes()) amine@2: fp.close() amine@2: audio_source = BufferAudioSource(wave_data, ads.get_sampling_rate(), amine@2: ads.get_sample_width(), ads.get_channels()) amine@2: audio_source.open() amine@2: amine@2: # Compare all blocks read from OverlapADS to those read amine@2: # from an audio source with a manual set_position amine@10: for j in range(i): amine@2: amine@2: tmp = audio_source.read(block_size) amine@2: amine@2: self.assertEqual(ads.read(), tmp, "Unexpected block (N={0}) read from OverlapADS".format(i)) amine@2: audio_source.set_position((j+1) * hop_size) amine@2: amine@2: ads.close() amine@2: audio_source.close() amine@2: amine@2: amine@2: def test_Limiter_Recorder_Overlap_Deco_rewind_and_read_limit(self): amine@2: amine@2: # Use arbitrary valid block_size and hop_size amine@2: block_size = 1000 amine@2: hop_size = 200 amine@2: amine@2: ads = ADSFactory.ads(audio_source=self.audio_source, max_time = 1.317, block_size=block_size, hop_size=hop_size, record=True) amine@2: amine@2: # Limiter + Overlap decos => read N block of actual data amine@2: # one block of size block_size amine@2: # N - 1 blocks of size hop_size amine@2: # the total size of read data might be a slightly greater amine@2: # than the required size calculated from max_time amine@2: amine@2: # theoretical size to reach amine@2: expected_size = int(ads.get_sampling_rate() * 1.317) * \ amine@2: ads.get_sample_width() * ads.get_channels() amine@2: amine@2: # minus block_size amine@2: expected_size -= (block_size * ads.get_sample_width() * ads.get_channels()) amine@2: amine@2: # how much data are required to get N - 1 blocks of size hop_size amine@2: hop_size_bytes = hop_size * ads.get_sample_width() * ads.get_channels() amine@2: r = expected_size % hop_size_bytes amine@2: if r > 0: amine@2: expected_size += hop_size_bytes - r amine@2: amine@2: expected_size += block_size * ads.get_sample_width() * ads.get_channels() amine@2: amine@2: cache_size = (block_size - hop_size) * ads.get_sample_width() * ads.get_channels() amine@2: total_read = cache_size amine@2: amine@2: ads.open() amine@2: i = 0 amine@2: while True: amine@2: block = ads.read() amine@2: if block is None: amine@2: break amine@2: i += 1 amine@2: total_read += len(block) - cache_size amine@2: amine@2: ads.close() amine@2: self.assertEqual(total_read, expected_size, "Wrong data length read from LimiterADS, expected: {0}, found: {1}".format(expected_size, total_read)) amine@2: amine@2: class TestADSFactoryBufferAudioSource(unittest.TestCase): amine@2: amine@2: def setUp(self): amine@2: self.signal = "ABCDEFGHIJKLMNOPQRSTUVWXYZ012345" amine@2: self.ads = ADSFactory.ads(data_buffer=self.signal, sampling_rate=16, amine@2: sample_width=2, channels=1) amine@2: amine@2: def test_ADS_BAS_type(self): amine@2: self.assertIsInstance(self.ads.get_audio_source(), amine@2: BufferAudioSource, "ads should \ amine@2: be an instance of BufferAudioSource") amine@2: amine@2: def test_ADS_BAS_sampling_rate(self): amine@2: srate = self.ads.get_sampling_rate() amine@2: self.assertEqual(srate, 16, "Wrong sampling rate, expected: 16000, found: {0}".format(srate)) amine@2: amine@2: amine@2: def test_ADS_BAS_get_sample_width(self): amine@2: swidth = self.ads.get_sample_width() amine@2: self.assertEqual(swidth, 2, "Wrong sample width, expected: 2, found: {0}".format(swidth)) amine@2: amine@2: def test_ADS_BAS_get_channels(self): amine@2: channels = self.ads.get_channels() amine@2: self.assertEqual(channels, 1, "Wrong number of channels, expected: 1, found: {0}".format(channels)) amine@2: amine@2: def test_Limiter_Recorder_Overlap_Deco_rewind_and_read(self): amine@2: amine@2: # Use arbitrary valid block_size and hop_size amine@2: block_size = 5 amine@2: hop_size = 4 amine@2: amine@2: ads = ADSFactory.ads(data_buffer=self.signal, sampling_rate=16, amine@2: sample_width=2, channels=1, max_time = 0.80, amine@2: block_size=block_size, hop_size=hop_size, amine@2: record=True) amine@2: amine@2: # Read all available data overlapping blocks amine@2: ads.open() amine@2: i = 0 amine@2: while True: amine@2: block = ads.read() amine@2: if block is None: amine@2: break amine@2: i += 1 amine@2: amine@2: ads.rewind() amine@2: amine@2: # Build a BufferAudioSource amine@2: audio_source = BufferAudioSource(self.signal, ads.get_sampling_rate(), amine@2: ads.get_sample_width(), ads.get_channels()) amine@2: audio_source.open() amine@2: amine@2: # Compare all blocks read from OverlapADS to those read amine@2: # from an audio source with a manual set_position amine@10: for j in range(i): amine@10: amine@10: tmp = audio_source.read(block_size) amine@10: amine@10: block = ads.read() amine@10: amine@10: self.assertEqual(block, tmp, "Unexpected block (N={0}) read from OverlapADS".format(i)) amine@10: audio_source.set_position((j+1) * hop_size) amine@10: amine@10: ads.close() amine@10: audio_source.close() amine@10: amine@10: amine@10: class TestADSFactoryAlias(unittest.TestCase): amine@10: amine@10: def setUp(self): amine@10: self.signal = "ABCDEFGHIJKLMNOPQRSTUVWXYZ012345" amine@10: amine@10: def test_sampling_rate_alias(self): amine@10: ads = ADSFactory.ads(data_buffer=self.signal, sr=16, amine@10: sample_width=2, channels=1) amine@10: srate = ads.get_sampling_rate() amine@10: self.assertEqual(srate, 16, "Wrong sampling rate, expected: 16000, found: {0}".format(srate)) amine@10: amine@10: def test_sampling_rate_duplicate(self): amine@10: func = partial(ADSFactory.ads, data_buffer=self.signal, sr=16, sampling_rate=16, amine@10: sample_width=2, channels=1) amine@10: self.assertRaises(DuplicateArgument, func) amine@10: amine@10: def test_sample_width_alias(self): amine@10: ads = ADSFactory.ads(data_buffer=self.signal, sampling_rate=16, amine@10: sw=2, channels=1) amine@10: swidth = ads.get_sample_width() amine@10: self.assertEqual(swidth, 2, "Wrong sample width, expected: 2, found: {0}".format(swidth)) amine@10: amine@10: def test_sample_width_duplicate(self): amine@10: func = partial(ADSFactory.ads, data_buffer=self.signal,sampling_rate=16, amine@10: sw=2, sample_width=2, channels=1) amine@10: self.assertRaises(DuplicateArgument, func) amine@10: amine@10: def test_channels_alias(self): amine@10: ads = ADSFactory.ads(data_buffer=self.signal, sampling_rate=16, amine@10: sample_width=2, ch=1) amine@10: channels = ads.get_channels() amine@10: self.assertEqual(channels, 1, "Wrong number of channels, expected: 1, found: {0}".format(channels)) amine@10: amine@10: def test_channels_duplicate(self): amine@10: func = partial(ADSFactory.ads, data_buffer=self.signal,sampling_rate=16, amine@10: sample_width=2, ch=1, channels=1) amine@10: self.assertRaises(DuplicateArgument, func) amine@10: amine@10: amine@10: def test_block_size_alias(self): amine@10: ads = ADSFactory.ads(data_buffer=self.signal, sampling_rate=16, amine@10: sample_width=2, channels=1, bs=8) amine@10: size = ads.get_block_size() amine@10: self.assertEqual(size, 8, "Wrong block_size using bs alias, expected: 8, found: {0}".format(size)) amine@10: amine@10: def test_block_size_duplicate(self): amine@10: func = partial(ADSFactory.ads, data_buffer=self.signal,sampling_rate=16, amine@10: sample_width=2, channels=1, bs=4, block_size=4) amine@10: self.assertRaises(DuplicateArgument, func) amine@10: amine@10: def test_block_duration_alias(self): amine@10: ads = ADSFactory.ads(data_buffer=self.signal, sampling_rate=16, amine@10: sample_width=2, channels=1, bd=0.75) amine@10: # 0.75 ms = 0.75 * 16 = 12 amine@10: size = ads.get_block_size() amine@10: self.assertEqual(size, 12, "Wrong block_size set with a block_dur alias 'bd', expected: 8, found: {0}".format(size)) amine@10: amine@10: def test_block_duration_duplicate(self): amine@10: func = partial(ADSFactory.ads, data_buffer=self.signal,sampling_rate=16, amine@10: sample_width=2, channels=1, bd=4, block_dur=4) amine@10: self.assertRaises(DuplicateArgument, func) amine@10: amine@10: def test_block_size_duration_duplicate(self): amine@10: func = partial(ADSFactory.ads, data_buffer=self.signal,sampling_rate=16, amine@10: sample_width=2, channels=1, bd=4, bs=12) amine@10: self.assertRaises(DuplicateArgument, func) amine@10: amine@10: def test_hop_duration_alias(self): amine@10: amine@10: ads = ADSFactory.ads(data_buffer=self.signal, sampling_rate=16, amine@10: sample_width=2, channels=1, bd=0.75, hd=0.5 ) amine@10: size = ads.hop_size amine@10: self.assertEqual(size, 8, "Wrong block_size using bs alias, expected: 8, found: {0}".format(size)) amine@10: self.assertIsInstance(ads, ADSFactory.OverlapADS, "ads expected to an ADSFactory.OverlapADS object") amine@10: amine@10: amine@10: def test_hop_duration_duplicate(self): amine@10: amine@10: func = partial(ADSFactory.ads, data_buffer=self.signal, sampling_rate=16, amine@10: sample_width=2, channels=1, bd=0.75, hd=0.5, hop_dur=0.5) amine@10: self.assertRaises(DuplicateArgument, func) amine@10: amine@10: amine@10: def test_hop_size_duration_duplicate(self): amine@10: func = partial(ADSFactory.ads, data_buffer=self.signal,sampling_rate=16, amine@10: sample_width=2, channels=1, bs=8, hs=4, hd=1) amine@10: self.assertRaises(DuplicateArgument, func) amine@10: amine@10: amine@10: def test_hop_size_greater_than_block_size(self): amine@10: func = partial(ADSFactory.ads, data_buffer=self.signal, sampling_rate=16, amine@10: sample_width=2, channels=1, bs=4, hs=8) amine@10: self.assertRaises(ValueError, func) amine@10: amine@10: amine@10: def test_filename_alias(self): amine@10: ads = ADSFactory.ads(fn=dataset.one_to_six_arabic_16000_mono_bc_noise) amine@10: amine@10: amine@10: def test_filename_duplicate(self): amine@10: amine@10: func = partial(ADSFactory.ads, fn=dataset.one_to_six_arabic_16000_mono_bc_noise, filename=dataset.one_to_six_arabic_16000_mono_bc_noise) amine@10: self.assertRaises(DuplicateArgument, func) amine@10: amine@10: amine@10: def test_data_buffer_alias(self): amine@10: ads = ADSFactory.ads(db=self.signal, sampling_rate=16, amine@10: sample_width=2, channels=1) amine@10: self.assertEqual(ads.get_audio_source().get_data_buffer(), self.signal, "Wrong value for data buffer") amine@10: amine@10: amine@10: def test_data_buffer_duplicate(self): amine@10: func = partial(ADSFactory.ads, data_buffer=self.signal, db=self.signal, sampling_rate=16, amine@10: sample_width=2, channels=1) amine@10: self.assertRaises(DuplicateArgument, func) amine@10: amine@10: amine@10: def test_max_time_alias(self): amine@10: ads = ADSFactory.ads(data_buffer=self.signal, sampling_rate=16, amine@10: sample_width=2, channels=1, mt=10) amine@10: self.assertIsInstance(ads, ADSFactory.LimiterADS, "ads expected to an ADSFactory.LimiterADS object") amine@10: amine@10: amine@10: def test_max_time_duplicate(self): amine@10: func = partial(ADSFactory.ads, data_buffer=self.signal, sampling_rate=16, amine@10: sample_width=2, channels=1, mt=True, max_time=True) amine@10: amine@10: self.assertRaises(DuplicateArgument, func) amine@10: amine@10: def test_record_alias(self): amine@10: ads = ADSFactory.ads(data_buffer=self.signal, sampling_rate=16, amine@10: sample_width=2, channels=1, rec=True) amine@10: self.assertIsInstance(ads, ADSFactory.RecorderADS, "ads expected to an ADSFactory.RecorderADS object") amine@10: amine@10: amine@10: def test_record_duplicate(self): amine@10: func = partial(ADSFactory.ads, data_buffer=self.signal, sampling_rate=16, amine@10: sample_width=2, channels=1, rec=True, record=True) amine@10: self.assertRaises(DuplicateArgument, func) amine@10: amine@10: amine@10: def test_Limiter_Recorder_Overlap_Deco_rewind_and_read_alias(self): amine@10: amine@10: # Use arbitrary valid block_size and hop_size amine@10: block_size = 5 amine@10: hop_size = 4 amine@10: amine@10: ads = ADSFactory.ads(db=self.signal, sr=16, amine@10: sw=2, ch=1, mt = 0.80, amine@10: bs=block_size, hs=hop_size, amine@10: rec=True) amine@10: amine@10: # Read all available data overlapping blocks amine@10: ads.open() amine@10: i = 0 amine@10: while True: amine@10: block = ads.read() amine@10: if block is None: amine@10: break amine@10: i += 1 amine@10: amine@10: ads.rewind() amine@10: amine@10: # Build a BufferAudioSource amine@10: audio_source = BufferAudioSource(self.signal, ads.get_sampling_rate(), amine@10: ads.get_sample_width(), ads.get_channels()) amine@10: audio_source.open() amine@10: amine@10: # Compare all blocks read from OverlapADS to those read amine@10: # from an audio source with a manual set_position amine@10: for j in range(i): amine@2: amine@2: tmp = audio_source.read(block_size) amine@2: amine@2: block = ads.read() amine@2: amine@2: self.assertEqual(block, tmp, "Unexpected block (N={0}) read from OverlapADS".format(i)) amine@2: audio_source.set_position((j+1) * hop_size) amine@2: amine@2: ads.close() amine@2: audio_source.close() amine@2: amine@2: amine@2: if __name__ == "__main__": amine@2: #import sys;sys.argv = ['', 'Test.testName'] amine@2: unittest.main()