Spaces:
Paused
Paused
File size: 1,603 Bytes
182c2c2 | 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 | import numpy as np
import soundfile as sf
import librosa
from io import BytesIO
from typing import Optional
class AudioBuffer:
"""Buffer to accumulate streaming audio data"""
def __init__(self, chunk_duration: float = 30.0, sample_rate: int = 16000):
self.chunk_duration = chunk_duration
self.sample_rate = sample_rate
self.buffer = np.array([], dtype=np.float32)
self.target_chunk_size = int(sample_rate * chunk_duration)
def add_data(self, audio_data: bytes) -> Optional[np.ndarray]:
"""Add audio data to buffer and return chunk if ready"""
# Convert bytes to audio array
audio_buffer = BytesIO(audio_data)
audio_chunk, sample_rate = sf.read(audio_buffer)
# Ensure mono audio
if len(audio_chunk.shape) > 1:
audio_chunk = audio_chunk.mean(axis=1)
# Resample to target sample rate if needed
if sample_rate != self.sample_rate:
audio_chunk = librosa.resample(audio_chunk, orig_sr=sample_rate, target_sr=self.sample_rate)
self.buffer = np.concatenate([self.buffer, audio_chunk.astype(np.float32)])
if len(self.buffer) >= self.target_chunk_size:
# Extract chunk
chunk = self.buffer[:self.target_chunk_size]
# Keep remaining data in buffer
self.buffer = self.buffer[self.target_chunk_size:]
return chunk
return None
def get_remaining(self) -> np.ndarray:
"""Get any remaining audio data"""
return self.buffer
|