amine@106
|
1 import os
|
amine@106
|
2 import sys
|
amine@106
|
3 import math
|
amine@107
|
4 from array import array
|
amine@110
|
5 from tempfile import NamedTemporaryFile
|
amine@110
|
6 import filecmp
|
amine@108
|
7 from unittest import TestCase
|
amine@108
|
8 from genty import genty, genty_dataset
|
amine@110
|
9 from auditok.io import (
|
amine@110
|
10 DATA_FORMAT,
|
amine@110
|
11 AudioParameterError,
|
amine@110
|
12 check_audio_data,
|
amine@116
|
13 _array_to_bytes,
|
amine@111
|
14 _save_raw,
|
amine@110
|
15 _save_wave,
|
amine@110
|
16 )
|
amine@106
|
17
|
amine@106
|
18
|
amine@106
|
19 if sys.version_info >= (3, 0):
|
amine@106
|
20 PYTHON_3 = True
|
amine@106
|
21 else:
|
amine@106
|
22 PYTHON_3 = False
|
amine@106
|
23
|
amine@106
|
24
|
amine@106
|
25 def _sample_generator(*data_buffers):
|
amine@106
|
26 """
|
amine@106
|
27 Takes a list of many mono audio data buffers and makes a sample generator
|
amine@106
|
28 of interleaved audio samples, one sample from each channel. The resulting
|
amine@106
|
29 generator can be used to build a multichannel audio buffer.
|
amine@106
|
30 >>> gen = _sample_generator("abcd", "ABCD")
|
amine@106
|
31 >>> list(gen)
|
amine@106
|
32 ["a", "A", "b", "B", "c", "C", "d", "D"]
|
amine@106
|
33 """
|
amine@106
|
34 frame_gen = zip(*data_buffers)
|
amine@106
|
35 return (sample for frame in frame_gen for sample in frame)
|
amine@106
|
36
|
amine@106
|
37
|
amine@107
|
38 def _generate_pure_tone(
|
amine@107
|
39 frequency, duration_sec=1, sampling_rate=16000, sample_width=2, volume=1e4
|
amine@107
|
40 ):
|
amine@107
|
41 """
|
amine@107
|
42 Generates a pure tone with the given frequency.
|
amine@107
|
43 """
|
amine@107
|
44 assert frequency <= sampling_rate / 2
|
amine@107
|
45 max_value = (2 ** (sample_width * 8) // 2) - 1
|
amine@107
|
46 if volume > max_value:
|
amine@107
|
47 volume = max_value
|
amine@107
|
48 fmt = DATA_FORMAT[sample_width]
|
amine@107
|
49 total_samples = int(sampling_rate * duration_sec)
|
amine@107
|
50 step = frequency / sampling_rate
|
amine@107
|
51 two_pi_step = 2 * math.pi * step
|
amine@107
|
52 data = array(
|
amine@107
|
53 fmt,
|
amine@107
|
54 (
|
amine@107
|
55 int(math.sin(two_pi_step * i) * volume)
|
amine@107
|
56 for i in range(total_samples)
|
amine@107
|
57 ),
|
amine@107
|
58 )
|
amine@107
|
59 return data
|
amine@107
|
60
|
amine@107
|
61
|
amine@107
|
62 PURE_TONE_DICT = {
|
amine@107
|
63 freq: _generate_pure_tone(freq, 1, 16000, 2) for freq in (400, 800, 1600)
|
amine@107
|
64 }
|
amine@107
|
65 PURE_TONE_DICT.update(
|
amine@107
|
66 {
|
amine@107
|
67 freq: _generate_pure_tone(freq, 0.1, 16000, 2)
|
amine@107
|
68 for freq in (600, 1150, 2400, 7220)
|
amine@107
|
69 }
|
amine@107
|
70 )
|
amine@108
|
71
|
amine@108
|
72
|
amine@108
|
73 @genty
|
amine@108
|
74 class TestIO(TestCase):
|
amine@108
|
75 @genty_dataset(
|
amine@108
|
76 valid_mono=(b"\0" * 113, 1, 1),
|
amine@108
|
77 valid_stereo=(b"\0" * 160, 1, 2),
|
amine@108
|
78 invalid_mono_sw_2=(b"\0" * 113, 2, 1, False),
|
amine@108
|
79 invalid_stereo_sw_1=(b"\0" * 113, 1, 2, False),
|
amine@108
|
80 invalid_stereo_sw_2=(b"\0" * 158, 2, 2, False),
|
amine@108
|
81 )
|
amine@108
|
82 def test_check_audio_data(self, data, sample_width, channels, valid=True):
|
amine@108
|
83
|
amine@108
|
84 if not valid:
|
amine@108
|
85 with self.assertRaises(AudioParameterError):
|
amine@108
|
86 check_audio_data(data, sample_width, channels)
|
amine@108
|
87 else:
|
amine@108
|
88 self.assertIsNone(check_audio_data(data, sample_width, channels))
|
amine@110
|
89
|
amine@110
|
90 @genty_dataset(
|
amine@111
|
91 mono=("mono_400Hz.raw", (400,)),
|
amine@111
|
92 three_channel=("3channel_400-800-1600Hz.raw", (400, 800, 1600)),
|
amine@111
|
93 )
|
amine@111
|
94 def test_save_raw(self, filename, frequencies):
|
amine@111
|
95 filename = "tests/data/test_16KHZ_{}".format(filename)
|
amine@111
|
96 sample_width = 2
|
amine@111
|
97 fmt = DATA_FORMAT[sample_width]
|
amine@111
|
98 mono_channels = [PURE_TONE_DICT[freq] for freq in frequencies]
|
amine@111
|
99 data = _array_to_bytes(array(fmt, _sample_generator(*mono_channels)))
|
amine@111
|
100 tmpfile = NamedTemporaryFile()
|
amine@111
|
101 _save_raw(tmpfile.name, data)
|
amine@111
|
102 self.assertTrue(filecmp.cmp(tmpfile.name, filename, shallow=False))
|
amine@111
|
103
|
amine@111
|
104 @genty_dataset(
|
amine@110
|
105 mono=("mono_400Hz.wav", (400,)),
|
amine@110
|
106 three_channel=("3channel_400-800-1600Hz.wav", (400, 800, 1600)),
|
amine@110
|
107 )
|
amine@110
|
108 def test_save_wave(self, filename, frequencies):
|
amine@110
|
109 filename = "tests/data/test_16KHZ_{}".format(filename)
|
amine@110
|
110 sampling_rate = 16000
|
amine@110
|
111 sample_width = 2
|
amine@110
|
112 channels = len(frequencies)
|
amine@110
|
113 fmt = DATA_FORMAT[sample_width]
|
amine@110
|
114 mono_channels = [PURE_TONE_DICT[freq] for freq in frequencies]
|
amine@110
|
115 data = _array_to_bytes(array(fmt, _sample_generator(*mono_channels)))
|
amine@110
|
116 tmpfile = NamedTemporaryFile()
|
amine@110
|
117 _save_wave(tmpfile.name, data, sampling_rate, sample_width, channels)
|
amine@110
|
118 self.assertTrue(filecmp.cmp(tmpfile.name, filename, shallow=False))
|