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 # Preprocessed audio (float32, normalized) original_audio: np.ndarray # Original int16/float32 audio vad_result: VADResult timestamp_ms: int chunk_index: int window_id: int = 0 # Monotonic id of the model input window 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) # Linear interpolation 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 # 1st order high-pass: y[n] = alpha * (y[n-1] + x[n] - x[n-1]) # alpha = exp(-2*pi*fc/fs) 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] 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() # 1. Convert to float32 audio = self._int16_to_float32(audio) # 2. Resample if needed audio = self._resample(audio, orig_sr) # 3. High-pass filter audio = self._highpass_filter(audio) # 4. VAD vad_result = self.vad.process(audio) # 5. Normalize if self.normalize: audio = self._normalize_audio(audio, method="peak") # 6. Ensure full model window size 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() # Convert and resample audio = self._int16_to_float32(audio) audio = self._resample(audio, orig_sr) # Write to ring buffer buffer.write(audio) # Check if we have a full processing window if not buffer.is_ready_for_inference(): return None, False # Get the latest full model window / rolling step window = buffer.get_processing_window() if window is None: return None, False self._window_id += 1 window_id = self._window_id # High-pass filter on window window = self._highpass_filter(window) # VAD on window vad_result = self.vad.process(window) # Normalize if self.normalize: window = self._normalize_audio(window, method="peak") # Ensure exact model window size 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