| import numpy as np |
| from typing import Optional, Tuple |
| from dataclasses import dataclass |
|
|
| from ..config import settings, BUFFER_SAMPLES, WINDOW_SAMPLES |
| from .vad import VoiceActivityDetector, VADResult |
| from .buffer import RingBuffer |
|
|
|
|
| @dataclass |
| class ProcessedChunk: |
| audio: np.ndarray |
| original_audio: np.ndarray |
| vad_result: VADResult |
| timestamp_ms: int |
| chunk_index: int |
| window_id: int = 0 |
|
|
|
|
| class AudioProcessor: |
| """ |
| Audio preprocessing pipeline: |
| 1. Convert int16 to float32 normalized [-1, 1] |
| 2. Resample to 16kHz if needed |
| 3. Voice Activity Detection |
| 4. Normalization (peak/RMS) |
| 5. Optional: High-pass filter for DC removal |
| """ |
|
|
| def __init__( |
| self, |
| target_sr: int = settings.sample_rate, |
| normalize: bool = True, |
| highpass_cutoff: float = 80.0, |
| vad: Optional[VoiceActivityDetector] = None, |
| ): |
| self.target_sr = target_sr |
| self.normalize = normalize |
| self.highpass_cutoff = highpass_cutoff |
| self.vad = vad or VoiceActivityDetector() |
| self._chunk_index = 0 |
| self._window_id = 0 |
| self._highpass_state = None |
|
|
| def _int16_to_float32(self, audio: np.ndarray) -> np.ndarray: |
| """Convert int16 PCM to float32 [-1, 1].""" |
| if audio.dtype == np.int16: |
| return audio.astype(np.float32) / 32768.0 |
| elif audio.dtype == np.float32: |
| return audio |
| elif audio.dtype == np.float64: |
| return audio.astype(np.float32) |
| else: |
| return audio.astype(np.float32) / np.iinfo(audio.dtype).max |
|
|
| def _resample(self, audio: np.ndarray, orig_sr: int) -> np.ndarray: |
| """Resample audio to target sample rate using linear interpolation.""" |
| if orig_sr == self.target_sr: |
| return audio |
|
|
| n_out = int(len(audio) * self.target_sr / orig_sr) |
| if n_out == 0: |
| return np.zeros(0, dtype=audio.dtype) |
|
|
| |
| x_old = np.linspace(0.0, 1.0, len(audio), endpoint=False) |
| x_new = np.linspace(0.0, 1.0, n_out, endpoint=False) |
| resampled = np.interp(x_new, x_old, audio) |
| return resampled.astype(np.float32, copy=False) |
|
|
| def _highpass_filter(self, audio: np.ndarray) -> np.ndarray: |
| """Apply high-pass filter for DC removal (1st order IIR).""" |
| if self.highpass_cutoff <= 0: |
| return audio |
|
|
| |
| |
| alpha = np.exp(-2 * np.pi * self.highpass_cutoff / self.target_sr) |
|
|
| if self._highpass_state is None: |
| self._highpass_state = np.zeros(2, dtype=np.float32) |
|
|
| y_prev, x_prev = self._highpass_state |
| output = np.empty_like(audio) |
|
|
| for i, x in enumerate(audio): |
| y = alpha * (y_prev + x - x_prev) |
| output[i] = y |
| y_prev, x_prev = y, x |
|
|
| self._highpass_state = np.array([y_prev, x_prev], dtype=np.float32) |
| return output |
|
|
| def _normalize_audio(self, audio: np.ndarray, method: str = "peak") -> np.ndarray: |
| """Normalize audio to prevent clipping and ensure consistent levels.""" |
| if method == "peak": |
| peak = np.max(np.abs(audio)) |
| if peak > 0: |
| return audio / peak * 0.95 |
| elif method == "rms": |
| rms = np.sqrt(np.mean(audio ** 2)) |
| if rms > 0: |
| target_rms = 0.1 |
| return audio * (target_rms / rms) |
| return audio |
|
|
| def _pad_to_window(self, audio: np.ndarray, n: int = WINDOW_SAMPLES) -> np.ndarray: |
| """Pad or crop to the model input window length.""" |
| if len(audio) > n: |
| return audio[:n] |
| if len(audio) < n: |
| return np.pad(audio, (0, n - len(audio))) |
| return audio |
|
|
| def process_chunk( |
| self, |
| audio: np.ndarray, |
| sample_rate: int = None, |
| timestamp_ms: int = None, |
| ) -> ProcessedChunk: |
| """ |
| Process a single audio chunk through the full pipeline |
| (one-shot / non-buffered path). |
| """ |
| import time |
| if timestamp_ms is None: |
| timestamp_ms = int(time.time() * 1000) |
|
|
| self._window_id += 1 |
| window_id = self._window_id |
|
|
| orig_sr = sample_rate or self.target_sr |
| original = audio.copy() |
|
|
| |
| audio = self._int16_to_float32(audio) |
|
|
| |
| audio = self._resample(audio, orig_sr) |
|
|
| |
| audio = self._highpass_filter(audio) |
|
|
| |
| vad_result = self.vad.process(audio) |
|
|
| |
| if self.normalize: |
| audio = self._normalize_audio(audio, method="peak") |
|
|
| |
| audio = self._pad_to_window(audio) |
|
|
| chunk = ProcessedChunk( |
| audio=audio.astype(np.float32), |
| original_audio=original, |
| vad_result=vad_result, |
| timestamp_ms=timestamp_ms, |
| chunk_index=self._chunk_index, |
| window_id=window_id, |
| ) |
|
|
| self._chunk_index += 1 |
| return chunk |
|
|
| def process_stream_chunk( |
| self, |
| audio: np.ndarray, |
| buffer: RingBuffer, |
| sample_rate: int = None, |
| ) -> Tuple[Optional[ProcessedChunk], bool]: |
| """ |
| Process streaming audio: write to buffer, extract the latest full model |
| window if ready. Returns (ProcessedChunk, window_ready). |
| """ |
| import time |
| timestamp_ms = int(time.time() * 1000) |
|
|
| orig_sr = sample_rate or self.target_sr |
| original = audio.copy() |
|
|
| |
| audio = self._int16_to_float32(audio) |
| audio = self._resample(audio, orig_sr) |
|
|
| |
| buffer.write(audio) |
|
|
| |
| if not buffer.is_ready_for_inference(): |
| return None, False |
|
|
| |
| window = buffer.get_processing_window() |
| if window is None: |
| return None, False |
|
|
| self._window_id += 1 |
| window_id = self._window_id |
|
|
| |
| window = self._highpass_filter(window) |
|
|
| |
| vad_result = self.vad.process(window) |
|
|
| |
| if self.normalize: |
| window = self._normalize_audio(window, method="peak") |
|
|
| |
| window = self._pad_to_window(window) |
|
|
| chunk = ProcessedChunk( |
| audio=window.astype(np.float32), |
| original_audio=original, |
| vad_result=vad_result, |
| timestamp_ms=timestamp_ms, |
| chunk_index=self._chunk_index, |
| window_id=window_id, |
| ) |
|
|
| self._chunk_index += 1 |
| return chunk, True |
|
|
| def reset(self) -> None: |
| """Reset processor state for new stream.""" |
| self.vad.reset() |
| self._highpass_state = None |
| self._chunk_index = 0 |
| self._window_id = 0 |
