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