feat: audio module — sounddevice recorder with WAV export
This commit is contained in:
@@ -0,0 +1,46 @@
|
||||
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
|
||||
Reference in New Issue
Block a user