amine@164: """
amine@164: @author: Amine Sehili <amine.sehili@gmail.com>
amine@164: """
amine@164: from array import array
amine@164: import unittest
amine@164: from genty import genty, genty_dataset
amine@164: from auditok.io import (
amine@164:     AudioParameterError,
amine@164:     _array_to_bytes,
amine@164:     DATA_FORMAT,
amine@164:     BufferAudioSource,
amine@164:     RawAudioSource,
amine@164:     WaveAudioSource,
amine@164: )
amine@164: from test_util import PURE_TONE_DICT
amine@164: 
amine@164: 
amine@164: def audio_source_read_all_gen(audio_source, size=None):
amine@164:     if size is None:
amine@164:         size = int(audio_source.sr * 0.1)  # 100ms
amine@164:     while True:
amine@164:         data = audio_source.read(size)
amine@164:         if data is None:
amine@164:             break
amine@164:         yield data
amine@164: 
amine@164: 
amine@164: @genty
amine@164: class TestAudioSource(unittest.TestCase):
amine@164: 
amine@164:     # TODO when use_channel is None, return samples from all channels
amine@164: 
amine@164:     @genty_dataset(
amine@164:         mono_default=("mono_400Hz", 1, None, 400),
amine@164:         mono_mix=("mono_400Hz", 1, "mix", 400),
amine@164:         mono_channel_selection=("mono_400Hz", 1, 2, 400),
amine@164:         multichannel_default=("3channel_400-800-1600Hz", 3, None, 400),
amine@208:         multichannel_channel_select_1st=("3channel_400-800-1600Hz", 3, 1, 400),
amine@208:         multichannel_channel_select_2nd=("3channel_400-800-1600Hz", 3, 2, 800),
amine@208:         multichannel_channel_select_3rd=("3channel_400-800-1600Hz", 3, 3, 1600),
amine@164:     )
amine@164:     def test_RawAudioSource(
amine@164:         self, file_suffix, channels, use_channel, frequency
amine@164:     ):
amine@164:         file = "tests/data/test_16KHZ_{}.raw".format(file_suffix)
amine@164:         audio_source = RawAudioSource(file, 16000, 2, channels, use_channel)
amine@164:         audio_source.open()
amine@164:         data = b"".join(audio_source_read_all_gen(audio_source))
amine@164:         audio_source.close()
amine@164:         expected = _array_to_bytes(PURE_TONE_DICT[frequency])
amine@164:         self.assertEqual(data, expected)
amine@164: 
amine@164:     def test_RawAudioSource_mix(self):
amine@164:         file = "tests/data/test_16KHZ_3channel_400-800-1600Hz.raw"
amine@164:         audio_source = RawAudioSource(file, 16000, 2, 3, use_channel="mix")
amine@164:         audio_source.open()
amine@164:         data = b"".join(audio_source_read_all_gen(audio_source))
amine@164:         audio_source.close()
amine@164: 
amine@164:         mono_channels = [PURE_TONE_DICT[freq] for freq in [400, 800, 1600]]
amine@164:         fmt = DATA_FORMAT[2]
amine@164:         expected = _array_to_bytes(
amine@164:             array(fmt, (sum(samples) // 3 for samples in zip(*mono_channels)))
amine@164:         )
amine@164:         expected = expected
amine@164:         self.assertEqual(data, expected)
amine@164: 
amine@164:     @genty_dataset(
amine@164:         mono_default=("mono_400Hz", 1, None, 400),
amine@164:         mono_mix=("mono_400Hz", 1, "mix", 400),
amine@164:         mono_channel_selection=("mono_400Hz", 1, 2, 400),
amine@164:         multichannel_default=("3channel_400-800-1600Hz", 3, None, 400),
amine@208:         multichannel_channel_select_1st=("3channel_400-800-1600Hz", 3, 1, 400),
amine@208:         multichannel_channel_select_2nd=("3channel_400-800-1600Hz", 3, 2, 800),
amine@208:         multichannel_channel_select_3rd=("3channel_400-800-1600Hz", 3, 3, 1600),
amine@164:     )
amine@164:     def test_WaveAudioSource(
amine@164:         self, file_suffix, channels, use_channel, frequency
amine@164:     ):
amine@164:         file = "tests/data/test_16KHZ_{}.wav".format(file_suffix)
amine@164:         audio_source = WaveAudioSource(file, use_channel)
amine@164:         audio_source.open()
amine@164:         data = b"".join(audio_source_read_all_gen(audio_source))
amine@164:         audio_source.close()
amine@164:         expected = _array_to_bytes(PURE_TONE_DICT[frequency])
amine@164:         self.assertEqual(data, expected)
amine@164: 
amine@164:     def test_WaveAudioSource_mix(self):
amine@164:         file = "tests/data/test_16KHZ_3channel_400-800-1600Hz.wav"
amine@164:         audio_source = WaveAudioSource(file, use_channel="mix")
amine@164:         audio_source.open()
amine@164:         data = b"".join(audio_source_read_all_gen(audio_source))
amine@164:         audio_source.close()
amine@164: 
amine@164:         mono_channels = [PURE_TONE_DICT[freq] for freq in [400, 800, 1600]]
amine@164:         fmt = DATA_FORMAT[2]
amine@164:         expected = _array_to_bytes(
amine@164:             array(fmt, (sum(samples) // 3 for samples in zip(*mono_channels)))
amine@164:         )
amine@164:         self.assertEqual(data, expected)
amine@164: 
amine@164: 
amine@166: @genty
amine@164: class TestBufferAudioSource_SR10_SW1_CH1(unittest.TestCase):
amine@164:     def setUp(self):
amine@164:         self.data = b"ABCDEFGHIJKLMNOPQRSTUVWXYZ012345"
amine@164:         self.audio_source = BufferAudioSource(
amine@164:             data_buffer=self.data, sampling_rate=10, sample_width=1, channels=1
amine@164:         )
amine@164:         self.audio_source.open()
amine@164: 
amine@164:     def tearDown(self):
amine@164:         self.audio_source.close()
amine@164: 
amine@164:     def test_sr10_sw1_ch1_read_1(self):
amine@164:         block = self.audio_source.read(1)
amine@164:         exp = b"A"
amine@164:         self.assertEqual(
amine@164:             block,
amine@164:             exp,
amine@164:             msg="wrong block, expected: {}, found: {} ".format(exp, block),
amine@164:         )
amine@164: 
amine@164:     def test_sr10_sw1_ch1_read_6(self):
amine@164:         block = self.audio_source.read(6)
amine@164:         exp = b"ABCDEF"
amine@164:         self.assertEqual(
amine@164:             block,
amine@164:             exp,
amine@164:             msg="wrong block, expected: {}, found: {} ".format(exp, block),
amine@164:         )
amine@164: 
amine@164:     def test_sr10_sw1_ch1_read_multiple(self):
amine@164:         block = self.audio_source.read(1)
amine@164:         exp = b"A"
amine@164:         self.assertEqual(
amine@164:             block,
amine@164:             exp,
amine@164:             msg="wrong block, expected: {}, found: {} ".format(exp, block),
amine@164:         )
amine@164: 
amine@164:         block = self.audio_source.read(6)
amine@164:         exp = b"BCDEFG"
amine@164:         self.assertEqual(
amine@164:             block,
amine@164:             exp,
amine@164:             msg="wrong block, expected: {}, found: {} ".format(exp, block),
amine@164:         )
amine@164: 
amine@164:         block = self.audio_source.read(13)
amine@164:         exp = b"HIJKLMNOPQRST"
amine@164:         self.assertEqual(
amine@164:             block,
amine@164:             exp,
amine@164:             msg="wrong block, expected: {}, found: {} ".format(exp, block),
amine@164:         )
amine@164: 
amine@164:         block = self.audio_source.read(9999)
amine@164:         exp = b"UVWXYZ012345"
amine@164:         self.assertEqual(
amine@164:             block,
amine@164:             exp,
amine@164:             msg="wrong block, expected: {}, found: {} ".format(exp, block),
amine@164:         )
amine@164: 
amine@164:     def test_sr10_sw1_ch1_read_all(self):
amine@164:         block = self.audio_source.read(9999)
amine@164:         self.assertEqual(
amine@164:             block,
amine@164:             self.data,
amine@164:             msg="wrong block, expected: {}, found: {} ".format(
amine@164:                 self.data, block
amine@164:             ),
amine@164:         )
amine@164: 
amine@164:         block = self.audio_source.read(1)
amine@164:         self.assertEqual(
amine@164:             block,
amine@164:             None,
amine@164:             msg="wrong block, expected: {}, found: {} ".format(None, block),
amine@164:         )
amine@164: 
amine@164:     def test_sr10_sw1_ch1_get_sampling_rate(self):
amine@164:         srate = self.audio_source.get_sampling_rate()
amine@164:         self.assertEqual(
amine@164:             srate,
amine@164:             10,
amine@164:             msg="wrong sampling rate, expected: 10, found: {0} ".format(srate),
amine@164:         )
amine@164: 
amine@164:     def test_sr10_sw1_ch1_get_sample_width(self):
amine@164:         swidth = self.audio_source.get_sample_width()
amine@164:         self.assertEqual(
amine@164:             swidth,
amine@164:             1,
amine@164:             msg="wrong sample width, expected: 1, found: {0} ".format(swidth),
amine@164:         )
amine@164: 
amine@164:     def test_sr10_sw1_ch1_get_channels(self):
amine@164:         channels = self.audio_source.get_channels()
amine@164:         self.assertEqual(
amine@164:             channels,
amine@164:             1,
amine@164:             msg="wrong number of channels, expected: 1, found: {0} ".format(
amine@164:                 channels
amine@164:             ),
amine@164:         )
amine@164: 
amine@166:     @genty_dataset(
amine@166:         empty=([], 0, 0, 0),
amine@166:         zero=([0], 0, 0, 0),
amine@166:         five=([5], 5, 0.5, 500),
amine@166:         multiple=([5, 20], 25, 2.5, 2500),
amine@166:     )
amine@166:     def test_position(
amine@166:         self, block_sizes, expected_sample, expected_second, expected_ms
amine@166:     ):
amine@166:         for block_size in block_sizes:
amine@166:             self.audio_source.read(block_size)
amine@166:         position = self.audio_source.position
amine@166:         self.assertEqual(
amine@166:             position,
amine@166:             expected_sample,
amine@166:             msg="wrong stream position, expected: {}, found: {}".format(
amine@166:                 expected_sample, position
amine@166:             ),
amine@166:         )
amine@166: 
amine@166:         position_s = self.audio_source.position_s
amine@166:         self.assertEqual(
amine@166:             position_s,
amine@166:             expected_second,
amine@166:             msg="wrong stream position_s, expected: {}, found: {}".format(
amine@166:                 expected_second, position_s
amine@166:             ),
amine@166:         )
amine@166: 
amine@166:         position_ms = self.audio_source.position_ms
amine@166:         self.assertEqual(
amine@166:             position_ms,
amine@166:             expected_ms,
amine@166:             msg="wrong stream position_s, expected: {}, found: {}".format(
amine@166:                 expected_ms, position_ms
amine@166:             ),
amine@166:         )
amine@166: 
amine@166:     @genty_dataset(
amine@166:         zero=(0, 0, 0, 0),
amine@166:         one=(1, 1, 0.1, 100),
amine@166:         ten=(10, 10, 1, 1000),
amine@166:         negative_1=(-1, 31, 3.1, 3100),
amine@166:         negative_2=(-7, 25, 2.5, 2500),
amine@166:     )
amine@166:     def test_position_setter(
amine@166:         self, position, expected_sample, expected_second, expected_ms
amine@166:     ):
amine@166:         self.audio_source.position = position
amine@166: 
amine@166:         position = self.audio_source.position
amine@166:         self.assertEqual(
amine@166:             position,
amine@166:             expected_sample,
amine@166:             msg="wrong stream position, expected: {}, found: {}".format(
amine@166:                 expected_sample, position
amine@166:             ),
amine@166:         )
amine@166: 
amine@166:         position_s = self.audio_source.position_s
amine@166:         self.assertEqual(
amine@166:             position_s,
amine@166:             expected_second,
amine@166:             msg="wrong stream position_s, expected: {}, found: {}".format(
amine@166:                 expected_second, position_s
amine@166:             ),
amine@166:         )
amine@166: 
amine@166:         position_ms = self.audio_source.position_ms
amine@166:         self.assertEqual(
amine@166:             position_ms,
amine@166:             expected_ms,
amine@166:             msg="wrong stream position_s, expected: {}, found: {}".format(
amine@166:                 expected_ms, position_ms
amine@166:             ),
amine@166:         )
amine@166: 
amine@166:     @genty_dataset(
amine@166:         zero=(0, 0, 0, 0),
amine@166:         one=(0.1, 1, 0.1, 100),
amine@166:         ten=(1, 10, 1, 1000),
amine@166:         negative_1=(-0.1, 31, 3.1, 3100),
amine@166:         negative_2=(-0.7, 25, 2.5, 2500),
amine@166:     )
amine@166:     def test_position_s_setter(
amine@166:         self, position_s, expected_sample, expected_second, expected_ms
amine@166:     ):
amine@166:         self.audio_source.position_s = position_s
amine@166: 
amine@166:         position = self.audio_source.position
amine@166:         self.assertEqual(
amine@166:             position,
amine@166:             expected_sample,
amine@166:             msg="wrong stream position, expected: {}, found: {}".format(
amine@166:                 expected_sample, position
amine@166:             ),
amine@166:         )
amine@166: 
amine@166:         position_s = self.audio_source.position_s
amine@166:         self.assertEqual(
amine@166:             position_s,
amine@166:             expected_second,
amine@166:             msg="wrong stream position_s, expected: {}, found: {}".format(
amine@166:                 expected_second, position_s
amine@166:             ),
amine@166:         )
amine@166: 
amine@166:         position_ms = self.audio_source.position_ms
amine@166:         self.assertEqual(
amine@166:             position_ms,
amine@166:             expected_ms,
amine@166:             msg="wrong stream position_s, expected: {}, found: {}".format(
amine@166:                 expected_ms, position_ms
amine@166:             ),
amine@166:         )
amine@166: 
amine@166:     @genty_dataset(
amine@166:         zero=(0, 0, 0, 0),
amine@166:         one=(100, 1, 0.1, 100),
amine@166:         ten=(1000, 10, 1, 1000),
amine@166:         negative_1=(-100, 31, 3.1, 3100),
amine@166:         negative_2=(-700, 25, 2.5, 2500),
amine@166:     )
amine@166:     def test_position_ms_setter(
amine@166:         self, position_ms, expected_sample, expected_second, expected_ms
amine@166:     ):
amine@166:         self.audio_source.position_ms = position_ms
amine@166: 
amine@166:         position = self.audio_source.position
amine@166:         self.assertEqual(
amine@166:             position,
amine@166:             expected_sample,
amine@166:             msg="wrong stream position, expected: {}, found: {}".format(
amine@166:                 expected_sample, position
amine@166:             ),
amine@166:         )
amine@166: 
amine@166:         position_s = self.audio_source.position_s
amine@166:         self.assertEqual(
amine@166:             position_s,
amine@166:             expected_second,
amine@166:             msg="wrong stream position_s, expected: {}, found: {}".format(
amine@166:                 expected_second, position_s
amine@166:             ),
amine@166:         )
amine@166: 
amine@166:         position_ms = self.audio_source.position_ms
amine@166:         self.assertEqual(
amine@166:             position_ms,
amine@166:             expected_ms,
amine@166:             msg="wrong stream position_s, expected: {}, found: {}".format(
amine@166:                 expected_ms, position_ms
amine@166:             ),
amine@166:         )
amine@166: 
amine@166:     @genty_dataset(positive=((100,)), negative=(-100,))
amine@166:     def test_position_setter_out_of_range(self, position):
amine@166:         with self.assertRaises(IndexError):
amine@166:             self.audio_source.position = position
amine@166: 
amine@166:     @genty_dataset(positive=((100,)), negative=(-100,))
amine@166:     def test_position_s_setter_out_of_range(self, position_s):
amine@166:         with self.assertRaises(IndexError):
amine@166:             self.audio_source.position_s = position_s
amine@166: 
amine@166:     @genty_dataset(positive=((10000,)), negative=(-10000,))
amine@166:     def test_position_ms_setter_out_of_range(self, position_ms):
amine@166:         with self.assertRaises(IndexError):
amine@166:             self.audio_source.position_ms = position_ms
amine@166: 
amine@164:     def test_sr10_sw1_ch1_get_time_position_0(self):
amine@164:         tp = self.audio_source.get_time_position()
amine@164:         self.assertEqual(
amine@164:             tp,
amine@164:             0.0,
amine@164:             msg="wrong time position, expected: 0.0, found: {0} ".format(tp),
amine@164:         )
amine@164: 
amine@164:     def test_sr10_sw1_ch1_get_time_position_1(self):
amine@164:         srate = self.audio_source.get_sampling_rate()
amine@164:         # read one second
amine@164:         self.audio_source.read(srate)
amine@164:         tp = self.audio_source.get_time_position()
amine@164:         self.assertEqual(
amine@164:             tp,
amine@164:             1.0,
amine@164:             msg="wrong time position, expected: 1.0, found: {0} ".format(tp),
amine@164:         )
amine@164: 
amine@164:     def test_sr10_sw1_ch1_get_time_position_2_5(self):
amine@164:         # read 2.5 seconds
amine@164:         self.audio_source.read(25)
amine@164:         tp = self.audio_source.get_time_position()
amine@164:         self.assertEqual(
amine@164:             tp,
amine@164:             2.5,
amine@164:             msg="wrong time position, expected: 2.5, found: {0} ".format(tp),
amine@164:         )
amine@164: 
amine@164:     def test_sr10_sw1_ch1_set_time_position_0(self):
amine@164:         self.audio_source.read(10)
amine@164:         self.audio_source.set_time_position(0)
amine@164:         tp = self.audio_source.get_time_position()
amine@164:         self.assertEqual(
amine@164:             tp,
amine@164:             0.0,
amine@164:             msg="wrong time position, expected: 0.0, found: {0} ".format(tp),
amine@164:         )
amine@164: 
amine@164:     def test_sr10_sw1_ch1_set_time_position_1(self):
amine@164:         self.audio_source.set_time_position(1)
amine@164:         tp = self.audio_source.get_time_position()
amine@164:         self.assertEqual(
amine@164:             tp,
amine@164:             1.0,
amine@164:             msg="wrong time position, expected: 1.0, found: {0} ".format(tp),
amine@164:         )
amine@164: 
amine@164:     def test_sr10_sw1_ch1_rewind(self):
amine@164:         self.audio_source.read(10)
amine@164:         self.audio_source.rewind()
amine@169:         tp = self.audio_source.position
amine@164:         self.assertEqual(
amine@164:             tp, 0, msg="wrong position, expected: 0.0, found: {0} ".format(tp)
amine@164:         )
amine@164: 
amine@164:     def test_sr10_sw1_ch1_set_data(self):
amine@164:         self.audio_source.set_data(b"12345")
amine@164:         block = self.audio_source.read(9999)
amine@164:         self.assertEqual(
amine@164:             block,
amine@164:             b"12345",
amine@164:             msg="wrong block, expected: '12345', found: {0} ".format(block),
amine@164:         )
amine@164: 
amine@164:     def test_sr10_sw1_ch1_read_closed(self):
amine@164:         self.audio_source.close()
amine@164:         with self.assertRaises(Exception):
amine@164:             self.audio_source.read(1)
amine@164: 
amine@169: @genty
amine@164: class TestBufferAudioSource_SR16_SW2_CH1(unittest.TestCase):
amine@164:     def setUp(self):
amine@164:         self.data = b"ABCDEFGHIJKLMNOPQRSTUVWXYZ012345"
amine@164:         self.audio_source = BufferAudioSource(
amine@164:             data_buffer=self.data, sampling_rate=16, sample_width=2, channels=1
amine@164:         )
amine@164:         self.audio_source.open()
amine@164: 
amine@164:     def tearDown(self):
amine@164:         self.audio_source.close()
amine@164: 
amine@164:     def test_sr16_sw2_ch1_read_1(self):
amine@164:         block = self.audio_source.read(1)
amine@164:         exp = b"AB"
amine@164:         self.assertEqual(
amine@164:             block,
amine@164:             exp,
amine@164:             msg="wrong block, expected: {}, found: {} ".format(exp, block),
amine@164:         )
amine@164: 
amine@164:     def test_sr16_sw2_ch1_read_6(self):
amine@164:         block = self.audio_source.read(6)
amine@164:         exp = b"ABCDEFGHIJKL"
amine@164:         self.assertEqual(
amine@164:             block,
amine@164:             exp,
amine@164:             msg="wrong block, expected: {}, found: {} ".format(exp, block),
amine@164:         )
amine@164: 
amine@164:     def test_sr16_sw2_ch1_read_multiple(self):
amine@164:         block = self.audio_source.read(1)
amine@164:         exp = b"AB"
amine@164:         self.assertEqual(
amine@164:             block,
amine@164:             exp,
amine@164:             msg="wrong block, expected: {}, found: {} ".format(exp, block),
amine@164:         )
amine@164: 
amine@164:         block = self.audio_source.read(6)
amine@164:         exp = b"CDEFGHIJKLMN"
amine@164:         self.assertEqual(
amine@164:             block,
amine@164:             exp,
amine@164:             msg="wrong block, expected: {}, found: {} ".format(exp, block),
amine@164:         )
amine@164: 
amine@164:         block = self.audio_source.read(5)
amine@164:         exp = b"OPQRSTUVWX"
amine@164:         self.assertEqual(
amine@164:             block,
amine@164:             exp,
amine@164:             msg="wrong block, expected: {}, found: {} ".format(exp, block),
amine@164:         )
amine@164: 
amine@164:         block = self.audio_source.read(9999)
amine@164:         exp = b"YZ012345"
amine@164:         self.assertEqual(
amine@164:             block,
amine@164:             exp,
amine@164:             msg="wrong block, expected: {}, found: {} ".format(exp, block),
amine@164:         )
amine@164: 
amine@164:     def test_sr16_sw2_ch1_read_all(self):
amine@164:         block = self.audio_source.read(9999)
amine@164:         self.assertEqual(
amine@164:             block,
amine@164:             self.data,
amine@164:             msg="wrong block, expected: {0}, found: {1} ".format(
amine@164:                 self.data, block
amine@164:             ),
amine@164:         )
amine@164: 
amine@164:         block = self.audio_source.read(1)
amine@164:         self.assertEqual(
amine@164:             block,
amine@164:             None,
amine@164:             msg="wrong block, expected: {0}, found: {1} ".format(None, block),
amine@164:         )
amine@164: 
amine@164:     def test_sr16_sw2_ch1_get_sampling_rate(self):
amine@164:         srate = self.audio_source.get_sampling_rate()
amine@164:         self.assertEqual(
amine@164:             srate,
amine@164:             16,
amine@164:             msg="wrong sampling rate, expected: 10, found: {0} ".format(srate),
amine@164:         )
amine@164: 
amine@164:     def test_sr16_sw2_ch1_get_sample_width(self):
amine@164:         swidth = self.audio_source.get_sample_width()
amine@164:         self.assertEqual(
amine@164:             swidth,
amine@164:             2,
amine@164:             msg="wrong sample width, expected: 1, found: {0} ".format(swidth),
amine@164:         )
amine@164: 
amine@164:     def test_sr16_sw2_ch1_get_channels(self):
amine@164: 
amine@164:         channels = self.audio_source.get_channels()
amine@164:         self.assertEqual(
amine@164:             channels,
amine@164:             1,
amine@164:             msg="wrong number of channels, expected: 1, found: {0} ".format(
amine@164:                 channels
amine@164:             ),
amine@164:         )
amine@164: 
amine@169:     @genty_dataset(
amine@169:         empty=([], 0, 0, 0),
amine@169:         zero=([0], 0, 0, 0),
amine@169:         two=([2], 2, 2/16, int(2000/16)),
amine@169:         eleven=([11], 11, 11/16, int(11*1000/16)),
amine@169:         multiple=([4, 8], 12, 0.75, 750),
amine@169: 
amine@169:     )
amine@169:     def test_position(
amine@169:         self, block_sizes, expected_sample, expected_second, expected_ms
amine@169:     ):
amine@169:         for block_size in block_sizes:
amine@169:             self.audio_source.read(block_size)
amine@169:         position = self.audio_source.position
amine@169:         self.assertEqual(
amine@169:             position,
amine@169:             expected_sample,
amine@169:             msg="wrong stream position, expected: {}, found: {}".format(
amine@169:                 expected_sample, position
amine@169:             ),
amine@169:         )
amine@169: 
amine@169:         position_s = self.audio_source.position_s
amine@169:         self.assertEqual(
amine@169:             position_s,
amine@169:             expected_second,
amine@169:             msg="wrong stream position_s, expected: {}, found: {}".format(
amine@169:                 expected_second, position_s
amine@169:             ),
amine@169:         )
amine@169: 
amine@169:         position_ms = self.audio_source.position_ms
amine@169:         self.assertEqual(
amine@169:             position_ms,
amine@169:             expected_ms,
amine@169:             msg="wrong stream position_s, expected: {}, found: {}".format(
amine@169:                 expected_ms, position_ms
amine@169:             ),
amine@169:         )
amine@169: 
amine@169:     def test_sr16_sw2_ch1_read_set_position_0(self):
amine@169:         self.audio_source.read(10)
amine@169:         self.audio_source.position = 0
amine@169:         pos = self.audio_source.position
amine@164:         self.assertEqual(
amine@164:             pos, 0, msg="wrong position, expected: 0, found: {0} ".format(pos)
amine@164:         )
amine@164: 
amine@169:     @genty_dataset(
amine@169:         zero=(0, 0, 0, 0),
amine@169:         one=(1, 1, 1/16, int(1000/16)),
amine@169:         ten=(10, 10, 10/16, int(10000/16)),
amine@169:         negative_1=(-1, 15, 15/16, int(15000/16)),
amine@169:         negative_2=(-7, 9, 9/16, int(9000/16)),
amine@169:     )
amine@169:     def test_position_setter(
amine@169:         self, position, expected_sample, expected_second, expected_ms
amine@169:     ):
amine@169:         self.audio_source.position = position
amine@169: 
amine@169:         position = self.audio_source.position
amine@164:         self.assertEqual(
amine@169:             position,
amine@169:             expected_sample,
amine@169:             msg="wrong stream position, expected: {}, found: {}".format(
amine@169:                 expected_sample, position
amine@169:             ),
amine@164:         )
amine@164: 
amine@169:         position_s = self.audio_source.position_s
amine@164:         self.assertEqual(
amine@169:             position_s,
amine@169:             expected_second,
amine@169:             msg="wrong stream position_s, expected: {}, found: {}".format(
amine@169:                 expected_second, position_s
amine@169:             ),
amine@164:         )
amine@164: 
amine@169:         position_ms = self.audio_source.position_ms
amine@164:         self.assertEqual(
amine@169:             position_ms,
amine@169:             expected_ms,
amine@169:             msg="wrong stream position_s, expected: {}, found: {}".format(
amine@169:                 expected_ms, position_ms
amine@169:             ),
amine@164:         )
amine@164: 
amine@169:     @genty_dataset(
amine@169:         zero=(0, 0, 0, 0),
amine@169:         one=(0.1, 1, 1/16, int(1000/16)),
amine@169:         two=(1/8, 2, 1/8, int(1/8 * 1000)),
amine@169:         twelve=(0.75, 12, .75, 750),
amine@169:         negative_1=(-0.1, 15, 15/16, int(15000/16)),
amine@169:         negative_2=(-0.7, 5, 5/16, int(5000/16)),
amine@169:     )
amine@169:     def test_position_s_setter(
amine@169:         self, position_s, expected_sample, expected_second, expected_ms
amine@169:     ):
amine@169:         self.audio_source.position_s = position_s
amine@169: 
amine@169:         position = self.audio_source.position
amine@164:         self.assertEqual(
amine@169:             position,
amine@169:             expected_sample,
amine@169:             msg="wrong stream position, expected: {}, found: {}".format(
amine@169:                 expected_sample, position
amine@169:             ),
amine@164:         )
amine@164: 
amine@169:         position_s = self.audio_source.position_s
amine@164:         self.assertEqual(
amine@169:             position_s,
amine@169:             expected_second,
amine@169:             msg="wrong stream position_s, expected: {}, found: {}".format(
amine@169:                 expected_second, position_s
amine@169:             ),
amine@164:         )
amine@164: 
amine@169:         position_ms = self.audio_source.position_ms
amine@164:         self.assertEqual(
amine@169:             position_ms,
amine@169:             expected_ms,
amine@169:             msg="wrong stream position_s, expected: {}, found: {}".format(
amine@169:                 expected_ms, position_ms
amine@169:             ),
amine@164:         )
amine@164: 
amine@169:     @genty_dataset(
amine@169:         zero=(0, 0, 0, 0),
amine@169:         one=(100, 1, 1/16, int(1000/16)),
amine@169:         ten=(1000, 16, 1, 1000),
amine@169:         negative_1=(-100, 15, 15/16, int(15*1000/16)),
amine@169:         negative_2=(-500, 8, 0.5, 500),
amine@169:         negative_3=(-700, 5, 5/16, int(5*1000/16)),
amine@169:     )
amine@169:     def test_position_ms_setter(
amine@169:         self, position_ms, expected_sample, expected_second, expected_ms
amine@169:     ):
amine@169:         self.audio_source.position_ms = position_ms
amine@169: 
amine@169:         position = self.audio_source.position
amine@164:         self.assertEqual(
amine@169:             position,
amine@169:             expected_sample,
amine@169:             msg="wrong stream position, expected: {}, found: {}".format(
amine@169:                 expected_sample, position
amine@169:             ),
amine@164:         )
amine@164: 
amine@169:         position_s = self.audio_source.position_s
amine@164:         self.assertEqual(
amine@169:             position_s,
amine@169:             expected_second,
amine@169:             msg="wrong stream position_s, expected: {}, found: {}".format(
amine@169:                 expected_second, position_s
amine@169:             ),
amine@164:         )
amine@164: 
amine@169:         position_ms = self.audio_source.position_ms
amine@164:         self.assertEqual(
amine@169:             position_ms,
amine@169:             expected_ms,
amine@169:             msg="wrong stream position_s, expected: {}, found: {}".format(
amine@169:                 expected_ms, position_ms
amine@169:             ),
amine@164:         )
amine@164: 
amine@169: 
amine@169: 
amine@164:     def test_sr16_sw2_ch1_rewind(self):
amine@164:         self.audio_source.read(10)
amine@164:         self.audio_source.rewind()
amine@164:         tp = self.audio_source.get_position()
amine@164:         self.assertEqual(
amine@164:             tp, 0, msg="wrong position, expected: 0.0, found: {0} ".format(tp)
amine@164:         )
amine@164: 
amine@164:     def test_sr16_sw2_ch1_set_data(self):
amine@164:         self.audio_source.set_data(b"abcdef")
amine@164:         block = self.audio_source.read(9999)
amine@164:         self.assertEqual(
amine@164:             block,
amine@164:             b"abcdef",
amine@164:             msg="wrong block, expected: 'abcdef', found: {0} ".format(block),
amine@164:         )
amine@164: 
amine@164:     def test_sr16_sw2_ch1_set_data_exception(self):
amine@164:         with self.assertRaises(AudioParameterError) as audio_param_err:
amine@164:             self.audio_source.set_data("abcde")
amine@164:             self.assertEqual(
amine@164:                 "The length of audio data must be an integer "
amine@164:                 "multiple of `sample_width * channels`",
amine@164:                 str(audio_param_err.exception),
amine@164:             )
amine@164: 
amine@164:     def test_sr16_sw2_ch1_append_data_exception(self):
amine@164:         with self.assertRaises(AudioParameterError) as audio_param_err:
amine@164:             self.audio_source.append_data("abcde")
amine@164:             self.assertEqual(
amine@164:                 "The length of audio data must be an integer "
amine@164:                 "multiple of `sample_width * channels`",
amine@164:                 str(audio_param_err.exception),
amine@164:             )
amine@164: 
amine@164: 
amine@164: class TestBufferAudioSource_SR11_SW4_CH1(unittest.TestCase):
amine@164:     def setUp(self):
amine@164:         self.data = b"ABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789abcdefgh"
amine@164:         self.audio_source = BufferAudioSource(
amine@164:             data_buffer=self.data, sampling_rate=11, sample_width=4, channels=1
amine@164:         )
amine@164:         self.audio_source.open()
amine@164: 
amine@164:     def tearDown(self):
amine@164:         self.audio_source.close()
amine@164: 
amine@164:     def test_sr11_sw4_ch1_read_1(self):
amine@164:         block = self.audio_source.read(1)
amine@164:         exp = b"ABCD"
amine@164:         self.assertEqual(
amine@164:             block,
amine@164:             exp,
amine@164:             msg="wrong block, expected: {}, found: {} ".format(exp, block),
amine@164:         )
amine@164: 
amine@164:     def test_sr11_sw4_ch1_read_6(self):
amine@164:         block = self.audio_source.read(6)
amine@164:         exp = b"ABCDEFGHIJKLMNOPQRSTUVWX"
amine@164:         self.assertEqual(
amine@164:             block,
amine@164:             exp,
amine@164:             msg="wrong block, expected: {}, found: {} ".format(exp, block),
amine@164:         )
amine@164: 
amine@164:     def test_sr11_sw4_ch1_read_multiple(self):
amine@164:         block = self.audio_source.read(1)
amine@164:         exp = b"ABCD"
amine@164:         self.assertEqual(
amine@164:             block,
amine@164:             exp,
amine@164:             msg="wrong block, expected: {}, found: {} ".format(exp, block),
amine@164:         )
amine@164: 
amine@164:         block = self.audio_source.read(6)
amine@164:         exp = b"EFGHIJKLMNOPQRSTUVWXYZ01"
amine@164:         self.assertEqual(
amine@164:             block,
amine@164:             exp,
amine@164:             msg="wrong block, expected: {}, found: {} ".format(exp, block),
amine@164:         )
amine@164: 
amine@164:         block = self.audio_source.read(3)
amine@164:         exp = b"23456789abcd"
amine@164:         self.assertEqual(
amine@164:             block,
amine@164:             exp,
amine@164:             msg="wrong block, expected: {}, found: {} ".format(exp, block),
amine@164:         )
amine@164: 
amine@164:         block = self.audio_source.read(9999)
amine@164:         exp = b"efgh"
amine@164:         self.assertEqual(
amine@164:             block,
amine@164:             exp,
amine@164:             msg="wrong block, expected: {}, found: {} ".format(exp, block),
amine@164:         )
amine@164: 
amine@164:     def test_sr11_sw4_ch1_read_all(self):
amine@164:         block = self.audio_source.read(9999)
amine@164:         self.assertEqual(
amine@164:             block,
amine@164:             self.data,
amine@164:             msg="wrong block, expected: {0}, found: {1} ".format(
amine@164:                 self.data, block
amine@164:             ),
amine@164:         )
amine@164: 
amine@164:         block = self.audio_source.read(1)
amine@164:         self.assertEqual(
amine@164:             block,
amine@164:             None,
amine@164:             msg="wrong block, expected: {0}, found: {1} ".format(None, block),
amine@164:         )
amine@164: 
amine@164:     def test_sr11_sw4_ch1_get_sampling_rate(self):
amine@164:         srate = self.audio_source.get_sampling_rate()
amine@164:         self.assertEqual(
amine@164:             srate,
amine@164:             11,
amine@164:             msg="wrong sampling rate, expected: 10, found: {0} ".format(srate),
amine@164:         )
amine@164: 
amine@164:     def test_sr11_sw4_ch1_get_sample_width(self):
amine@164:         swidth = self.audio_source.get_sample_width()
amine@164:         self.assertEqual(
amine@164:             swidth,
amine@164:             4,
amine@164:             msg="wrong sample width, expected: 1, found: {0} ".format(swidth),
amine@164:         )
amine@164: 
amine@164:     def test_sr11_sw4_ch1_get_channels(self):
amine@164:         channels = self.audio_source.get_channels()
amine@164:         self.assertEqual(
amine@164:             channels,
amine@164:             1,
amine@164:             msg="wrong number of channels, expected: 1, found: {0} ".format(
amine@164:                 channels
amine@164:             ),
amine@164:         )
amine@164: 
amine@164:     def test_sr11_sw4_ch1_get_position_0(self):
amine@164:         pos = self.audio_source.get_position()
amine@164:         self.assertEqual(
amine@164:             pos, 0, msg="wrong position, expected: 0, found: {0} ".format(pos)
amine@164:         )
amine@164: 
amine@164:     def test_sr11_sw4_ch1_get_position_5(self):
amine@164:         self.audio_source.read(5)
amine@164:         pos = self.audio_source.get_position()
amine@164:         self.assertEqual(
amine@164:             pos, 5, msg="wrong position, expected: 5, found: {0} ".format(pos)
amine@164:         )
amine@164: 
amine@164:     def test_sr11_sw4_ch1_get_position_9(self):
amine@164:         self.audio_source.read(5)
amine@164:         self.audio_source.read(4)
amine@164:         pos = self.audio_source.get_position()
amine@164:         self.assertEqual(
amine@164:             pos, 9, msg="wrong position, expected: 5, found: {0} ".format(pos)
amine@164:         )
amine@164: 
amine@164:     def test_sr11_sw4_ch1_set_position_0(self):
amine@164:         self.audio_source.read(10)
amine@164:         self.audio_source.set_position(0)
amine@164:         pos = self.audio_source.get_position()
amine@164:         self.assertEqual(
amine@164:             pos, 0, msg="wrong position, expected: 0, found: {0} ".format(pos)
amine@164:         )
amine@164: 
amine@164:     def test_sr11_sw4_ch1_set_position_10(self):
amine@164:         self.audio_source.set_position(10)
amine@164:         pos = self.audio_source.get_position()
amine@164:         self.assertEqual(
amine@164:             pos,
amine@164:             10,
amine@164:             msg="wrong position, expected: 10, found: {0} ".format(pos),
amine@164:         )
amine@164: 
amine@164:     def test_sr11_sw4_ch1_get_time_position_0(self):
amine@164:         tp = self.audio_source.get_time_position()
amine@164:         self.assertEqual(
amine@164:             tp,
amine@164:             0.0,
amine@164:             msg="wrong time position, expected: 0.0, found: {0} ".format(tp),
amine@164:         )
amine@164: 
amine@164:     def test_sr11_sw4_ch1_get_time_position_1(self):
amine@164:         srate = self.audio_source.get_sampling_rate()
amine@164:         # read one second
amine@164:         self.audio_source.read(srate)
amine@164:         tp = self.audio_source.get_time_position()
amine@164:         self.assertEqual(
amine@164:             tp,
amine@164:             1.0,
amine@164:             msg="wrong time position, expected: 1.0, found: {0} ".format(tp),
amine@164:         )
amine@164: 
amine@164:     def test_sr11_sw4_ch1_get_time_position_0_63(self):
amine@164:         # read 2.5 seconds
amine@164:         self.audio_source.read(7)
amine@164:         tp = self.audio_source.get_time_position()
amine@164:         self.assertAlmostEqual(
amine@164:             tp,
amine@164:             0.636363636364,
amine@164:             msg="wrong time position, expected: 0.636363636364, "
amine@164:             "found: {0} ".format(tp),
amine@164:         )
amine@164: 
amine@164:     def test_sr11_sw4_ch1_set_time_position_0(self):
amine@164:         self.audio_source.read(10)
amine@164:         self.audio_source.set_time_position(0)
amine@164:         tp = self.audio_source.get_time_position()
amine@164:         self.assertEqual(
amine@164:             tp,
amine@164:             0.0,
amine@164:             msg="wrong time position, expected: 0.0, found: {0} ".format(tp),
amine@164:         )
amine@164: 
amine@164:     def test_sr11_sw4_ch1_set_time_position_1(self):
amine@164: 
amine@164:         self.audio_source.set_time_position(1)
amine@164:         tp = self.audio_source.get_time_position()
amine@164:         self.assertEqual(
amine@164:             tp,
amine@164:             1.0,
amine@164:             msg="wrong time position, expected: 1.0, found: {0} ".format(tp),
amine@164:         )
amine@164: 
amine@164:     def test_sr11_sw4_ch1_rewind(self):
amine@164:         self.audio_source.read(10)
amine@164:         self.audio_source.rewind()
amine@169:         tp = self.audio_source.position
amine@164:         self.assertEqual(
amine@164:             tp, 0, msg="wrong position, expected: 0.0, found: {0} ".format(tp)
amine@164:         )
amine@164: 
amine@164:     def test_sr11_sw4_ch1_set_data(self):
amine@164:         self.audio_source.set_data(b"abcdefgh")
amine@164:         block = self.audio_source.read(9999)
amine@164:         exp = b"abcdefgh"
amine@164:         self.assertEqual(
amine@164:             block,
amine@164:             exp,
amine@164:             msg="wrong block, expected: {}, found: {} ".format(exp, block),
amine@164:         )
amine@164: 
amine@164:     def test_sr11_sw4_ch1_set_data_exception(self):
amine@164:         with self.assertRaises(AudioParameterError) as audio_param_err:
amine@164:             self.audio_source.set_data(b"abcdef")
amine@164:         self.assertEqual(
amine@164:             "The length of audio data must be an integer "
amine@164:             "multiple of `sample_width * channels`",
amine@164:             str(audio_param_err.exception),
amine@164:         )
amine@164: 
amine@164:     def test_sr11_sw4_ch1_append_data_exception(self):
amine@164:         with self.assertRaises(AudioParameterError) as audio_param_err:
amine@164:             self.audio_source.append_data(b"abcdef")
amine@164:         self.assertEqual(
amine@164:             "The length of audio data must be an integer "
amine@164:             "multiple of `sample_width * channels`",
amine@164:             str(audio_param_err.exception),
amine@164:         )
amine@164: 
amine@164: 
amine@164: class TestBufferAudioSourceCreationException(unittest.TestCase):
amine@164:     def test_wrong_sample_width_value(self):
amine@164:         with self.assertRaises(AudioParameterError) as audio_param_err:
amine@164:             _ = BufferAudioSource(
amine@164:                 data_buffer=b"ABCDEFGHI",
amine@164:                 sampling_rate=9,
amine@164:                 sample_width=3,
amine@164:                 channels=1,
amine@164:             )
amine@164:         self.assertEqual(
amine@164:             "Sample width must be one of: 1, 2 or 4 (bytes)",
amine@164:             str(audio_param_err.exception),
amine@164:         )
amine@164: 
amine@164:     def test_wrong_data_buffer_size(self):
amine@164:         with self.assertRaises(AudioParameterError) as audio_param_err:
amine@164:             _ = BufferAudioSource(
amine@164:                 data_buffer=b"ABCDEFGHI",
amine@164:                 sampling_rate=8,
amine@164:                 sample_width=2,
amine@164:                 channels=1,
amine@164:             )
amine@164:         self.assertEqual(
amine@164:             "The length of audio data must be an integer "
amine@164:             "multiple of `sample_width * channels`",
amine@164:             str(audio_param_err.exception),
amine@164:         )
amine@164: 
amine@164: 
amine@164: class TestAudioSourceProperties(unittest.TestCase):
amine@164:     def test_read_properties(self):
amine@164:         data = b""
amine@164:         sampling_rate = 8000
amine@164:         sample_width = 2
amine@164:         channels = 1
amine@164:         a_source = BufferAudioSource(
amine@164:             data, sampling_rate, sample_width, channels
amine@164:         )
amine@164: 
amine@164:         self.assertEqual(a_source.sampling_rate, sampling_rate)
amine@164:         self.assertEqual(a_source.sample_width, sample_width)
amine@164:         self.assertEqual(a_source.channels, channels)
amine@164: 
amine@164:     def test_set_readonly_properties_exception(self):
amine@164:         data = b""
amine@164:         sampling_rate = 8000
amine@164:         sample_width = 2
amine@164:         channels = 1
amine@164:         a_source = BufferAudioSource(
amine@164:             data, sampling_rate, sample_width, channels
amine@164:         )
amine@164: 
amine@164:         with self.assertRaises(AttributeError):
amine@164:             a_source.sampling_rate = 16000
amine@164:             a_source.sample_width = 1
amine@164:             a_source.channels = 2
amine@164: 
amine@164: 
amine@164: class TestAudioSourceShortProperties(unittest.TestCase):
amine@164:     def test_read_short_properties(self):
amine@164:         data = b""
amine@164:         sampling_rate = 8000
amine@164:         sample_width = 2
amine@164:         channels = 1
amine@164:         a_source = BufferAudioSource(
amine@164:             data, sampling_rate, sample_width, channels
amine@164:         )
amine@164: 
amine@164:         self.assertEqual(a_source.sr, sampling_rate)
amine@164:         self.assertEqual(a_source.sw, sample_width)
amine@164:         self.assertEqual(a_source.ch, channels)
amine@164: 
amine@164:     def test_set_readonly_short_properties_exception(self):
amine@164:         data = b""
amine@164:         sampling_rate = 8000
amine@164:         sample_width = 2
amine@164:         channels = 1
amine@164:         a_source = BufferAudioSource(
amine@164:             data, sampling_rate, sample_width, channels
amine@164:         )
amine@164: 
amine@164:         with self.assertRaises(AttributeError):
amine@164:             a_source.sr = 16000
amine@164:             a_source.sw = 1
amine@164:             a_source.ch = 2
amine@164: 
amine@164: 
amine@164: if __name__ == "__main__":
amine@208:     unittest.main()