142 lines
4.4 KiB
Python
142 lines
4.4 KiB
Python
import sounddevice as sd
|
|
import numpy as np
|
|
import os
|
|
from datetime import datetime
|
|
import scipy.io.wavfile as wavfile
|
|
|
|
class AudioRecorder:
|
|
_instance = None
|
|
|
|
def __new__(cls, *args, **kwargs):
|
|
if cls._instance is None:
|
|
print("Creating new AudioRecorder instance")
|
|
cls._instance = super().__new__(cls)
|
|
return cls._instance
|
|
def __init__(self, duration=30, sample_rate=44100, channels=2, recordings_dir='recordings'):
|
|
"""
|
|
Initialize audio recorder with configurable parameters.
|
|
|
|
:param duration: Length of audio buffer in seconds
|
|
:param sample_rate: Audio sample rate (if None, use default device sample rate)
|
|
:param channels: Number of audio channels
|
|
"""
|
|
|
|
self.duration = duration
|
|
self.sample_rate = sample_rate
|
|
self.channels = channels
|
|
self.buffer = np.zeros((int(duration * sample_rate), channels), dtype=np.float32)
|
|
self.recordings_dir = recordings_dir
|
|
self.stream = sd.InputStream(
|
|
samplerate=self.sample_rate,
|
|
channels=self.channels,
|
|
callback=self.record_callback
|
|
)
|
|
|
|
def refresh_stream(self):
|
|
"""
|
|
Refresh the audio stream with updated parameters.
|
|
"""
|
|
was_active = self.stream.active
|
|
if was_active:
|
|
self.stream.stop()
|
|
|
|
self.buffer = np.zeros((int(self.duration * self.sample_rate), self.channels), dtype=np.float32)
|
|
|
|
self.stream = sd.InputStream(
|
|
samplerate=self.sample_rate,
|
|
channels=self.channels,
|
|
callback=self.record_callback
|
|
)
|
|
|
|
if was_active:
|
|
self.stream.start()
|
|
|
|
|
|
|
|
|
|
def record_callback(self, indata, frames, time, status):
|
|
"""
|
|
Circular buffer callback for continuous recording.
|
|
|
|
:param indata: Input audio data
|
|
:param frames: Number of frames
|
|
:param time: Timestamp
|
|
:param status: Recording status
|
|
"""
|
|
if status:
|
|
print(f"Recording status: {status}")
|
|
|
|
# Circular buffer implementation
|
|
self.buffer = np.roll(self.buffer, -frames, axis=0)
|
|
self.buffer[-frames:] = indata
|
|
|
|
def save_last_n_seconds(self):
|
|
"""
|
|
Save the last n seconds of audio to a file.
|
|
|
|
:param output_dir: Directory to save recordings
|
|
:return: Path to saved audio file
|
|
"""
|
|
# Create output directory if it doesn't exist
|
|
os.makedirs(self.recordings_dir, exist_ok=True)
|
|
|
|
# Generate filename with timestamp
|
|
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
|
|
filename = os.path.join(self.recordings_dir, f"audio_capture_{timestamp}.wav")
|
|
|
|
# Normalize audio to prevent clipping
|
|
audio_data = self.buffer / np.max(np.abs(self.buffer)) * .5
|
|
|
|
# Convert float32 to int16 for WAV file
|
|
audio_data_int16 = (audio_data * 32767).astype(np.int16)
|
|
|
|
# Write buffer to file
|
|
wavfile.write(filename, int(self.sample_rate), audio_data_int16)
|
|
|
|
return filename
|
|
|
|
def set_buffer_duration(self, duration):
|
|
"""
|
|
Set the duration of the audio buffer.
|
|
|
|
:param duration: New buffer duration in seconds
|
|
"""
|
|
self.duration = duration
|
|
self.buffer = np.zeros((int(duration * self.sample_rate), self.channels), dtype=np.float32)
|
|
|
|
def set_recording_directory(self, directory):
|
|
"""
|
|
Set the directory where recordings will be saved.
|
|
|
|
:param directory: Path to the recordings directory
|
|
"""
|
|
self.recordings_dir = directory
|
|
|
|
def start_recording(self):
|
|
"""
|
|
Start continuous audio recording with circular buffer.
|
|
"""
|
|
if(self.stream.active):
|
|
print("Already recording")
|
|
return
|
|
print('number of channels', self.channels)
|
|
|
|
self.stream.start()
|
|
|
|
def stop_recording(self):
|
|
"""
|
|
Stop continuous audio recording with circular buffer.
|
|
"""
|
|
if(not self.stream.active):
|
|
print("Already stopped")
|
|
return
|
|
|
|
self.stream.stop()
|
|
|
|
def is_recording(self):
|
|
"""
|
|
Check if the audio stream is currently active.
|
|
|
|
:return: True if recording, False otherwise
|
|
"""
|
|
return self.stream.active |