-
Notifications
You must be signed in to change notification settings - Fork 15
/
Copy pathpyaudio_recording.py
69 lines (56 loc) · 2.33 KB
/
pyaudio_recording.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
import wave
from dataclasses import dataclass, asdict
import pyaudio
@dataclass
class StreamParams:
format: int = pyaudio.paInt16
channels: int = 2
rate: int = 44100
frames_per_buffer: int = 1024
input: bool = True
output: bool = False
def to_dict(self) -> dict:
return asdict(self)
class Recorder:
"""Recorder uses the blocking I/O facility from pyaudio to record sound
from mic.
Attributes:
- stream_params: StreamParams object with values for pyaudio Stream
object
"""
def __init__(self, stream_params: StreamParams) -> None:
self.stream_params = stream_params
self._pyaudio = None
self._stream = None
self._wav_file = None
def record(self, duration: int, save_path: str) -> None:
"""Record sound from mic for a given amount of seconds.
:param duration: Number of seconds we want to record for
:param save_path: Where to store recording
"""
print(f"Start recording for {duration} seconds...")
self._create_recording_resources(save_path)
self._write_wav_file_reading_from_stream(duration)
self._close_recording_resources()
print("Stop recording")
def _create_recording_resources(self, save_path: str) -> None:
self._pyaudio = pyaudio.PyAudio()
self._stream = self._pyaudio.open(**self.stream_params.to_dict())
self._create_wav_file(save_path)
def _create_wav_file(self, save_path: str):
self._wav_file = wave.open(save_path, "wb")
self._wav_file.setnchannels(self.stream_params.channels)
self._wav_file.setsampwidth(self._pyaudio.get_sample_size(self.stream_params.format))
self._wav_file.setframerate(self.stream_params.rate)
def _write_wav_file_reading_from_stream(self, duration: int) -> None:
for _ in range(int(self.stream_params.rate * duration / self.stream_params.frames_per_buffer)):
audio_data = self._stream.read(self.stream_params.frames_per_buffer)
self._wav_file.writeframes(audio_data)
def _close_recording_resources(self) -> None:
self._wav_file.close()
self._stream.close()
self._pyaudio.terminate()
if __name__ == "__main__":
stream_params = StreamParams()
recorder = Recorder(stream_params)
recorder.record(5, "audio.wav")