47 lines
1.3 KiB
Python
47 lines
1.3 KiB
Python
import wave
|
|
import threading
|
|
import numpy as np
|
|
|
|
|
|
class AudioRecorder:
|
|
def __init__(self, sample_rate: int = 16000):
|
|
self.sample_rate = sample_rate
|
|
self._buffer: list[np.ndarray] = []
|
|
self._stream = None
|
|
self.is_recording = False
|
|
self._lock = threading.Lock()
|
|
|
|
def _callback(self, indata, frames, time, status):
|
|
if self.is_recording:
|
|
with self._lock:
|
|
self._buffer.append(indata[:, 0].copy().astype(np.int16))
|
|
|
|
def start(self):
|
|
import sounddevice as sd
|
|
self._buffer = []
|
|
self.is_recording = True
|
|
self._stream = sd.InputStream(
|
|
samplerate=self.sample_rate,
|
|
channels=1,
|
|
dtype="int16",
|
|
callback=self._callback,
|
|
)
|
|
self._stream.start()
|
|
|
|
def stop(self):
|
|
self.is_recording = False
|
|
if self._stream:
|
|
self._stream.stop()
|
|
self._stream.close()
|
|
self._stream = None
|
|
|
|
def save_wav(self, path: str) -> str:
|
|
with self._lock:
|
|
data = np.concatenate(self._buffer) if self._buffer else np.zeros(0, dtype=np.int16)
|
|
with wave.open(path, "wb") as wf:
|
|
wf.setnchannels(1)
|
|
wf.setsampwidth(2)
|
|
wf.setframerate(self.sample_rate)
|
|
wf.writeframes(data.tobytes())
|
|
return path
|