amine@173: """ amine@172: @author: Amine Sehili 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@293: import wave amine@293: from genty import genty, genty_dataset amine@173: from auditok import ( amine@173: dataset, amine@173: ADSFactory, amine@173: AudioDataSource, amine@296: AudioReader, amine@296: Recorder, amine@173: BufferAudioSource, amine@173: WaveAudioSource, amine@173: DuplicateArgument, amine@173: ) 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@216: # from an audio source with a manual position setting 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@216: audio_source.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@216: # from an audio source with a manual position setting 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@216: audio_source.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@216: # from an audio source with a manual position setting 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@216: audio_source.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@216: # from an audio source with a manual position setting 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@216: audio_source.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@218: block_size=4, 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@216: # from an audio source with a manual position setting 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@216: audio_source.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@294: self.signal = b"ABCDEFGHIJKLMNOPQRSTUVWXYZ012345" amine@173: amine@172: def test_sampling_rate_alias(self): amine@173: ads = ADSFactory.ads( amine@218: data_buffer=self.signal, amine@218: sr=16, amine@218: sample_width=2, amine@218: channels=1, amine@218: block_dur=0.5, 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@218: data_buffer=self.signal, amine@218: sampling_rate=16, amine@218: sw=2, amine@218: channels=1, amine@218: block_dur=0.5, 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@218: data_buffer=self.signal, amine@218: sampling_rate=16, amine@218: sample_width=2, amine@218: ch=1, amine@218: block_dur=4, 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@218: block_dur=0.5, 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@218: block_dur=0.5, 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@216: audio_source.position = (j + 1) * hop_size amine@172: ads.close() amine@172: audio_source.close() amine@173: amine@172: amine@294: def _read_all_data(reader): amine@294: blocks = [] amine@294: while True: amine@294: data = reader.read() amine@294: if data is None: amine@294: break amine@294: blocks.append(data) amine@294: return b"".join(blocks) amine@294: amine@294: amine@293: @genty amine@293: class TestAudioReader(unittest.TestCase): amine@293: amine@293: # TODO move all tests here when backward compatibility amine@293: # with ADSFactory is dropped amine@293: amine@293: @genty_dataset( amine@293: mono=("mono_400", 0.5, 16000), amine@293: multichannel=("3channel_400-800-1600", 0.5, 16000 * 3), amine@293: ) amine@293: def test_Limiter(self, file_id, max_read, size): amine@293: input_wav = "tests/data/test_16KHZ_{}Hz.wav".format(file_id) amine@293: input_raw = "tests/data/test_16KHZ_{}Hz.raw".format(file_id) amine@293: with open(input_raw, "rb") as fp: amine@293: expected = fp.read(size) amine@293: amine@296: reader = AudioReader(input_wav, block_dur=0.1, max_read=max_read) amine@293: reader.open() amine@294: data = _read_all_data(reader) amine@294: reader.close() amine@293: self.assertEqual(data, expected) amine@293: amine@294: @genty_dataset(mono=("mono_400",), multichannel=("3channel_400-800-1600",)) amine@294: def test_Recorder(self, file_id): amine@294: input_wav = "tests/data/test_16KHZ_{}Hz.wav".format(file_id) amine@294: input_raw = "tests/data/test_16KHZ_{}Hz.raw".format(file_id) amine@294: with open(input_raw, "rb") as fp: amine@294: expected = fp.read() amine@294: amine@296: reader = AudioReader(input_wav, block_dur=0.1, record=True) amine@294: reader.open() amine@294: data = _read_all_data(reader) amine@294: self.assertEqual(data, expected) amine@294: amine@294: # rewind many times amine@294: for _ in range(3): amine@294: reader.rewind() amine@294: data = _read_all_data(reader) amine@294: self.assertEqual(data, expected) amine@294: self.assertEqual(data, reader.data) amine@294: reader.close() amine@294: amine@296: @genty_dataset(mono=("mono_400",), multichannel=("3channel_400-800-1600",)) amine@296: def test_Recorder_alias(self, file_id): amine@296: input_wav = "tests/data/test_16KHZ_{}Hz.wav".format(file_id) amine@296: input_raw = "tests/data/test_16KHZ_{}Hz.raw".format(file_id) amine@296: with open(input_raw, "rb") as fp: amine@296: expected = fp.read() amine@296: amine@296: reader = Recorder(input_wav, block_dur=0.1) amine@296: reader.open() amine@296: data = _read_all_data(reader) amine@296: self.assertEqual(data, expected) amine@296: amine@296: # rewind many times amine@296: for _ in range(3): amine@296: reader.rewind() amine@296: data = _read_all_data(reader) amine@296: self.assertEqual(data, expected) amine@296: self.assertEqual(data, reader.data) amine@296: reader.close() amine@296: amine@293: amine@172: if __name__ == "__main__": amine@172: unittest.main()