diff --git a/src/pipecat/audio/filters/aic_filter_v2.py b/src/pipecat/audio/filters/aic_filter_v2.py new file mode 100644 index 000000000..be2912107 --- /dev/null +++ b/src/pipecat/audio/filters/aic_filter_v2.py @@ -0,0 +1,296 @@ +# +# Copyright (c) 2024-2026, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + +"""ai-coustics AIC SDK audio filter for Pipecat (aic-sdk >= 2.0.0). + +This module provides an audio filter implementation using ai-coustics' AIC SDK to +enhance audio streams in real time. It mirrors the structure of other filters like +the Koala filter and integrates with Pipecat's input transport pipeline. + +.. note:: + This module is compatible with aic-sdk versions >= 2.0.0. + For aic-sdk < 2.0.0, use :mod:`pipecat.audio.filters.aic_filter` instead. +""" + +import os +from typing import List, Optional + +import numpy as np +from loguru import logger + +from pipecat.audio.filters.base_audio_filter import BaseAudioFilter +from pipecat.audio.utils import check_aic_sdk_version +from pipecat.frames.frames import FilterControlFrame, FilterEnableFrame + +# Check aic-sdk is installed and version is compatible (>= 2.0.0) +check_aic_sdk_version("v2") + +# AIC SDK v2 (https://ai-coustics.github.io/aic-sdk-py/api/) +import aic +from aic import Model, ProcessorAsync, ProcessorConfig, ProcessorParameter, VadParameter + + +class AICFilter(BaseAudioFilter): + """Audio filter using ai-coustics' AIC SDK v2 for real-time enhancement. + + Buffers incoming audio to the model's preferred block size and processes + planar frames in-place using float32 samples in the linear -1..+1 range. + + .. note:: + This class requires aic-sdk >= 2.0.0. + """ + + def __init__( + self, + *, + license_key: str = "", + model_id: Optional[str] = None, + model_path: Optional[str] = None, + model_download_dir: Optional[str] = None, + enhancement_level: Optional[float] = 1.0, + voice_gain: Optional[float] = 1.0, + ) -> None: + """Initialize the AIC filter. + + Args: + license_key: ai-coustics license key for authentication. + model_id: Model identifier to download from CDN. Required if model_path + is not provided. See https://artifacts.ai-coustics.io/ for available models. + model_path: Optional path to a local .aicmodel file. If provided, + model_id is ignored and no download occurs. + model_download_dir: Directory for downloading models. Defaults to + a cache directory in user's home folder. + enhancement_level: Optional overall enhancement strength (0.0..1.0). + voice_gain: Optional linear gain applied to detected speech (0.1..4.0). + + Raises: + ValueError: If neither model_id nor model_path is provided. + """ + if model_id is None and model_path is None: + raise ValueError( + "Either 'model_id' or 'model_path' must be provided. " + "See https://artifacts.ai-coustics.io/ for available models." + ) + + self._license_key = license_key + self._model_id = model_id + self._model_path = model_path + self._model_download_dir = model_download_dir or os.path.expanduser( + "~/.cache/pipecat/aic-models" + ) + + self._enhancement_level = enhancement_level + self._voice_gain = voice_gain + + self._enabled = True + self._sample_rate = 0 + self._aic_ready = False + self._frames_per_block = 0 + self._audio_buffer = bytearray() + + # v2 API objects + self._model: Optional[Model] = None + self._processor: Optional[ProcessorAsync] = None + self._processor_ctx = None + self._vad_ctx = None + + def get_vad_context(self): + """Return the VAD context once the processor exists. + + Returns: + The VadContext instance bound to the underlying processor. + Raises RuntimeError if the processor has not been initialized. + """ + if self._vad_ctx is None: + raise RuntimeError("AIC processor not initialized yet. Call start(sample_rate) first.") + return self._vad_ctx + + def create_vad_analyzer( + self, + *, + speech_hold_duration: Optional[float] = None, + sensitivity: Optional[float] = None, + ): + """Return an analyzer that will lazily instantiate the AIC VAD when ready. + + AIC VAD parameters (v2): + - speech_hold_duration: + How long VAD continues detecting after speech ends (in seconds). + Range: 0.0 .. 20x model window length, Default (SDK): 0.05s + - sensitivity: + Energy threshold sensitivity. Energy threshold = 10 ** (-sensitivity). + Range: 1.0 .. 15.0, Default (SDK): 6.0 + + Args: + speech_hold_duration: Optional speech hold duration to configure on the VAD. + If None, SDK default (0.05s) is used. + sensitivity: Optional sensitivity (energy threshold) to configure on the VAD. + Range: 1.0 .. 15.0. If None, SDK default (6.0) is used. + + Returns: + A lazily-initialized AICVADAnalyzer that will bind to the VAD context + once the filter's processor has been created (after start(sample_rate)). + """ + from pipecat.audio.vad.aic_vad_v2 import AICVADAnalyzer + + return AICVADAnalyzer( + vad_context_factory=lambda: self.get_vad_context(), + speech_hold_duration=speech_hold_duration, + sensitivity=sensitivity, + ) + + async def start(self, sample_rate: int): + """Initialize the filter with the transport's sample rate. + + Args: + sample_rate: The sample rate of the input transport in Hz. + + Returns: + None + """ + self._sample_rate = sample_rate + + try: + # Load or download model + if self._model_path: + logger.debug(f"Loading AIC model from: {self._model_path}") + self._model = Model.from_file(self._model_path) + else: + logger.debug(f"Downloading AIC model: {self._model_id}") + os.makedirs(self._model_download_dir, exist_ok=True) + model_path = await Model.download_async(self._model_id, self._model_download_dir) + logger.debug(f"Model downloaded to: {model_path}") + self._model = Model.from_file(model_path) + + # Create async processor + self._processor = ProcessorAsync(self._model, self._license_key or "") + + # Get optimal frames for this sample rate + self._frames_per_block = self._model.get_optimal_num_frames(self._sample_rate) + + # Create configuration + config = ProcessorConfig( + sample_rate=self._sample_rate, + num_channels=1, + num_frames=self._frames_per_block, + allow_variable_frames=False, + ) + + # Initialize processor + await self._processor.initialize_async(config) + + # Get contexts for parameter control and VAD + self._processor_ctx = self._processor.get_processor_context() + self._vad_ctx = self._processor.get_vad_context() + + # Apply initial parameters + if self._enhancement_level is not None: + level = float(self._enhancement_level if self._enabled else 0.0) + self._processor_ctx.set_parameter(ProcessorParameter.EnhancementLevel, level) + if self._voice_gain is not None: + self._processor_ctx.set_parameter( + ProcessorParameter.VoiceGain, float(self._voice_gain) + ) + + self._aic_ready = True + + # Log processor information + logger.debug(f"ai-coustics filter (v2) started:") + logger.debug(f" Model ID: {self._model.get_id()}") + logger.debug(f" Sample rate: {self._sample_rate} Hz") + logger.debug(f" Frames per chunk: {self._frames_per_block}") + logger.debug(f" Enhancement strength: {int((self._enhancement_level or 1.0) * 100)}%") + logger.debug(f" Optimal sample rate: {self._model.get_optimal_sample_rate()} Hz") + logger.debug( + f" Output delay: {self._processor_ctx.get_output_delay()} samples " + f"({self._processor_ctx.get_output_delay() / self._sample_rate * 1000:.2f}ms)" + ) + except Exception as e: # noqa: BLE001 - surfacing SDK initialization errors + logger.error(f"AIC model initialization failed: {e}") + self._aic_ready = False + + async def stop(self): + """Clean up the AIC processor when stopping. + + Returns: + None + """ + try: + if self._processor_ctx is not None: + self._processor_ctx.reset() + finally: + self._processor = None + self._processor_ctx = None + self._vad_ctx = None + self._model = None + self._aic_ready = False + self._audio_buffer.clear() + + async def process_frame(self, frame: FilterControlFrame): + """Process control frames to enable/disable filtering. + + Args: + frame: The control frame containing filter commands. + + Returns: + None + """ + if isinstance(frame, FilterEnableFrame): + self._enabled = frame.enable + if self._processor_ctx is not None: + try: + level = float(self._enhancement_level if self._enabled else 0.0) + self._processor_ctx.set_parameter(ProcessorParameter.EnhancementLevel, level) + except Exception as e: # noqa: BLE001 + logger.error(f"AIC set_parameter failed: {e}") + + async def filter(self, audio: bytes) -> bytes: + """Apply AIC enhancement to audio data. + + Buffers incoming audio and processes it in chunks that match the AIC + model's required block length. Returns enhanced audio data. + + Args: + audio: Raw audio data as bytes to be filtered (int16 PCM, planar). + + Returns: + Enhanced audio data as bytes (int16 PCM, planar). + """ + if not self._aic_ready or self._processor is None: + return audio + + self._audio_buffer.extend(audio) + + filtered_chunks: List[bytes] = [] + + # Number of int16 samples currently buffered + available_frames = len(self._audio_buffer) // 2 + + while available_frames >= self._frames_per_block: + # Consume exactly one block worth of frames + samples_to_consume = self._frames_per_block * 1 + bytes_to_consume = samples_to_consume * 2 + block_bytes = bytes(self._audio_buffer[:bytes_to_consume]) + + # Convert to float32 in -1..+1 range and reshape to (channels, frames) + block_i16 = np.frombuffer(block_bytes, dtype=np.int16) + block_f32 = (block_i16.astype(np.float32) / 32768.0).reshape( + (1, self._frames_per_block) + ) + + # Process via async processor; returns ndarray (same shape) + out_f32 = await self._processor.process_async(block_f32) + + # Convert back to int16 bytes + out_i16 = np.clip(out_f32 * 32768.0, -32768, 32767).astype(np.int16) + filtered_chunks.append(out_i16.reshape(-1).tobytes()) + + # Slide buffer + self._audio_buffer = self._audio_buffer[bytes_to_consume:] + available_frames = len(self._audio_buffer) // 2 + + # Do not flush incomplete frames; keep them buffered for the next call + return b"".join(filtered_chunks) diff --git a/src/pipecat/audio/utils.py b/src/pipecat/audio/utils.py index 65f451675..002375f35 100644 --- a/src/pipecat/audio/utils.py +++ b/src/pipecat/audio/utils.py @@ -12,6 +12,9 @@ various audio formats used in Pipecat pipelines. """ import audioop +from typing import Literal + +from loguru import logger import numpy as np import pyloudnorm as pyln @@ -311,3 +314,57 @@ def is_silence(pcm_bytes: bytes) -> bool: # If max value is lower than SPEAKING_THRESHOLD, consider it as silence return max_value <= SPEAKING_THRESHOLD + + +def check_aic_sdk_version(required_version: Literal["v1", "v2"]) -> None: + """Check if the aic-sdk is installed and compatible with the module. + + This function checks both that the aic-sdk is installed and that its version + is compatible with the module requirements. + + Args: + required_version: Either "v1" (for aic-sdk < 2.0.0) or "v2" (for aic-sdk >= 2.0.0). + + Raises: + ImportError: If aic-sdk is not installed or version is incompatible. + """ + try: + import aic # noqa: F401 - check if module is installed + except ModuleNotFoundError as e: + logger.error(f"Exception: {e}") + logger.error("In order to use the AIC filter, you need to `pip install pipecat-ai[aic]`.") + raise ImportError(f"Missing module: {e}") from e + + try: + from importlib.metadata import version as get_version + + aic_version = get_version("aic-sdk") + major_version = int(aic_version.split(".")[0]) + + if required_version == "v1" and major_version >= 2: + error_msg = ( + f"aic-sdk version {aic_version} detected, but aic-sdk < 2.0.0 is required. " + "Please use the v2 modules instead: " + "'from pipecat.audio.filters.aic_filter_v2 import AICFilter' or " + "'from pipecat.audio.vad.aic_vad_v2 import AICVADAnalyzer'." + ) + logger.error(error_msg) + raise ImportError(error_msg) + + if required_version == "v2" and major_version < 2: + error_msg = ( + f"aic-sdk version {aic_version} detected, but aic-sdk >= 2.0.0 is required. " + "Please update with 'pip install --upgrade aic-sdk>=2.0.0' " + "or use the v1 modules: " + "'from pipecat.audio.filters.aic_filter import AICFilter' or " + "'from pipecat.audio.vad.aic_vad import AICVADAnalyzer'." + ) + logger.error(error_msg) + raise ImportError(error_msg) + + except ImportError: + # Re-raise if it's our version mismatch error + raise + except Exception: + # If we can't determine version for other reasons, log warning and allow to proceed + logger.warning("Could not determine aic-sdk version. Proceeding anyway.") diff --git a/src/pipecat/audio/vad/aic_vad_v2.py b/src/pipecat/audio/vad/aic_vad_v2.py new file mode 100644 index 000000000..2f8129a0d --- /dev/null +++ b/src/pipecat/audio/vad/aic_vad_v2.py @@ -0,0 +1,163 @@ +"""AIC-integrated VAD analyzer that lazily binds to the AIC SDK v2 backend. + +This analyzer queries the backend's is_speech_detected() and maps it to a float +confidence (1.0/0.0). It uses 10 ms windows based on the sample rate and applies +optional AIC VAD parameters (speech_hold_duration, sensitivity) when available. + +.. note:: + This module is compatible with aic-sdk versions >= 2.0.0. + For aic-sdk < 2.0.0, use :mod:`pipecat.audio.vad.aic_vad` instead. +""" + +from typing import Any, Callable, Optional + +from loguru import logger + +from pipecat.audio.utils import check_aic_sdk_version +from pipecat.audio.vad.vad_analyzer import VADAnalyzer, VADParams + +# Check aic-sdk is installed and version is compatible (>= 2.0.0) +check_aic_sdk_version("v2") + +from aic import VadParameter + + + +class AICVADAnalyzer(VADAnalyzer): + """VAD analyzer that lazily binds to the AIC VadContext via a factory. + + The analyzer can be constructed before the AIC Processor exists. Once the filter has + started and the Processor is available, the provided factory will succeed and the + VadContext will be obtained. We then use the context's is_speech_detected() state + to derive confidence values. + + AIC VAD runtime parameters (v2): + - speech_hold_duration: + Controls for how long the VAD continues to detect speech after the audio signal + no longer contains speech (in seconds). + Range: 0.0 .. 20x model window length + Default (SDK): 0.05s (50ms) + - sensitivity: + Controls the energy threshold sensitivity. Higher values make the detector + less sensitive (require more energy to count as speech). + Range: 1.0 .. 15.0 + Formula: Energy threshold = 10 ** (-sensitivity) + Default (SDK): 6.0 + + .. note:: + This class requires aic-sdk >= 2.0.0. + """ + + def __init__( + self, + *, + vad_context_factory: Optional[Callable[[], Any]] = None, + speech_hold_duration: Optional[float] = None, + sensitivity: Optional[float] = None, + ): + """Create an AIC VAD analyzer. + + Args: + vad_context_factory: + Zero-arg callable that returns the AIC VadContext. + This may raise until the filter's Processor has been created; the analyzer + will retry on set_sample_rate/first use. + speech_hold_duration: + Optional override for AIC VAD speech hold duration (in seconds). + Range: 0.0 .. 20x model window length. + If None, the SDK default (0.05s) is used. + sensitivity: + Optional override for AIC VAD sensitivity (energy threshold). + Range: 1.0 .. 15.0. Energy threshold = 10 ** (-sensitivity). + If None, the SDK default (6.0) is used. + """ + # Use fixed VAD parameters for AIC: no user override + fixed_params = VADParams(confidence=0.5, start_secs=0.0, stop_secs=0.0, min_volume=0.0) + super().__init__(sample_rate=None, params=fixed_params) + self._vad_context_factory = vad_context_factory + self._vad_ctx: Optional[Any] = None + self._pending_speech_hold_duration: Optional[float] = speech_hold_duration + self._pending_sensitivity: Optional[float] = sensitivity + + def bind_vad_context_factory(self, vad_context_factory: Callable[[], Any]): + """Attach or replace the factory post-construction.""" + self._vad_context_factory = vad_context_factory + self._ensure_vad_context_initialized() + + def _apply_vad_params(self): + """Apply optional AIC VAD parameters if available.""" + if self._vad_ctx is None or VadParameter is None: + return + try: + if self._pending_speech_hold_duration is not None: + self._vad_ctx.set_parameter( + VadParameter.SpeechHoldDuration, float(self._pending_speech_hold_duration) + ) + if self._pending_sensitivity is not None: + self._vad_ctx.set_parameter( + VadParameter.Sensitivity, float(self._pending_sensitivity) + ) + except Exception as e: # noqa: BLE001 + logger.debug(f"AIC VAD parameter application deferred/failed: {e}") + + def _ensure_vad_context_initialized(self): + if self._vad_ctx is not None: + return + if not self._vad_context_factory: + return + try: + self._vad_ctx = self._vad_context_factory() + self._apply_vad_params() + # With VAD context ready, recompute internal frame sizing + super().set_params(self._params) + logger.debug("AIC VAD context (v2) initialized in analyzer.") + except Exception as e: # noqa: BLE001 + # Filter may not be started yet; try again later + logger.debug(f"Deferring AIC VAD context initialization: {e}") + + def set_sample_rate(self, sample_rate: int): + """Set the sample rate for audio processing. + + Args: + sample_rate: Audio sample rate in Hz. + """ + # Set rate and attempt VAD context initialization once we know SR + self._sample_rate = self._init_sample_rate or sample_rate + self._ensure_vad_context_initialized() + # Ensure params are initialized even if VAD context not ready yet + try: + super().set_params(self._params) + except Exception: + pass + + def num_frames_required(self) -> int: + """Get the number of audio frames required for analysis. + + Returns: + Number of frames needed for VAD processing. + """ + # Use 10 ms windows based on sample rate + return int(self.sample_rate * 0.01) if self.sample_rate > 0 else 160 + + def voice_confidence(self, buffer: bytes) -> float: + """Calculate voice activity confidence for the given audio buffer. + + Args: + buffer: Audio buffer to analyze. + + Returns: + Voice confidence score is 0.0 or 1.0. + """ + # Ensure VAD context exists (filter might have started since last call) + self._ensure_vad_context_initialized() + if self._vad_ctx is None: + return 0.0 + + # We do not need to analyze 'buffer' here since the processor's VAD is updated + # as part of the enhancement pipeline. Simply query the boolean and map it. + try: + is_speech = self._vad_ctx.is_speech_detected() + return 1.0 if is_speech else 0.0 + except Exception as e: # noqa: BLE001 + logger.error(f"AIC VAD inference error: {e}") + return 0.0