Spaces:
Sleeping
Sleeping
File size: 1,722 Bytes
c1e08a0 29e2c21 c1e08a0 |
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 |
"""Module providing abstract base class for audio input handling."""
from abc import ABC, abstractmethod
from typing import Callable
import numpy as np
class AudioProcessor(ABC):
"""Abstract base class for audio input handling."""
def __init__(
self,
sample_rate: int,
callback: Callable[[np.ndarray], None] | None = None,
buffer_duration: float = 0.3,
):
"""Initialize AudioInput.
Args:
sample_rate: Audio sample rate in Hz
callback: Optional callback function to process audio data
buffer_duration: Duration of audio buffer in seconds
"""
self.sample_rate = sample_rate
self.is_recording = False
self._callback = callback
self._buffer = np.array([], dtype=np.float32)
self._buffer_size = int(sample_rate * buffer_duration)
def _append_to_buffer(self, audio_data: np.ndarray) -> None:
"""Append new audio data to the buffer."""
# Convert stereo to mono if necessary
if audio_data.ndim > 1:
audio_data = np.mean(audio_data, axis=1)
self._buffer = np.concatenate([self._buffer, audio_data])
def _process_buffer(self) -> None:
"""Process buffer data if it has reached the desired size."""
if len(self._buffer) >= self._buffer_size:
if self._callback is not None:
self._callback(self._buffer[: self._buffer_size])
self._buffer = self._buffer[self._buffer_size :]
@abstractmethod
def start_recording(self):
"""Start recording audio."""
pass
@abstractmethod
def stop_recording(self):
"""Stop recording audio."""
pass
|