amine@173
|
1 """
|
amine@172
|
2 @author: Amine Sehili <amine.sehili@gmail.com>
|
amine@172
|
3 September 2015
|
amine@172
|
4
|
amine@173
|
5 """
|
amine@172
|
6
|
amine@172
|
7 import unittest
|
amine@172
|
8 from functools import partial
|
amine@172
|
9 import sys
|
amine@173
|
10 from auditok import (
|
amine@173
|
11 dataset,
|
amine@173
|
12 ADSFactory,
|
amine@173
|
13 AudioDataSource,
|
amine@173
|
14 BufferAudioSource,
|
amine@173
|
15 WaveAudioSource,
|
amine@173
|
16 DuplicateArgument,
|
amine@173
|
17 )
|
amine@172
|
18 import wave
|
amine@172
|
19
|
amine@172
|
20
|
amine@172
|
21 try:
|
amine@172
|
22 from builtins import range
|
amine@172
|
23 except ImportError:
|
amine@172
|
24 if sys.version_info < (3, 0):
|
amine@172
|
25 range = xrange
|
amine@172
|
26
|
amine@173
|
27
|
amine@172
|
28 class TestADSFactoryFileAudioSource(unittest.TestCase):
|
amine@172
|
29 def setUp(self):
|
amine@173
|
30 self.audio_source = WaveAudioSource(
|
amine@173
|
31 filename=dataset.one_to_six_arabic_16000_mono_bc_noise
|
amine@173
|
32 )
|
amine@173
|
33
|
amine@172
|
34 def test_ADS_type(self):
|
amine@173
|
35
|
amine@172
|
36 ads = ADSFactory.ads(audio_source=self.audio_source)
|
amine@173
|
37
|
amine@173
|
38 self.assertIsInstance(
|
amine@173
|
39 ads,
|
amine@173
|
40 AudioDataSource,
|
amine@173
|
41 msg="wrong type for ads object, expected: 'AudioDataSource', found: {0}".format(
|
amine@173
|
42 type(ads)
|
amine@173
|
43 ),
|
amine@173
|
44 )
|
amine@173
|
45
|
amine@172
|
46 def test_default_block_size(self):
|
amine@172
|
47 ads = ADSFactory.ads(audio_source=self.audio_source)
|
amine@173
|
48 size = ads.block_size
|
amine@173
|
49 self.assertEqual(
|
amine@173
|
50 size,
|
amine@173
|
51 160,
|
amine@173
|
52 "Wrong default block_size, expected: 160, found: {0}".format(size),
|
amine@173
|
53 )
|
amine@173
|
54
|
amine@172
|
55 def test_block_size(self):
|
amine@172
|
56 ads = ADSFactory.ads(audio_source=self.audio_source, block_size=512)
|
amine@173
|
57 size = ads.block_size
|
amine@173
|
58 self.assertEqual(
|
amine@173
|
59 size,
|
amine@173
|
60 512,
|
amine@173
|
61 "Wrong block_size, expected: 512, found: {0}".format(size),
|
amine@173
|
62 )
|
amine@173
|
63
|
amine@172
|
64 # with alias keyword
|
amine@172
|
65 ads = ADSFactory.ads(audio_source=self.audio_source, bs=160)
|
amine@173
|
66 size = ads.block_size
|
amine@173
|
67 self.assertEqual(
|
amine@173
|
68 size,
|
amine@173
|
69 160,
|
amine@173
|
70 "Wrong block_size, expected: 160, found: {0}".format(size),
|
amine@173
|
71 )
|
amine@173
|
72
|
amine@172
|
73 def test_block_duration(self):
|
amine@173
|
74
|
amine@173
|
75 ads = ADSFactory.ads(
|
amine@173
|
76 audio_source=self.audio_source, block_dur=0.01
|
amine@173
|
77 ) # 10 ms
|
amine@173
|
78 size = ads.block_size
|
amine@173
|
79 self.assertEqual(
|
amine@173
|
80 size,
|
amine@173
|
81 160,
|
amine@173
|
82 "Wrong block_size, expected: 160, found: {0}".format(size),
|
amine@173
|
83 )
|
amine@173
|
84
|
amine@172
|
85 # with alias keyword
|
amine@173
|
86 ads = ADSFactory.ads(audio_source=self.audio_source, bd=0.025) # 25 ms
|
amine@173
|
87 size = ads.block_size
|
amine@173
|
88 self.assertEqual(
|
amine@173
|
89 size,
|
amine@173
|
90 400,
|
amine@173
|
91 "Wrong block_size, expected: 400, found: {0}".format(size),
|
amine@173
|
92 )
|
amine@173
|
93
|
amine@172
|
94 def test_hop_duration(self):
|
amine@173
|
95
|
amine@173
|
96 ads = ADSFactory.ads(
|
amine@173
|
97 audio_source=self.audio_source, block_dur=0.02, hop_dur=0.01
|
amine@173
|
98 ) # 10 ms
|
amine@172
|
99 size = ads.hop_size
|
amine@173
|
100 self.assertEqual(
|
amine@173
|
101 size, 160, "Wrong hop_size, expected: 160, found: {0}".format(size)
|
amine@173
|
102 )
|
amine@173
|
103
|
amine@172
|
104 # with alias keyword
|
amine@173
|
105 ads = ADSFactory.ads(
|
amine@173
|
106 audio_source=self.audio_source, bd=0.025, hop_dur=0.015
|
amine@173
|
107 ) # 15 ms
|
amine@172
|
108 size = ads.hop_size
|
amine@173
|
109 self.assertEqual(
|
amine@173
|
110 size,
|
amine@173
|
111 240,
|
amine@173
|
112 "Wrong block_size, expected: 240, found: {0}".format(size),
|
amine@173
|
113 )
|
amine@173
|
114
|
amine@172
|
115 def test_sampling_rate(self):
|
amine@172
|
116 ads = ADSFactory.ads(audio_source=self.audio_source)
|
amine@173
|
117
|
amine@173
|
118 srate = ads.sampling_rate
|
amine@173
|
119 self.assertEqual(
|
amine@173
|
120 srate,
|
amine@173
|
121 16000,
|
amine@173
|
122 "Wrong sampling rate, expected: 16000, found: {0}".format(srate),
|
amine@173
|
123 )
|
amine@173
|
124
|
amine@172
|
125 def test_sample_width(self):
|
amine@172
|
126 ads = ADSFactory.ads(audio_source=self.audio_source)
|
amine@173
|
127
|
amine@173
|
128 swidth = ads.sample_width
|
amine@173
|
129 self.assertEqual(
|
amine@173
|
130 swidth,
|
amine@173
|
131 2,
|
amine@173
|
132 "Wrong sample width, expected: 2, found: {0}".format(swidth),
|
amine@173
|
133 )
|
amine@173
|
134
|
amine@172
|
135 def test_channels(self):
|
amine@172
|
136 ads = ADSFactory.ads(audio_source=self.audio_source)
|
amine@173
|
137
|
amine@173
|
138 channels = ads.channels
|
amine@173
|
139 self.assertEqual(
|
amine@173
|
140 channels,
|
amine@173
|
141 1,
|
amine@173
|
142 "Wrong number of channels, expected: 1, found: {0}".format(
|
amine@173
|
143 channels
|
amine@173
|
144 ),
|
amine@173
|
145 )
|
amine@173
|
146
|
amine@172
|
147 def test_read(self):
|
amine@173
|
148 ads = ADSFactory.ads(audio_source=self.audio_source, block_size=256)
|
amine@173
|
149
|
amine@172
|
150 ads.open()
|
amine@172
|
151 ads_data = ads.read()
|
amine@172
|
152 ads.close()
|
amine@173
|
153
|
amine@173
|
154 audio_source = WaveAudioSource(
|
amine@173
|
155 filename=dataset.one_to_six_arabic_16000_mono_bc_noise
|
amine@173
|
156 )
|
amine@172
|
157 audio_source.open()
|
amine@172
|
158 audio_source_data = audio_source.read(256)
|
amine@172
|
159 audio_source.close()
|
amine@173
|
160
|
amine@173
|
161 self.assertEqual(
|
amine@173
|
162 ads_data, audio_source_data, "Unexpected data read from ads"
|
amine@173
|
163 )
|
amine@173
|
164
|
amine@172
|
165 def test_Limiter_Deco_read(self):
|
amine@172
|
166 # read a maximum of 0.75 seconds from audio source
|
amine@172
|
167 ads = ADSFactory.ads(audio_source=self.audio_source, max_time=0.75)
|
amine@173
|
168
|
amine@172
|
169 ads_data = []
|
amine@172
|
170 ads.open()
|
amine@172
|
171 while True:
|
amine@172
|
172 block = ads.read()
|
amine@172
|
173 if block is None:
|
amine@172
|
174 break
|
amine@172
|
175 ads_data.append(block)
|
amine@172
|
176 ads.close()
|
amine@173
|
177 ads_data = b"".join(ads_data)
|
amine@173
|
178
|
amine@173
|
179 audio_source = WaveAudioSource(
|
amine@173
|
180 filename=dataset.one_to_six_arabic_16000_mono_bc_noise
|
amine@173
|
181 )
|
amine@172
|
182 audio_source.open()
|
amine@172
|
183 audio_source_data = audio_source.read(int(16000 * 0.75))
|
amine@172
|
184 audio_source.close()
|
amine@173
|
185
|
amine@173
|
186 self.assertEqual(
|
amine@173
|
187 ads_data, audio_source_data, "Unexpected data read from LimiterADS"
|
amine@173
|
188 )
|
amine@173
|
189
|
amine@172
|
190 def test_Limiter_Deco_read_limit(self):
|
amine@173
|
191 # read a maximum of 1.191 seconds from audio source
|
amine@172
|
192 ads = ADSFactory.ads(audio_source=self.audio_source, max_time=1.191)
|
amine@173
|
193 total_samples = round(ads.sampling_rate * 1.191)
|
amine@173
|
194 nb_full_blocks, last_block_size = divmod(total_samples, ads.block_size)
|
amine@173
|
195 total_samples_with_overlap = (
|
amine@173
|
196 nb_full_blocks * ads.block_size + last_block_size
|
amine@173
|
197 )
|
amine@173
|
198 expected_read_bytes = (
|
amine@173
|
199 total_samples_with_overlap * ads.sw * ads.channels
|
amine@173
|
200 )
|
amine@173
|
201
|
amine@172
|
202 total_read = 0
|
amine@172
|
203 ads.open()
|
amine@172
|
204 i = 0
|
amine@172
|
205 while True:
|
amine@172
|
206 block = ads.read()
|
amine@172
|
207 if block is None:
|
amine@172
|
208 break
|
amine@172
|
209 i += 1
|
amine@172
|
210 total_read += len(block)
|
amine@173
|
211
|
amine@172
|
212 ads.close()
|
amine@173
|
213
|
amine@173
|
214 self.assertEqual(
|
amine@173
|
215 total_read,
|
amine@173
|
216 expected_read_bytes,
|
amine@173
|
217 "Wrong data length read from LimiterADS, expected: {0}, found: {1}".format(
|
amine@173
|
218 expected_read_bytes, total_read
|
amine@173
|
219 ),
|
amine@173
|
220 )
|
amine@173
|
221
|
amine@172
|
222 def test_Recorder_Deco_read(self):
|
amine@173
|
223 ads = ADSFactory.ads(
|
amine@173
|
224 audio_source=self.audio_source, record=True, block_size=500
|
amine@173
|
225 )
|
amine@173
|
226
|
amine@172
|
227 ads_data = []
|
amine@172
|
228 ads.open()
|
amine@172
|
229 for i in range(10):
|
amine@172
|
230 block = ads.read()
|
amine@172
|
231 if block is None:
|
amine@172
|
232 break
|
amine@172
|
233 ads_data.append(block)
|
amine@172
|
234 ads.close()
|
amine@173
|
235 ads_data = b"".join(ads_data)
|
amine@173
|
236
|
amine@173
|
237 audio_source = WaveAudioSource(
|
amine@173
|
238 filename=dataset.one_to_six_arabic_16000_mono_bc_noise
|
amine@173
|
239 )
|
amine@172
|
240 audio_source.open()
|
amine@172
|
241 audio_source_data = audio_source.read(500 * 10)
|
amine@172
|
242 audio_source.close()
|
amine@173
|
243
|
amine@173
|
244 self.assertEqual(
|
amine@173
|
245 ads_data,
|
amine@173
|
246 audio_source_data,
|
amine@173
|
247 "Unexpected data read from RecorderADS",
|
amine@173
|
248 )
|
amine@173
|
249
|
amine@172
|
250 def test_Recorder_Deco_is_rewindable(self):
|
amine@172
|
251 ads = ADSFactory.ads(audio_source=self.audio_source, record=True)
|
amine@173
|
252
|
amine@173
|
253 self.assertTrue(
|
amine@173
|
254 ads.rewindable, "RecorderADS.is_rewindable should return True"
|
amine@173
|
255 )
|
amine@173
|
256
|
amine@172
|
257 def test_Recorder_Deco_rewind_and_read(self):
|
amine@173
|
258 ads = ADSFactory.ads(
|
amine@173
|
259 audio_source=self.audio_source, record=True, block_size=320
|
amine@173
|
260 )
|
amine@173
|
261
|
amine@172
|
262 ads.open()
|
amine@172
|
263 for i in range(10):
|
amine@172
|
264 ads.read()
|
amine@173
|
265
|
amine@172
|
266 ads.rewind()
|
amine@173
|
267
|
amine@172
|
268 # read all available data after rewind
|
amine@172
|
269 ads_data = []
|
amine@172
|
270 while True:
|
amine@172
|
271 block = ads.read()
|
amine@172
|
272 if block is None:
|
amine@172
|
273 break
|
amine@172
|
274 ads_data.append(block)
|
amine@172
|
275 ads.close()
|
amine@173
|
276 ads_data = b"".join(ads_data)
|
amine@173
|
277
|
amine@173
|
278 audio_source = WaveAudioSource(
|
amine@173
|
279 filename=dataset.one_to_six_arabic_16000_mono_bc_noise
|
amine@173
|
280 )
|
amine@172
|
281 audio_source.open()
|
amine@172
|
282 audio_source_data = audio_source.read(320 * 10)
|
amine@172
|
283 audio_source.close()
|
amine@173
|
284
|
amine@173
|
285 self.assertEqual(
|
amine@173
|
286 ads_data,
|
amine@173
|
287 audio_source_data,
|
amine@173
|
288 "Unexpected data read from RecorderADS",
|
amine@173
|
289 )
|
amine@173
|
290
|
amine@172
|
291 def test_Overlap_Deco_read(self):
|
amine@173
|
292
|
amine@172
|
293 # Use arbitrary valid block_size and hop_size
|
amine@172
|
294 block_size = 1714
|
amine@172
|
295 hop_size = 313
|
amine@173
|
296
|
amine@173
|
297 ads = ADSFactory.ads(
|
amine@173
|
298 audio_source=self.audio_source,
|
amine@173
|
299 block_size=block_size,
|
amine@173
|
300 hop_size=hop_size,
|
amine@173
|
301 )
|
amine@173
|
302
|
amine@172
|
303 # Read all available data overlapping blocks
|
amine@172
|
304 ads.open()
|
amine@172
|
305 ads_data = []
|
amine@172
|
306 while True:
|
amine@172
|
307 block = ads.read()
|
amine@172
|
308 if block is None:
|
amine@172
|
309 break
|
amine@172
|
310 ads_data.append(block)
|
amine@172
|
311 ads.close()
|
amine@173
|
312
|
amine@172
|
313 # Read all data from file and build a BufferAudioSource
|
amine@172
|
314 fp = wave.open(dataset.one_to_six_arabic_16000_mono_bc_noise, "r")
|
amine@172
|
315 wave_data = fp.readframes(fp.getnframes())
|
amine@172
|
316 fp.close()
|
amine@173
|
317 audio_source = BufferAudioSource(
|
amine@173
|
318 wave_data, ads.sampling_rate, ads.sample_width, ads.channels
|
amine@173
|
319 )
|
amine@172
|
320 audio_source.open()
|
amine@173
|
321
|
amine@172
|
322 # Compare all blocks read from OverlapADS to those read
|
amine@172
|
323 # from an audio source with a manual set_position
|
amine@173
|
324 for i, block in enumerate(ads_data):
|
amine@173
|
325
|
amine@172
|
326 tmp = audio_source.read(block_size)
|
amine@173
|
327
|
amine@173
|
328 self.assertEqual(
|
amine@173
|
329 block,
|
amine@173
|
330 tmp,
|
amine@173
|
331 "Unexpected block (N={0}) read from OverlapADS".format(i),
|
amine@173
|
332 )
|
amine@173
|
333
|
amine@173
|
334 audio_source.set_position((i + 1) * hop_size)
|
amine@173
|
335
|
amine@172
|
336 audio_source.close()
|
amine@173
|
337
|
amine@173
|
338 def test_Limiter_Overlap_Deco_read(self):
|
amine@173
|
339
|
amine@172
|
340 block_size = 256
|
amine@172
|
341 hop_size = 200
|
amine@173
|
342
|
amine@173
|
343 ads = ADSFactory.ads(
|
amine@173
|
344 audio_source=self.audio_source,
|
amine@173
|
345 max_time=0.50,
|
amine@173
|
346 block_size=block_size,
|
amine@173
|
347 hop_size=hop_size,
|
amine@173
|
348 )
|
amine@173
|
349
|
amine@172
|
350 # Read all available data overlapping blocks
|
amine@172
|
351 ads.open()
|
amine@172
|
352 ads_data = []
|
amine@172
|
353 while True:
|
amine@172
|
354 block = ads.read()
|
amine@172
|
355 if block is None:
|
amine@172
|
356 break
|
amine@172
|
357 ads_data.append(block)
|
amine@172
|
358 ads.close()
|
amine@173
|
359
|
amine@172
|
360 # Read all data from file and build a BufferAudioSource
|
amine@172
|
361 fp = wave.open(dataset.one_to_six_arabic_16000_mono_bc_noise, "r")
|
amine@172
|
362 wave_data = fp.readframes(fp.getnframes())
|
amine@172
|
363 fp.close()
|
amine@173
|
364 audio_source = BufferAudioSource(
|
amine@173
|
365 wave_data, ads.sampling_rate, ads.sample_width, ads.channels
|
amine@173
|
366 )
|
amine@172
|
367 audio_source.open()
|
amine@173
|
368
|
amine@172
|
369 # Compare all blocks read from OverlapADS to those read
|
amine@172
|
370 # from an audio source with a manual set_position
|
amine@173
|
371 for i, block in enumerate(ads_data):
|
amine@173
|
372 tmp = audio_source.read(len(block) // (ads.sw * ads.ch))
|
amine@173
|
373 self.assertEqual(
|
amine@173
|
374 len(block),
|
amine@173
|
375 len(tmp),
|
amine@173
|
376 "Unexpected block (N={0}) read from OverlapADS".format(i),
|
amine@173
|
377 )
|
amine@173
|
378 audio_source.set_position((i + 1) * hop_size)
|
amine@173
|
379
|
amine@172
|
380 audio_source.close()
|
amine@173
|
381
|
amine@172
|
382 def test_Limiter_Overlap_Deco_read_limit(self):
|
amine@173
|
383
|
amine@172
|
384 block_size = 313
|
amine@172
|
385 hop_size = 207
|
amine@173
|
386 ads = ADSFactory.ads(
|
amine@173
|
387 audio_source=self.audio_source,
|
amine@173
|
388 max_time=1.932,
|
amine@173
|
389 block_size=block_size,
|
amine@173
|
390 hop_size=hop_size,
|
amine@173
|
391 )
|
amine@173
|
392
|
amine@173
|
393 total_samples = round(ads.sampling_rate * 1.932)
|
amine@173
|
394 first_read_size = block_size
|
amine@173
|
395 next_read_size = block_size - hop_size
|
amine@173
|
396 nb_next_blocks, last_block_size = divmod(
|
amine@173
|
397 (total_samples - first_read_size), next_read_size
|
amine@173
|
398 )
|
amine@173
|
399 total_samples_with_overlap = (
|
amine@173
|
400 first_read_size + next_read_size * nb_next_blocks + last_block_size
|
amine@173
|
401 )
|
amine@173
|
402 expected_read_bytes = (
|
amine@173
|
403 total_samples_with_overlap * ads.sw * ads.channels
|
amine@173
|
404 )
|
amine@173
|
405
|
amine@173
|
406 cache_size = (block_size - hop_size) * ads.sample_width * ads.channels
|
amine@172
|
407 total_read = cache_size
|
amine@173
|
408
|
amine@172
|
409 ads.open()
|
amine@172
|
410 i = 0
|
amine@172
|
411 while True:
|
amine@172
|
412 block = ads.read()
|
amine@172
|
413 if block is None:
|
amine@172
|
414 break
|
amine@172
|
415 i += 1
|
amine@172
|
416 total_read += len(block) - cache_size
|
amine@173
|
417
|
amine@172
|
418 ads.close()
|
amine@173
|
419 self.assertEqual(
|
amine@173
|
420 total_read,
|
amine@173
|
421 expected_read_bytes,
|
amine@173
|
422 "Wrong data length read from LimiterADS, expected: {0}, found: {1}".format(
|
amine@173
|
423 expected_read_bytes, total_read
|
amine@173
|
424 ),
|
amine@173
|
425 )
|
amine@173
|
426
|
amine@172
|
427 def test_Recorder_Overlap_Deco_is_rewindable(self):
|
amine@173
|
428 ads = ADSFactory.ads(
|
amine@173
|
429 audio_source=self.audio_source,
|
amine@173
|
430 block_size=320,
|
amine@173
|
431 hop_size=160,
|
amine@173
|
432 record=True,
|
amine@173
|
433 )
|
amine@173
|
434 self.assertTrue(
|
amine@173
|
435 ads.rewindable, "RecorderADS.is_rewindable should return True"
|
amine@173
|
436 )
|
amine@172
|
437
|
amine@172
|
438 def test_Recorder_Overlap_Deco_rewind_and_read(self):
|
amine@173
|
439
|
amine@172
|
440 # Use arbitrary valid block_size and hop_size
|
amine@172
|
441 block_size = 1600
|
amine@172
|
442 hop_size = 400
|
amine@173
|
443
|
amine@173
|
444 ads = ADSFactory.ads(
|
amine@173
|
445 audio_source=self.audio_source,
|
amine@173
|
446 block_size=block_size,
|
amine@173
|
447 hop_size=hop_size,
|
amine@173
|
448 record=True,
|
amine@173
|
449 )
|
amine@173
|
450
|
amine@172
|
451 # Read all available data overlapping blocks
|
amine@172
|
452 ads.open()
|
amine@172
|
453 i = 0
|
amine@172
|
454 while True:
|
amine@172
|
455 block = ads.read()
|
amine@172
|
456 if block is None:
|
amine@172
|
457 break
|
amine@172
|
458 i += 1
|
amine@173
|
459
|
amine@172
|
460 ads.rewind()
|
amine@173
|
461
|
amine@172
|
462 # Read all data from file and build a BufferAudioSource
|
amine@172
|
463 fp = wave.open(dataset.one_to_six_arabic_16000_mono_bc_noise, "r")
|
amine@172
|
464 wave_data = fp.readframes(fp.getnframes())
|
amine@172
|
465 fp.close()
|
amine@173
|
466 audio_source = BufferAudioSource(
|
amine@173
|
467 wave_data, ads.sampling_rate, ads.sample_width, ads.channels
|
amine@173
|
468 )
|
amine@172
|
469 audio_source.open()
|
amine@173
|
470
|
amine@172
|
471 # Compare all blocks read from OverlapADS to those read
|
amine@172
|
472 # from an audio source with a manual set_position
|
amine@172
|
473 for j in range(i):
|
amine@173
|
474
|
amine@172
|
475 tmp = audio_source.read(block_size)
|
amine@173
|
476
|
amine@173
|
477 self.assertEqual(
|
amine@173
|
478 ads.read(),
|
amine@173
|
479 tmp,
|
amine@173
|
480 "Unexpected block (N={0}) read from OverlapADS".format(i),
|
amine@173
|
481 )
|
amine@173
|
482 audio_source.set_position((j + 1) * hop_size)
|
amine@173
|
483
|
amine@172
|
484 ads.close()
|
amine@172
|
485 audio_source.close()
|
amine@173
|
486
|
amine@172
|
487 def test_Limiter_Recorder_Overlap_Deco_rewind_and_read(self):
|
amine@173
|
488
|
amine@172
|
489 # Use arbitrary valid block_size and hop_size
|
amine@172
|
490 block_size = 1600
|
amine@172
|
491 hop_size = 400
|
amine@173
|
492
|
amine@173
|
493 ads = ADSFactory.ads(
|
amine@173
|
494 audio_source=self.audio_source,
|
amine@173
|
495 max_time=1.50,
|
amine@173
|
496 block_size=block_size,
|
amine@173
|
497 hop_size=hop_size,
|
amine@173
|
498 record=True,
|
amine@173
|
499 )
|
amine@173
|
500
|
amine@172
|
501 # Read all available data overlapping blocks
|
amine@172
|
502 ads.open()
|
amine@172
|
503 i = 0
|
amine@172
|
504 while True:
|
amine@172
|
505 block = ads.read()
|
amine@172
|
506 if block is None:
|
amine@172
|
507 break
|
amine@172
|
508 i += 1
|
amine@173
|
509
|
amine@172
|
510 ads.rewind()
|
amine@173
|
511
|
amine@172
|
512 # Read all data from file and build a BufferAudioSource
|
amine@172
|
513 fp = wave.open(dataset.one_to_six_arabic_16000_mono_bc_noise, "r")
|
amine@172
|
514 wave_data = fp.readframes(fp.getnframes())
|
amine@172
|
515 fp.close()
|
amine@173
|
516 audio_source = BufferAudioSource(
|
amine@173
|
517 wave_data, ads.sampling_rate, ads.sample_width, ads.channels
|
amine@173
|
518 )
|
amine@172
|
519 audio_source.open()
|
amine@173
|
520
|
amine@172
|
521 # Compare all blocks read from OverlapADS to those read
|
amine@172
|
522 # from an audio source with a manual set_position
|
amine@172
|
523 for j in range(i):
|
amine@173
|
524
|
amine@172
|
525 tmp = audio_source.read(block_size)
|
amine@173
|
526
|
amine@173
|
527 self.assertEqual(
|
amine@173
|
528 ads.read(),
|
amine@173
|
529 tmp,
|
amine@173
|
530 "Unexpected block (N={0}) read from OverlapADS".format(i),
|
amine@173
|
531 )
|
amine@173
|
532 audio_source.set_position((j + 1) * hop_size)
|
amine@173
|
533
|
amine@172
|
534 ads.close()
|
amine@172
|
535 audio_source.close()
|
amine@173
|
536
|
amine@172
|
537 def test_Limiter_Recorder_Overlap_Deco_rewind_and_read_limit(self):
|
amine@173
|
538
|
amine@172
|
539 # Use arbitrary valid block_size and hop_size
|
amine@172
|
540 block_size = 1000
|
amine@172
|
541 hop_size = 200
|
amine@173
|
542
|
amine@173
|
543 ads = ADSFactory.ads(
|
amine@173
|
544 audio_source=self.audio_source,
|
amine@173
|
545 max_time=1.317,
|
amine@173
|
546 block_size=block_size,
|
amine@173
|
547 hop_size=hop_size,
|
amine@173
|
548 record=True,
|
amine@173
|
549 )
|
amine@173
|
550 total_samples = round(ads.sampling_rate * 1.317)
|
amine@173
|
551 first_read_size = block_size
|
amine@173
|
552 next_read_size = block_size - hop_size
|
amine@173
|
553 nb_next_blocks, last_block_size = divmod(
|
amine@173
|
554 (total_samples - first_read_size), next_read_size
|
amine@173
|
555 )
|
amine@173
|
556 total_samples_with_overlap = (
|
amine@173
|
557 first_read_size + next_read_size * nb_next_blocks + last_block_size
|
amine@173
|
558 )
|
amine@173
|
559 expected_read_bytes = (
|
amine@173
|
560 total_samples_with_overlap * ads.sw * ads.channels
|
amine@173
|
561 )
|
amine@173
|
562
|
amine@173
|
563 cache_size = (block_size - hop_size) * ads.sample_width * ads.channels
|
amine@172
|
564 total_read = cache_size
|
amine@173
|
565
|
amine@172
|
566 ads.open()
|
amine@172
|
567 i = 0
|
amine@172
|
568 while True:
|
amine@172
|
569 block = ads.read()
|
amine@172
|
570 if block is None:
|
amine@172
|
571 break
|
amine@172
|
572 i += 1
|
amine@172
|
573 total_read += len(block) - cache_size
|
amine@173
|
574
|
amine@172
|
575 ads.close()
|
amine@173
|
576 self.assertEqual(
|
amine@173
|
577 total_read,
|
amine@173
|
578 expected_read_bytes,
|
amine@173
|
579 "Wrong data length read from LimiterADS, expected: {0}, found: {1}".format(
|
amine@173
|
580 expected_read_bytes, total_read
|
amine@173
|
581 ),
|
amine@173
|
582 )
|
amine@173
|
583
|
amine@173
|
584
|
amine@172
|
585 class TestADSFactoryBufferAudioSource(unittest.TestCase):
|
amine@172
|
586 def setUp(self):
|
amine@173
|
587 self.signal = b"ABCDEFGHIJKLMNOPQRSTUVWXYZ012345"
|
amine@173
|
588 self.ads = ADSFactory.ads(
|
amine@173
|
589 data_buffer=self.signal,
|
amine@173
|
590 sampling_rate=16,
|
amine@173
|
591 sample_width=2,
|
amine@173
|
592 channels=1,
|
amine@173
|
593 )
|
amine@173
|
594
|
amine@172
|
595 def test_ADS_BAS_sampling_rate(self):
|
amine@173
|
596 srate = self.ads.sampling_rate
|
amine@173
|
597 self.assertEqual(
|
amine@173
|
598 srate,
|
amine@173
|
599 16,
|
amine@173
|
600 "Wrong sampling rate, expected: 16000, found: {0}".format(srate),
|
amine@173
|
601 )
|
amine@173
|
602
|
amine@172
|
603 def test_ADS_BAS_get_sample_width(self):
|
amine@173
|
604 swidth = self.ads.sample_width
|
amine@173
|
605 self.assertEqual(
|
amine@173
|
606 swidth,
|
amine@173
|
607 2,
|
amine@173
|
608 "Wrong sample width, expected: 2, found: {0}".format(swidth),
|
amine@173
|
609 )
|
amine@173
|
610
|
amine@172
|
611 def test_ADS_BAS_get_channels(self):
|
amine@173
|
612 channels = self.ads.channels
|
amine@173
|
613 self.assertEqual(
|
amine@173
|
614 channels,
|
amine@173
|
615 1,
|
amine@173
|
616 "Wrong number of channels, expected: 1, found: {0}".format(
|
amine@173
|
617 channels
|
amine@173
|
618 ),
|
amine@173
|
619 )
|
amine@173
|
620
|
amine@172
|
621 def test_Limiter_Recorder_Overlap_Deco_rewind_and_read(self):
|
amine@173
|
622
|
amine@172
|
623 # Use arbitrary valid block_size and hop_size
|
amine@172
|
624 block_size = 5
|
amine@172
|
625 hop_size = 4
|
amine@173
|
626
|
amine@173
|
627 ads = ADSFactory.ads(
|
amine@173
|
628 data_buffer=self.signal,
|
amine@173
|
629 sampling_rate=16,
|
amine@173
|
630 sample_width=2,
|
amine@173
|
631 channels=1,
|
amine@173
|
632 max_time=0.80,
|
amine@173
|
633 block_size=block_size,
|
amine@173
|
634 hop_size=hop_size,
|
amine@173
|
635 record=True,
|
amine@173
|
636 )
|
amine@173
|
637
|
amine@172
|
638 # Read all available data overlapping blocks
|
amine@172
|
639 ads.open()
|
amine@172
|
640 i = 0
|
amine@172
|
641 while True:
|
amine@172
|
642 block = ads.read()
|
amine@172
|
643 if block is None:
|
amine@172
|
644 break
|
amine@172
|
645 i += 1
|
amine@173
|
646
|
amine@172
|
647 ads.rewind()
|
amine@173
|
648
|
amine@172
|
649 # Build a BufferAudioSource
|
amine@173
|
650 audio_source = BufferAudioSource(
|
amine@173
|
651 self.signal, ads.sampling_rate, ads.sample_width, ads.channels
|
amine@173
|
652 )
|
amine@172
|
653 audio_source.open()
|
amine@173
|
654
|
amine@172
|
655 # Compare all blocks read from OverlapADS to those read
|
amine@172
|
656 # from an audio source with a manual set_position
|
amine@172
|
657 for j in range(i):
|
amine@173
|
658
|
amine@172
|
659 tmp = audio_source.read(block_size)
|
amine@173
|
660
|
amine@172
|
661 block = ads.read()
|
amine@173
|
662
|
amine@173
|
663 self.assertEqual(
|
amine@173
|
664 block,
|
amine@173
|
665 tmp,
|
amine@173
|
666 "Unexpected block '{}' (N={}) read from OverlapADS".format(
|
amine@173
|
667 block, i
|
amine@173
|
668 ),
|
amine@173
|
669 )
|
amine@173
|
670 audio_source.set_position((j + 1) * hop_size)
|
amine@173
|
671
|
amine@172
|
672 ads.close()
|
amine@172
|
673 audio_source.close()
|
amine@172
|
674
|
amine@172
|
675
|
amine@172
|
676 class TestADSFactoryAlias(unittest.TestCase):
|
amine@172
|
677 def setUp(self):
|
amine@172
|
678 self.signal = "ABCDEFGHIJKLMNOPQRSTUVWXYZ012345"
|
amine@173
|
679
|
amine@172
|
680 def test_sampling_rate_alias(self):
|
amine@173
|
681 ads = ADSFactory.ads(
|
amine@173
|
682 data_buffer=self.signal, sr=16, sample_width=2, channels=1
|
amine@173
|
683 )
|
amine@173
|
684 srate = ads.sampling_rate
|
amine@173
|
685 self.assertEqual(
|
amine@173
|
686 srate,
|
amine@173
|
687 16,
|
amine@173
|
688 "Wrong sampling rate, expected: 16000, found: {0}".format(srate),
|
amine@173
|
689 )
|
amine@173
|
690
|
amine@172
|
691 def test_sampling_rate_duplicate(self):
|
amine@173
|
692 func = partial(
|
amine@173
|
693 ADSFactory.ads,
|
amine@173
|
694 data_buffer=self.signal,
|
amine@173
|
695 sr=16,
|
amine@173
|
696 sampling_rate=16,
|
amine@173
|
697 sample_width=2,
|
amine@173
|
698 channels=1,
|
amine@173
|
699 )
|
amine@172
|
700 self.assertRaises(DuplicateArgument, func)
|
amine@173
|
701
|
amine@172
|
702 def test_sample_width_alias(self):
|
amine@173
|
703 ads = ADSFactory.ads(
|
amine@173
|
704 data_buffer=self.signal, sampling_rate=16, sw=2, channels=1
|
amine@173
|
705 )
|
amine@173
|
706 swidth = ads.sample_width
|
amine@173
|
707 self.assertEqual(
|
amine@173
|
708 swidth,
|
amine@173
|
709 2,
|
amine@173
|
710 "Wrong sample width, expected: 2, found: {0}".format(swidth),
|
amine@173
|
711 )
|
amine@173
|
712
|
amine@172
|
713 def test_sample_width_duplicate(self):
|
amine@173
|
714 func = partial(
|
amine@173
|
715 ADSFactory.ads,
|
amine@173
|
716 data_buffer=self.signal,
|
amine@173
|
717 sampling_rate=16,
|
amine@173
|
718 sw=2,
|
amine@173
|
719 sample_width=2,
|
amine@173
|
720 channels=1,
|
amine@173
|
721 )
|
amine@172
|
722 self.assertRaises(DuplicateArgument, func)
|
amine@173
|
723
|
amine@172
|
724 def test_channels_alias(self):
|
amine@173
|
725 ads = ADSFactory.ads(
|
amine@173
|
726 data_buffer=self.signal, sampling_rate=16, sample_width=2, ch=1
|
amine@173
|
727 )
|
amine@173
|
728 channels = ads.channels
|
amine@173
|
729 self.assertEqual(
|
amine@173
|
730 channels,
|
amine@173
|
731 1,
|
amine@173
|
732 "Wrong number of channels, expected: 1, found: {0}".format(
|
amine@173
|
733 channels
|
amine@173
|
734 ),
|
amine@173
|
735 )
|
amine@173
|
736
|
amine@172
|
737 def test_channels_duplicate(self):
|
amine@173
|
738 func = partial(
|
amine@173
|
739 ADSFactory.ads,
|
amine@173
|
740 data_buffer=self.signal,
|
amine@173
|
741 sampling_rate=16,
|
amine@173
|
742 sample_width=2,
|
amine@173
|
743 ch=1,
|
amine@173
|
744 channels=1,
|
amine@173
|
745 )
|
amine@172
|
746 self.assertRaises(DuplicateArgument, func)
|
amine@173
|
747
|
amine@172
|
748 def test_block_size_alias(self):
|
amine@173
|
749 ads = ADSFactory.ads(
|
amine@173
|
750 data_buffer=self.signal,
|
amine@173
|
751 sampling_rate=16,
|
amine@173
|
752 sample_width=2,
|
amine@173
|
753 channels=1,
|
amine@173
|
754 bs=8,
|
amine@173
|
755 )
|
amine@173
|
756 size = ads.block_size
|
amine@173
|
757 self.assertEqual(
|
amine@173
|
758 size,
|
amine@173
|
759 8,
|
amine@173
|
760 "Wrong block_size using bs alias, expected: 8, found: {0}".format(
|
amine@173
|
761 size
|
amine@173
|
762 ),
|
amine@173
|
763 )
|
amine@173
|
764
|
amine@172
|
765 def test_block_size_duplicate(self):
|
amine@173
|
766 func = partial(
|
amine@173
|
767 ADSFactory.ads,
|
amine@173
|
768 data_buffer=self.signal,
|
amine@173
|
769 sampling_rate=16,
|
amine@173
|
770 sample_width=2,
|
amine@173
|
771 channels=1,
|
amine@173
|
772 bs=4,
|
amine@173
|
773 block_size=4,
|
amine@173
|
774 )
|
amine@172
|
775 self.assertRaises(DuplicateArgument, func)
|
amine@173
|
776
|
amine@172
|
777 def test_block_duration_alias(self):
|
amine@173
|
778 ads = ADSFactory.ads(
|
amine@173
|
779 data_buffer=self.signal,
|
amine@173
|
780 sampling_rate=16,
|
amine@173
|
781 sample_width=2,
|
amine@173
|
782 channels=1,
|
amine@173
|
783 bd=0.75,
|
amine@173
|
784 )
|
amine@172
|
785 # 0.75 ms = 0.75 * 16 = 12
|
amine@173
|
786 size = ads.block_size
|
amine@173
|
787 self.assertEqual(
|
amine@173
|
788 size,
|
amine@173
|
789 12,
|
amine@173
|
790 "Wrong block_size set with a block_dur alias 'bd', expected: 8, found: {0}".format(
|
amine@173
|
791 size
|
amine@173
|
792 ),
|
amine@173
|
793 )
|
amine@173
|
794
|
amine@172
|
795 def test_block_duration_duplicate(self):
|
amine@173
|
796 func = partial(
|
amine@173
|
797 ADSFactory.ads,
|
amine@173
|
798 data_buffer=self.signal,
|
amine@173
|
799 sampling_rate=16,
|
amine@173
|
800 sample_width=2,
|
amine@173
|
801 channels=1,
|
amine@173
|
802 bd=4,
|
amine@173
|
803 block_dur=4,
|
amine@173
|
804 )
|
amine@172
|
805 self.assertRaises(DuplicateArgument, func)
|
amine@173
|
806
|
amine@172
|
807 def test_block_size_duration_duplicate(self):
|
amine@173
|
808 func = partial(
|
amine@173
|
809 ADSFactory.ads,
|
amine@173
|
810 data_buffer=self.signal,
|
amine@173
|
811 sampling_rate=16,
|
amine@173
|
812 sample_width=2,
|
amine@173
|
813 channels=1,
|
amine@173
|
814 bd=4,
|
amine@173
|
815 bs=12,
|
amine@173
|
816 )
|
amine@172
|
817 self.assertRaises(DuplicateArgument, func)
|
amine@173
|
818
|
amine@172
|
819 def test_hop_duration_alias(self):
|
amine@173
|
820
|
amine@173
|
821 ads = ADSFactory.ads(
|
amine@173
|
822 data_buffer=self.signal,
|
amine@173
|
823 sampling_rate=16,
|
amine@173
|
824 sample_width=2,
|
amine@173
|
825 channels=1,
|
amine@173
|
826 bd=0.75,
|
amine@173
|
827 hd=0.5,
|
amine@173
|
828 )
|
amine@172
|
829 size = ads.hop_size
|
amine@173
|
830 self.assertEqual(
|
amine@173
|
831 size,
|
amine@173
|
832 8,
|
amine@173
|
833 "Wrong block_size using bs alias, expected: 8, found: {0}".format(
|
amine@173
|
834 size
|
amine@173
|
835 ),
|
amine@173
|
836 )
|
amine@173
|
837
|
amine@172
|
838 def test_hop_duration_duplicate(self):
|
amine@173
|
839
|
amine@173
|
840 func = partial(
|
amine@173
|
841 ADSFactory.ads,
|
amine@173
|
842 data_buffer=self.signal,
|
amine@173
|
843 sampling_rate=16,
|
amine@173
|
844 sample_width=2,
|
amine@173
|
845 channels=1,
|
amine@173
|
846 bd=0.75,
|
amine@173
|
847 hd=0.5,
|
amine@173
|
848 hop_dur=0.5,
|
amine@173
|
849 )
|
amine@172
|
850 self.assertRaises(DuplicateArgument, func)
|
amine@173
|
851
|
amine@172
|
852 def test_hop_size_duration_duplicate(self):
|
amine@173
|
853 func = partial(
|
amine@173
|
854 ADSFactory.ads,
|
amine@173
|
855 data_buffer=self.signal,
|
amine@173
|
856 sampling_rate=16,
|
amine@173
|
857 sample_width=2,
|
amine@173
|
858 channels=1,
|
amine@173
|
859 bs=8,
|
amine@173
|
860 hs=4,
|
amine@173
|
861 hd=1,
|
amine@173
|
862 )
|
amine@172
|
863 self.assertRaises(DuplicateArgument, func)
|
amine@173
|
864
|
amine@172
|
865 def test_hop_size_greater_than_block_size(self):
|
amine@173
|
866 func = partial(
|
amine@173
|
867 ADSFactory.ads,
|
amine@173
|
868 data_buffer=self.signal,
|
amine@173
|
869 sampling_rate=16,
|
amine@173
|
870 sample_width=2,
|
amine@173
|
871 channels=1,
|
amine@173
|
872 bs=4,
|
amine@173
|
873 hs=8,
|
amine@173
|
874 )
|
amine@172
|
875 self.assertRaises(ValueError, func)
|
amine@173
|
876
|
amine@172
|
877 def test_filename_alias(self):
|
amine@172
|
878 ads = ADSFactory.ads(fn=dataset.one_to_six_arabic_16000_mono_bc_noise)
|
amine@173
|
879
|
amine@172
|
880 def test_filename_duplicate(self):
|
amine@173
|
881
|
amine@173
|
882 func = partial(
|
amine@173
|
883 ADSFactory.ads,
|
amine@173
|
884 fn=dataset.one_to_six_arabic_16000_mono_bc_noise,
|
amine@173
|
885 filename=dataset.one_to_six_arabic_16000_mono_bc_noise,
|
amine@173
|
886 )
|
amine@172
|
887 self.assertRaises(DuplicateArgument, func)
|
amine@173
|
888
|
amine@172
|
889 def test_data_buffer_duplicate(self):
|
amine@173
|
890 func = partial(
|
amine@173
|
891 ADSFactory.ads,
|
amine@173
|
892 data_buffer=self.signal,
|
amine@173
|
893 db=self.signal,
|
amine@173
|
894 sampling_rate=16,
|
amine@173
|
895 sample_width=2,
|
amine@173
|
896 channels=1,
|
amine@173
|
897 )
|
amine@172
|
898 self.assertRaises(DuplicateArgument, func)
|
amine@173
|
899
|
amine@172
|
900 def test_max_time_alias(self):
|
amine@173
|
901 ads = ADSFactory.ads(
|
amine@173
|
902 data_buffer=self.signal,
|
amine@173
|
903 sampling_rate=16,
|
amine@173
|
904 sample_width=2,
|
amine@173
|
905 channels=1,
|
amine@173
|
906 mt=10,
|
amine@173
|
907 )
|
amine@173
|
908 self.assertEqual(
|
amine@173
|
909 ads.max_read,
|
amine@173
|
910 10,
|
amine@173
|
911 "Wrong AudioDataSource.max_read, expected: 10, found: {}".format(
|
amine@173
|
912 ads.max_read
|
amine@173
|
913 ),
|
amine@173
|
914 )
|
amine@173
|
915
|
amine@172
|
916 def test_max_time_duplicate(self):
|
amine@173
|
917 func = partial(
|
amine@173
|
918 ADSFactory.ads,
|
amine@173
|
919 data_buffer=self.signal,
|
amine@173
|
920 sampling_rate=16,
|
amine@173
|
921 sample_width=2,
|
amine@173
|
922 channels=1,
|
amine@173
|
923 mt=True,
|
amine@173
|
924 max_time=True,
|
amine@173
|
925 )
|
amine@173
|
926
|
amine@172
|
927 self.assertRaises(DuplicateArgument, func)
|
amine@173
|
928
|
amine@172
|
929 def test_record_alias(self):
|
amine@173
|
930 ads = ADSFactory.ads(
|
amine@173
|
931 data_buffer=self.signal,
|
amine@173
|
932 sampling_rate=16,
|
amine@173
|
933 sample_width=2,
|
amine@173
|
934 channels=1,
|
amine@173
|
935 rec=True,
|
amine@173
|
936 )
|
amine@173
|
937 self.assertTrue(
|
amine@173
|
938 ads.rewindable, "AudioDataSource.rewindable expected to be True"
|
amine@173
|
939 )
|
amine@173
|
940
|
amine@172
|
941 def test_record_duplicate(self):
|
amine@173
|
942 func = partial(
|
amine@173
|
943 ADSFactory.ads,
|
amine@173
|
944 data_buffer=self.signal,
|
amine@173
|
945 sampling_rate=16,
|
amine@173
|
946 sample_width=2,
|
amine@173
|
947 channels=1,
|
amine@173
|
948 rec=True,
|
amine@173
|
949 record=True,
|
amine@173
|
950 )
|
amine@172
|
951 self.assertRaises(DuplicateArgument, func)
|
amine@173
|
952
|
amine@172
|
953 def test_Limiter_Recorder_Overlap_Deco_rewind_and_read_alias(self):
|
amine@173
|
954
|
amine@172
|
955 # Use arbitrary valid block_size and hop_size
|
amine@172
|
956 block_size = 5
|
amine@172
|
957 hop_size = 4
|
amine@173
|
958
|
amine@173
|
959 ads = ADSFactory.ads(
|
amine@173
|
960 db=self.signal,
|
amine@173
|
961 sr=16,
|
amine@173
|
962 sw=2,
|
amine@173
|
963 ch=1,
|
amine@173
|
964 mt=0.80,
|
amine@173
|
965 bs=block_size,
|
amine@173
|
966 hs=hop_size,
|
amine@173
|
967 rec=True,
|
amine@173
|
968 )
|
amine@173
|
969
|
amine@172
|
970 # Read all available data overlapping blocks
|
amine@172
|
971 ads.open()
|
amine@172
|
972 i = 0
|
amine@172
|
973 while True:
|
amine@172
|
974 block = ads.read()
|
amine@172
|
975 if block is None:
|
amine@172
|
976 break
|
amine@172
|
977 i += 1
|
amine@173
|
978
|
amine@172
|
979 ads.rewind()
|
amine@173
|
980
|
amine@172
|
981 # Build a BufferAudioSource
|
amine@173
|
982 audio_source = BufferAudioSource(
|
amine@173
|
983 self.signal, ads.sampling_rate, ads.sample_width, ads.channels
|
amine@173
|
984 )
|
amine@172
|
985 audio_source.open()
|
amine@173
|
986
|
amine@173
|
987 # Compare all blocks read from AudioDataSource to those read
|
amine@173
|
988 # from an audio source with manual position definition
|
amine@172
|
989 for j in range(i):
|
amine@172
|
990 tmp = audio_source.read(block_size)
|
amine@172
|
991 block = ads.read()
|
amine@173
|
992 self.assertEqual(
|
amine@173
|
993 block,
|
amine@173
|
994 tmp,
|
amine@173
|
995 "Unexpected block (N={0}) read from OverlapADS".format(i),
|
amine@173
|
996 )
|
amine@173
|
997 audio_source.set_position((j + 1) * hop_size)
|
amine@172
|
998 ads.close()
|
amine@172
|
999 audio_source.close()
|
amine@173
|
1000
|
amine@172
|
1001
|
amine@172
|
1002 if __name__ == "__main__":
|
amine@173
|
1003 # import sys;sys.argv = ['', 'Test.testName']
|
amine@172
|
1004 unittest.main()
|