-
Notifications
You must be signed in to change notification settings - Fork 145
/
rec_gui.py
executable file
·236 lines (188 loc) · 7.47 KB
/
rec_gui.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
#!/usr/bin/env python3
"""Simple GUI for recording into a WAV file.
There are 3 concurrent activities: GUI, audio callback, file-writing thread.
Neither the GUI nor the audio callback is supposed to block.
Blocking in any of the GUI functions could make the GUI "freeze", blocking in
the audio callback could lead to drop-outs in the recording.
Blocking the file-writing thread for some time is no problem, as long as the
recording can be stopped successfully when it is supposed to.
"""
import contextlib
import queue
import sys
import tempfile
import threading
import tkinter as tk
from tkinter import ttk
from tkinter.simpledialog import Dialog
import numpy as np
import sounddevice as sd
import soundfile as sf
def file_writing_thread(*, q, **soundfile_args):
"""Write data from queue to file until *None* is received."""
# NB: If you want fine-grained control about the buffering of the file, you
# can use Python's open() function (with the "buffering" argument) and
# pass the resulting file object to sf.SoundFile().
with sf.SoundFile(**soundfile_args) as f:
while True:
data = q.get()
if data is None:
break
f.write(data)
class SettingsWindow(Dialog):
"""Dialog window for choosing sound device."""
def body(self, master):
ttk.Label(master, text='Select host API:').pack(anchor='w')
self.hostapi_list = ttk.Combobox(master, state='readonly', width=50)
self.hostapi_list.pack()
self.hostapi_list['values'] = [
hostapi['name'] for hostapi in sd.query_hostapis()]
ttk.Label(master, text='Select sound device:').pack(anchor='w')
self.device_ids = []
self.device_list = ttk.Combobox(master, state='readonly', width=50)
self.device_list.pack()
self.hostapi_list.bind('<<ComboboxSelected>>', self.update_device_list)
with contextlib.suppress(sd.PortAudioError):
self.hostapi_list.current(sd.default.hostapi)
self.hostapi_list.event_generate('<<ComboboxSelected>>')
def update_device_list(self, *args):
hostapi = sd.query_hostapis(self.hostapi_list.current())
self.device_ids = [
idx
for idx in hostapi['devices']
if sd.query_devices(idx)['max_input_channels'] > 0]
self.device_list['values'] = [
sd.query_devices(idx)['name'] for idx in self.device_ids]
default = hostapi['default_input_device']
if default >= 0:
self.device_list.current(self.device_ids.index(default))
def validate(self):
self.result = self.device_ids[self.device_list.current()]
return True
class RecGui(tk.Tk):
stream = None
def __init__(self):
super().__init__()
self.title('Recording GUI')
padding = 10
f = ttk.Frame()
self.rec_button = ttk.Button(f)
self.rec_button.pack(side='left', padx=padding, pady=padding)
self.settings_button = ttk.Button(
f, text='settings', command=self.on_settings)
self.settings_button.pack(side='left', padx=padding, pady=padding)
f.pack(expand=True, padx=padding, pady=padding)
self.file_label = ttk.Label(text='<file name>')
self.file_label.pack(anchor='w')
self.input_overflows = 0
self.status_label = ttk.Label()
self.status_label.pack(anchor='w')
self.meter = ttk.Progressbar()
self.meter['orient'] = 'horizontal'
self.meter['mode'] = 'determinate'
self.meter['maximum'] = 1.0
self.meter.pack(fill='x')
# We try to open a stream with default settings first, if that doesn't
# work, the user can manually change the device(s)
self.create_stream()
self.recording = self.previously_recording = False
self.audio_q = queue.Queue()
self.peak = 0
self.metering_q = queue.Queue(maxsize=1)
self.protocol('WM_DELETE_WINDOW', self.close_window)
self.init_buttons()
self.update_gui()
def create_stream(self, device=None):
if self.stream is not None:
self.stream.close()
self.stream = sd.InputStream(
device=device, channels=1, callback=self.audio_callback)
self.stream.start()
def audio_callback(self, indata, frames, time, status):
"""This is called (from a separate thread) for each audio block."""
if status.input_overflow:
# NB: This increment operation is not atomic, but this doesn't
# matter since no other thread is writing to the attribute.
self.input_overflows += 1
# NB: self.recording is accessed from different threads.
# This is safe because here we are only accessing it once (with a
# single bytecode instruction).
if self.recording:
self.audio_q.put(indata.copy())
self.previously_recording = True
else:
if self.previously_recording:
self.audio_q.put(None)
self.previously_recording = False
self.peak = max(self.peak, np.max(np.abs(indata)))
try:
self.metering_q.put_nowait(self.peak)
except queue.Full:
pass
else:
self.peak = 0
def on_rec(self):
self.settings_button['state'] = 'disabled'
self.recording = True
filename = tempfile.mktemp(
prefix='delme_rec_gui_', suffix='.wav', dir='')
if self.audio_q.qsize() != 0:
print('WARNING: Queue not empty!')
self.thread = threading.Thread(
target=file_writing_thread,
kwargs=dict(
file=filename,
mode='x',
samplerate=int(self.stream.samplerate),
channels=self.stream.channels,
q=self.audio_q,
),
)
self.thread.start()
# NB: File creation might fail! For brevity, we don't check for this.
self.rec_button['text'] = 'stop'
self.rec_button['command'] = self.on_stop
self.rec_button['state'] = 'normal'
self.file_label['text'] = filename
def on_stop(self, *args):
self.rec_button['state'] = 'disabled'
self.recording = False
self.wait_for_thread()
def wait_for_thread(self):
# NB: Waiting time could be calculated based on stream.latency
self.after(10, self._wait_for_thread)
def _wait_for_thread(self):
if self.thread.is_alive():
self.wait_for_thread()
return
self.thread.join()
self.init_buttons()
def on_settings(self, *args):
w = SettingsWindow(self, 'Settings')
if w.result is not None:
self.create_stream(device=w.result)
def init_buttons(self):
self.rec_button['text'] = 'record'
self.rec_button['command'] = self.on_rec
if self.stream:
self.rec_button['state'] = 'normal'
self.settings_button['state'] = 'normal'
def update_gui(self):
self.status_label['text'] = 'input overflows: {}'.format(
self.input_overflows)
try:
peak = self.metering_q.get_nowait()
except queue.Empty:
pass
else:
self.meter['value'] = peak
self.after(100, self.update_gui)
def close_window(self):
if self.recording:
self.on_stop()
self.destroy()
def main():
app = RecGui()
app.mainloop()
if __name__ == '__main__':
main()