diff --git a/changelog/3408.added.md b/changelog/3408.added.md new file mode 100644 index 000000000..04f1311cc --- /dev/null +++ b/changelog/3408.added.md @@ -0,0 +1,4 @@ +- Additions for `AICFilter` and `AICVADAnalyzer`: + - Added model downloading support to `AICFilter` with `model_id` and `model_download_dir` parameters. + - Added `model_path` parameter to `AICFilter` for loading local `.aicmodel` files. + - Added unit tests for `AICFilter` and `AICVADAnalyzer`. diff --git a/changelog/3408.changed.md b/changelog/3408.changed.md new file mode 100644 index 000000000..9436b6074 --- /dev/null +++ b/changelog/3408.changed.md @@ -0,0 +1 @@ +- Updated `AICFilter` and `AICVADAnalyzer` to use aic-sdk ~= 2.0.1. diff --git a/changelog/3408.removed.md b/changelog/3408.removed.md new file mode 100644 index 000000000..f578bf5d0 --- /dev/null +++ b/changelog/3408.removed.md @@ -0,0 +1 @@ +- Removed deprecated `AICFilter` parameters: `enhancement_level`, `voice_gain`, `noise_gate_enable`. diff --git a/examples/foundational/07zd-interruptible-aicoustics.py b/examples/foundational/07zd-interruptible-aicoustics.py index 8a49e734f..ca205fc1d 100644 --- a/examples/foundational/07zd-interruptible-aicoustics.py +++ b/examples/foundational/07zd-interruptible-aicoustics.py @@ -50,7 +50,7 @@ def _create_aic_filter() -> AICFilter: return AICFilter( license_key=license_key, - enhancement_level=0.5, + model_id="quail-vf-l-16khz", ) @@ -62,7 +62,9 @@ transport_params = { lambda aic: DailyParams( audio_in_enabled=True, audio_out_enabled=True, - vad_analyzer=aic.create_vad_analyzer(lookback_buffer_size=6.0, sensitivity=6.0), + vad_analyzer=aic.create_vad_analyzer( + speech_hold_duration=0.05, minimum_speech_duration=0.0, sensitivity=6.0 + ), audio_in_filter=aic, ) )(_create_aic_filter()), @@ -70,7 +72,9 @@ transport_params = { lambda aic: FastAPIWebsocketParams( audio_in_enabled=True, audio_out_enabled=True, - vad_analyzer=aic.create_vad_analyzer(lookback_buffer_size=6.0, sensitivity=6.0), + vad_analyzer=aic.create_vad_analyzer( + speech_hold_duration=0.05, minimum_speech_duration=0.0, sensitivity=6.0 + ), audio_in_filter=aic, ) )(_create_aic_filter()), @@ -78,7 +82,9 @@ transport_params = { lambda aic: TransportParams( audio_in_enabled=True, audio_out_enabled=True, - vad_analyzer=aic.create_vad_analyzer(lookback_buffer_size=6.0, sensitivity=6.0), + vad_analyzer=aic.create_vad_analyzer( + speech_hold_duration=0.05, minimum_speech_duration=0.0, sensitivity=6.0 + ), audio_in_filter=aic, ) )(_create_aic_filter()), diff --git a/pyproject.toml b/pyproject.toml index 884b5afd1..339b498f6 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -48,7 +48,7 @@ Issues = "https://github.com/pipecat-ai/pipecat/issues" Changelog = "https://github.com/pipecat-ai/pipecat/blob/main/CHANGELOG.md" [project.optional-dependencies] -aic = [ "aic-sdk~=1.2.0" ] +aic = [ "aic-sdk~=2.0.1" ] anthropic = [ "anthropic~=0.49.0" ] assemblyai = [ "pipecat-ai[websockets-base]" ] asyncai = [ "pipecat-ai[websockets-base]" ] diff --git a/scripts/evals/run-release-evals.py b/scripts/evals/run-release-evals.py index e544c732e..d4290be00 100644 --- a/scripts/evals/run-release-evals.py +++ b/scripts/evals/run-release-evals.py @@ -133,8 +133,7 @@ TESTS_07 = [ ("07zb-interruptible-inworld-http.py", EVAL_SIMPLE_MATH), ("07zc-interruptible-asyncai.py", EVAL_SIMPLE_MATH), ("07zc-interruptible-asyncai-http.py", EVAL_SIMPLE_MATH), - # Need license key to run - # ("07zd-interruptible-aicoustics.py", EVAL_SIMPLE_MATH), + ("07zd-interruptible-aicoustics.py", EVAL_SIMPLE_MATH), ("07ze-interruptible-hume.py", EVAL_SIMPLE_MATH), ("07zf-interruptible-gradium.py", EVAL_SIMPLE_MATH), ("07zg-interruptible-camb.py", EVAL_SIMPLE_MATH), diff --git a/src/pipecat/audio/filters/aic_filter.py b/src/pipecat/audio/filters/aic_filter.py index e4242521b..7f0626776 100644 --- a/src/pipecat/audio/filters/aic_filter.py +++ b/src/pipecat/audio/filters/aic_filter.py @@ -9,129 +9,145 @@ 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. + +Classes: + AICFilter: For aic-sdk (uses 'aic_sdk' module) """ +from pathlib import Path from typing import List, Optional import numpy as np +from aic_sdk import ( + Model, + ParameterFixedError, + ProcessorAsync, + ProcessorConfig, + ProcessorParameter, + set_sdk_id, +) from loguru import logger from pipecat.audio.filters.base_audio_filter import BaseAudioFilter +from pipecat.audio.vad.aic_vad import AICVADAnalyzer from pipecat.frames.frames import FilterControlFrame, FilterEnableFrame -try: - # AIC SDK (https://ai-coustics.github.io/aic-sdk-py/api/) - from aic import AICModelType, AICParameter, Model -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 Exception(f"Missing module: {e}") - class AICFilter(BaseAudioFilter): """Audio filter using ai-coustics' AIC SDK 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. + frames using float32 samples normalized to the range -1 to +1. """ def __init__( self, *, - license_key: str = "", - model_type: AICModelType = AICModelType.QUAIL_STT, - enhancement_level: Optional[float] = 1.0, - voice_gain: Optional[float] = 1.0, - noise_gate_enable: Optional[bool] = True, + license_key: str, + model_id: Optional[str] = None, + model_path: Optional[Path] = None, + model_download_dir: Optional[Path] = None, ) -> None: """Initialize the AIC filter. Args: license_key: ai-coustics license key for authentication. - model_type: Model variant to load. - enhancement_level: Optional overall enhancement strength (0.0..1.0). - voice_gain: Optional linear gain applied to detected speech (0.0..4.0). - noise_gate_enable: Optional enable/disable noise gate (default: True). + 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 as a Path object. + Defaults to a cache directory in user's home folder. - .. deprecated:: 1.3.0 - The `noise_gate_enable` parameter is deprecated and no longer has any effect. - It will be removed in a future version. + Raises: + ValueError: If neither model_id nor model_path is provided. """ + # Set SDK ID for telemetry identification (6 = pipecat) + set_sdk_id(6) + + 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_type = model_type + self._model_id = model_id + self._model_path = model_path + self._model_download_dir = model_download_dir or ( + Path.home() / ".cache" / "pipecat" / "aic-models" + ) - self._enhancement_level = enhancement_level - self._voice_gain = voice_gain - if noise_gate_enable is not None: - import warnings - - with warnings.catch_warnings(): - warnings.simplefilter("always") - warnings.warn( - "Parameter `noise_gate_enable` is deprecated and no longer has any effect. " - "It will be removed in a future version. Use AIC VAD instead (create_vad_analyzer()).", - DeprecationWarning, - ) - - self._noise_gate_enable = noise_gate_enable - - self._enabled = True + self._bypass = False self._sample_rate = 0 self._aic_ready = False self._frames_per_block = 0 self._audio_buffer = bytearray() - # Model will be created in start() since the API now requires sample_rate - self._aic = None - def get_vad_factory(self): - """Return a zero-arg factory that will create the VAD once the model exists. + # Audio format constants + self._bytes_per_sample = 2 # int16 = 2 bytes + self._dtype = np.int16 + self._scale = ( + 32768.0 # 2^15, for normalizing int16 (-32768 to 32767) to float32 (-1.0 to 1.0) + ) + + # AIC SDK objects + self._model = None + self._processor = None + self._processor_ctx = None + self._vad_ctx = None + + # Pre-allocated buffers (resized in start() once frames_per_block is known) + self._in_f32 = None + self._out_i16 = None + + def get_vad_context(self): + """Return the VAD context once the processor exists. Returns: - A zero-argument callable that, when invoked, returns an initialized - VoiceActivityDetector bound to the underlying AIC model. Raises a - RuntimeError if the model has not been initialized (i.e. start() - has not been called successfully). + The VadContext instance bound to the underlying processor. + Raises RuntimeError if the processor has not been initialized. """ - - def _factory(): - if self._aic is None: - raise RuntimeError("AIC model not initialized yet. Call start(sample_rate) first.") - return self._aic.create_vad() - - return _factory + 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, *, - lookback_buffer_size: Optional[float] = None, + speech_hold_duration: Optional[float] = None, + minimum_speech_duration: Optional[float] = None, sensitivity: Optional[float] = None, ): """Return an analyzer that will lazily instantiate the AIC VAD when ready. AIC VAD parameters: - - lookback_buffer_size: - Number of window-length audio buffers used as a lookback buffer. - Higher values increase prediction stability but add latency. - Range: 1.0 .. 20.0, Default (SDK): 6.0 + - speech_hold_duration: + How long VAD continues detecting after speech ends (in seconds). + Range: 0.0 to 100x model window length, Default (SDK): 0.05s + - minimum_speech_duration: + Minimum duration of speech required before VAD reports speech detected + (in seconds). Range: 0.0 to 1.0, Default (SDK): 0.0s - sensitivity: Energy threshold sensitivity. Energy threshold = 10 ** (-sensitivity). - Range: 1.0 .. 15.0, Default (SDK): 6.0 + Range: 1.0 to 15.0, Default (SDK): 6.0 Args: - lookback_buffer_size: Optional lookback buffer size to configure on the VAD. - Range: 1.0 .. 20.0. If None, SDK default is used. + speech_hold_duration: Optional speech hold duration to configure on the VAD. + If None, SDK default (0.05s) is used. + minimum_speech_duration: Optional minimum speech duration before VAD reports + speech detected. If None, SDK default (0.0s) is used. sensitivity: Optional sensitivity (energy threshold) to configure on the VAD. - Range: 1.0 .. 15.0. If None, SDK default is used. + Range: 1.0 to 15.0. If None, SDK default (6.0) is used. Returns: - A lazily-initialized AICVADAnalyzer that will bind to the VAD backend - once the filter's model has been created (after start(sample_rate)). + 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 import AICVADAnalyzer - return AICVADAnalyzer( - vad_factory=self.get_vad_factory(), - lookback_buffer_size=lookback_buffer_size, + vad_context_factory=lambda: self.get_vad_context(), + speech_hold_duration=speech_hold_duration, + minimum_speech_duration=minimum_speech_duration, sensitivity=sensitivity, ) @@ -146,52 +162,83 @@ class AICFilter(BaseAudioFilter): """ self._sample_rate = sample_rate + # Load or download model + if self._model_path: + logger.debug(f"Loading AIC model from: {self._model_path}") + self._model = Model.from_file(str(self._model_path)) + else: + logger.debug(f"Downloading AIC model: {self._model_id}") + self._model_download_dir.mkdir(parents=True, exist_ok=True) + model_path = await Model.download_async(self._model_id, str(self._model_download_dir)) + logger.debug(f"Model downloaded to: {model_path}") + self._model = Model.from_file(model_path) + + # Get optimal frames for this sample rate + self._frames_per_block = self._model.get_optimal_num_frames(self._sample_rate) + + # Allocate processing buffers now that we know the block size + self._in_f32 = np.zeros((1, self._frames_per_block), dtype=np.float32) + self._out_i16 = np.zeros(self._frames_per_block, dtype=np.int16) + + # Create configuration + config = ProcessorConfig.optimal( + self._model, + sample_rate=self._sample_rate, + ) + + # Create async processor try: - # Create model with required runtime parameters - self._aic = Model( - model_type=self._model_type, - license_key=self._license_key or None, - sample_rate=self._sample_rate, - channels=1, - ) - self._frames_per_block = self._aic.optimal_num_frames() - - # Optional parameter configuration - if self._enhancement_level is not None: - self._aic.set_parameter( - AICParameter.ENHANCEMENT_LEVEL, - float(self._enhancement_level if self._enabled else 0.0), - ) - if self._voice_gain is not None: - self._aic.set_parameter(AICParameter.VOICE_GAIN, float(self._voice_gain)) - - self._aic_ready = True - - # Log processor information - logger.debug(f"ai-coustics filter started:") - 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 * 100)}%") - logger.debug(f" Optimal input buffer size: {self._aic.optimal_num_frames()} samples") - logger.debug(f" Optimal sample rate: {self._aic.optimal_sample_rate()} Hz") - logger.debug( - f" Current algorithmic latency: {self._aic.processing_latency() / self._sample_rate * 1000:.2f}ms" - ) + self._processor = ProcessorAsync(self._model, self._license_key, config) except Exception as e: # noqa: BLE001 - surfacing SDK initialization errors logger.error(f"AIC model initialization failed: {e}") - self._aic_ready = False + self._processor = None + + self._aic_ready = self._processor is not None + + if not self._aic_ready: + logger.debug(f"ai-coustics filter is not ready.") + return + + # 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 + try: + self._processor_ctx.set_parameter( + ProcessorParameter.Bypass, 1.0 if self._bypass else 0.0 + ) + except ParameterFixedError as e: + logger.error(f"AIC parameter update failed: {e}") + + # Log processor information + logger.debug(f"ai-coustics filter 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" Optimal sample rate: {self._model.get_optimal_sample_rate()} Hz") + logger.debug( + f" Optimal number of frames for {self._sample_rate} Hz: {self._model.get_optimal_num_frames(self._sample_rate)}" + ) + 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)" + ) async def stop(self): - """Clean up the AIC model when stopping. + """Clean up the AIC processor when stopping. Returns: None """ try: - if self._aic is not None: - self._aic.close() + if self._processor_ctx is not None: + self._processor_ctx.reset() finally: - self._aic = None + self._processor = None + self._processor_ctx = None + self._vad_ctx = None + self._model = None self._aic_ready = False self._audio_buffer.clear() @@ -205,11 +252,12 @@ class AICFilter(BaseAudioFilter): None """ if isinstance(frame, FilterEnableFrame): - self._enabled = frame.enable - if self._aic is not None: + self._bypass = not frame.enable + if self._processor_ctx is not None: try: - level = float(self._enhancement_level if self._enabled else 0.0) - self._aic.set_parameter(AICParameter.ENHANCEMENT_LEVEL, level) + self._processor_ctx.set_parameter( + ProcessorParameter.Bypass, 1.0 if self._bypass else 0.0 + ) except Exception as e: # noqa: BLE001 logger.error(f"AIC set_parameter failed: {e}") @@ -220,43 +268,41 @@ class AICFilter(BaseAudioFilter): model's required block length. Returns enhanced audio data. Args: - audio: Raw audio data as bytes to be filtered (int16 PCM, planar). + audio: Raw audio data as bytes (int16 PCM). Returns: - Enhanced audio data as bytes (int16 PCM, planar). + Enhanced audio data as bytes (int16 PCM). """ - if not self._aic_ready or self._aic is None: + if not self._aic_ready or self._processor is None: return audio self._audio_buffer.extend(audio) + available_frames = len(self._audio_buffer) // self._bytes_per_sample + num_blocks = available_frames // self._frames_per_block + + if num_blocks == 0: + return b"" filtered_chunks: List[bytes] = [] + mv = memoryview(self._audio_buffer) + block_size = self._frames_per_block * self._bytes_per_sample - # Number of int16 samples currently buffered - available_frames = len(self._audio_buffer) // 2 + for i in range(num_blocks): + start = i * block_size + block_i16 = np.frombuffer(mv[start : start + block_size], dtype=self._dtype) - 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]) + # Reuse input buffer, in-place divide + np.copyto(self._in_f32[0], block_i16) + self._in_f32 /= self._scale - # Convert to float32 in -1..+1 range and reshape to planar (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) - ) + out_f32 = await self._processor.process_async(self._in_f32) - # Process planar in-place; returns ndarray (same shape) - out_f32 = await self._aic.process_async(block_f32) + # Convert float32 output back to int16 + np.multiply(out_f32, self._scale, out=self._in_f32) # reuse in_f32 as temp + np.clip(self._in_f32, -self._scale, self._scale - 1, out=self._in_f32) + np.copyto(self._out_i16, self._in_f32[0].astype(self._dtype)) - # Convert back to int16 bytes, planar layout - out_i16 = np.clip(out_f32 * 32768.0, -32768, 32767).astype(np.int16) - filtered_chunks.append(out_i16.reshape(-1).tobytes()) + filtered_chunks.append(self._out_i16.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 + self._audio_buffer = self._audio_buffer[num_blocks * block_size :] return b"".join(filtered_chunks) diff --git a/src/pipecat/audio/vad/aic_vad.py b/src/pipecat/audio/vad/aic_vad.py index 4907e4f55..813029e2b 100644 --- a/src/pipecat/audio/vad/aic_vad.py +++ b/src/pipecat/audio/vad/aic_vad.py @@ -1,44 +1,44 @@ """AIC-integrated VAD analyzer that lazily binds to the AIC SDK 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 (lookback_buffer_size, sensitivity) when available. +This module provides VAD analyzer implementations that query the AIC SDK's +is_speech_detected() and map it to a float confidence (1.0/0.0). + +Classes: + AICVADAnalyzer: For aic-sdk (uses 'aic_sdk' module) """ from typing import Any, Callable, Optional +from aic_sdk import VadParameter from loguru import logger from pipecat.audio.vad.vad_analyzer import VADAnalyzer, VADParams -try: - from aic import AICVadParameter -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 Exception(f"Missing module: {e}") - class AICVADAnalyzer(VADAnalyzer): - """VAD analyzer that lazily instantiates the AIC VoiceActivityDetector via a factory. + """VAD analyzer that lazily binds to the AIC VadContext via a factory. - The analyzer can be constructed before the AIC Model exists. Once the filter has - started and the Model is available, the provided factory will succeed and the - backend VAD will be created. We then switch to single-sample updates where - num_frames_required() returns 1 and confidence is derived from the backend's - boolean is_speech_detected() state. + 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. The context's is_speech_detected() boolean state is + then mapped to 1.0 (speech) or 0.0 (no speech) to satisfy the VADAnalyzer interface. AIC VAD runtime parameters: - - lookback_buffer_size: - Controls the lookback buffer size used by the VAD, i.e. the number of - window-length audio buffers used as a lookback buffer. Larger values improve - stability but increase latency. - Range: 1.0 .. 20.0 - Default (SDK): 6.0 + - 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 to 100x model window length + Default (SDK): 0.05s (50ms) + - minimum_speech_duration: + Controls for how long speech needs to be present in the audio signal before the + VAD considers it speech (in seconds). + Range: 0.0 to 1.0 + Default (SDK): 0.0s - 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 + Controls the sensitivity (energy threshold) of the VAD. This value is used by + the VAD as the threshold a speech audio signal's energy has to exceed in order + to be considered speech. + Range: 1.0 to 15.0 Formula: Energy threshold = 10 ** (-sensitivity) Default (SDK): 6.0 """ @@ -46,69 +46,80 @@ class AICVADAnalyzer(VADAnalyzer): def __init__( self, *, - vad_factory: Optional[Callable[[], Any]] = None, - lookback_buffer_size: Optional[float] = None, + vad_context_factory: Optional[Callable[[], Any]] = None, + speech_hold_duration: Optional[float] = None, + minimum_speech_duration: Optional[float] = None, sensitivity: Optional[float] = None, ): """Create an AIC VAD analyzer. Args: - vad_factory: - Zero-arg callable that returns an initialized AIC VoiceActivityDetector. - This may raise until the filter's Model has been created; the analyzer + 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. - lookback_buffer_size: - Optional override for AIC VAD lookback buffer size. - Range: 1.0 .. 20.0. Larger values increase stability at the cost of latency. - If None, the SDK default (6.0) is used. + speech_hold_duration: + Optional override for AIC VAD speech hold duration (in seconds). + Range: 0.0 to 100x model window length. + If None, the SDK default (0.05s) is used. + minimum_speech_duration: + Optional override for minimum speech duration before VAD reports + speech detected (in seconds). + Range: 0.0 to 1.0. + If None, the SDK default (0.0s) is used. sensitivity: Optional override for AIC VAD sensitivity (energy threshold). - Range: 1.0 .. 15.0. Energy threshold = 10 ** (-sensitivity). + Range: 1.0 to 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_factory = vad_factory - self._backend_vad: Optional[Any] = None - self._pending_lookback: Optional[float] = lookback_buffer_size + + 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_minimum_speech_duration: Optional[float] = minimum_speech_duration self._pending_sensitivity: Optional[float] = sensitivity - def bind_vad_factory(self, vad_factory: Callable[[], Any]): + def bind_vad_context_factory(self, vad_context_factory: Callable[[], Any]): """Attach or replace the factory post-construction.""" - self._vad_factory = vad_factory - self._ensure_backend_initialized() + self._vad_context_factory = vad_context_factory + self._ensure_vad_context_initialized() - def _apply_backend_params(self): + def _apply_vad_params(self): """Apply optional AIC VAD parameters if available.""" - if self._backend_vad is None or AICVadParameter is None: + if self._vad_ctx is None or VadParameter is None: return + try: - if self._pending_lookback is not None: - self._backend_vad.set_parameter( - AICVadParameter.LOOKBACK_BUFFER_SIZE, float(self._pending_lookback) + if self._pending_speech_hold_duration is not None: + self._vad_ctx.set_parameter( + VadParameter.SpeechHoldDuration, self._pending_speech_hold_duration + ) + if self._pending_minimum_speech_duration is not None: + self._vad_ctx.set_parameter( + VadParameter.MinimumSpeechDuration, self._pending_minimum_speech_duration ) if self._pending_sensitivity is not None: - self._backend_vad.set_parameter( - AICVadParameter.SENSITIVITY, float(self._pending_sensitivity) - ) + self._vad_ctx.set_parameter(VadParameter.Sensitivity, self._pending_sensitivity) except Exception as e: # noqa: BLE001 logger.debug(f"AIC VAD parameter application deferred/failed: {e}") - def _ensure_backend_initialized(self): - if self._backend_vad is not None: + def _ensure_vad_context_initialized(self): + if self._vad_ctx is not None: return - if not self._vad_factory: + if not self._vad_context_factory: return try: - self._backend_vad = self._vad_factory() - self._apply_backend_params() - # With backend ready, recompute internal frame sizing + 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 backend initialized in analyzer.") + logger.debug("AIC VAD context initialized in analyzer.") except Exception as e: # noqa: BLE001 # Filter may not be started yet; try again later - logger.debug(f"Deferring AIC VAD backend initialization: {e}") + logger.debug(f"Deferring AIC VAD context initialization: {e}") def set_sample_rate(self, sample_rate: int): """Set the sample rate for audio processing. @@ -116,10 +127,10 @@ class AICVADAnalyzer(VADAnalyzer): Args: sample_rate: Audio sample rate in Hz. """ - # Set rate and attempt backend initialization once we know SR + # Set rate and attempt VAD context initialization once we know SR self._sample_rate = self._init_sample_rate or sample_rate - self._ensure_backend_initialized() - # Ensure params are initialized even if backend not ready yet + self._ensure_vad_context_initialized() + # Ensure params are initialized even if VAD context not ready yet try: super().set_params(self._params) except Exception: @@ -135,23 +146,29 @@ class AICVADAnalyzer(VADAnalyzer): 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. + """Return voice activity detection result for the given audio buffer. + + Note: + The AIC SDK provides binary speech detection (not a probability score). + This method returns 1.0 when speech is detected and 0.0 otherwise, + rather than a true confidence value. Args: - buffer: Audio buffer to analyze. + buffer: Audio buffer (unused - AIC VAD state is updated internally + by the enhancement pipeline). Returns: - Voice confidence score is 0.0 or 1.0. + 1.0 if speech is detected, 0.0 otherwise. """ - # Ensure backend exists (filter might have started since last call) - self._ensure_backend_initialized() - if self._backend_vad is None: + # 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 model's VAD is updated + # 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._backend_vad.is_speech_detected() + 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}") diff --git a/tests/test_aic_filter.py b/tests/test_aic_filter.py new file mode 100644 index 000000000..6499084af --- /dev/null +++ b/tests/test_aic_filter.py @@ -0,0 +1,471 @@ +# +# Copyright (c) 2024-2026, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + +import asyncio +import unittest +from pathlib import Path +from unittest.mock import AsyncMock, MagicMock, patch + +import numpy as np + +# Check if aic_sdk is available +try: + import aic_sdk + + HAS_AIC_SDK = True +except ImportError: + HAS_AIC_SDK = False + +# Module path for patching +AIC_FILTER_MODULE = "pipecat.audio.filters.aic_filter" + + +class MockProcessor: + """A lightweight mock for AIC ProcessorAsync that mimics real behavior.""" + + def __init__(self): + self.processor_ctx = MockProcessorContext() + self.vad_ctx = MockVadContext() + + def get_processor_context(self): + return self.processor_ctx + + def get_vad_context(self): + return self.vad_ctx + + async def process_async(self, audio_array): + # Return a copy of the input (simulating passthrough) + return audio_array.copy() + + +class MockProcessorContext: + """A lightweight mock for AIC ProcessorContext.""" + + def __init__(self): + self.parameters_set: list[tuple] = [] + self.reset_called = False + self._output_delay = 0 + + def get_output_delay(self): + return self._output_delay + + def set_parameter(self, param, value): + self.parameters_set.append((param, value)) + + def reset(self): + self.reset_called = True + + +class MockVadContext: + """A lightweight mock for AIC VadContext.""" + + def __init__(self, speech_detected: bool = False): + self.speech_detected = speech_detected + self.parameters_set: list[tuple] = [] + + def is_speech_detected(self) -> bool: + return self.speech_detected + + def set_parameter(self, param, value): + self.parameters_set.append((param, value)) + + +class MockModel: + """A lightweight mock for AIC Model.""" + + def __init__(self, model_id: str = "test-model"): + self._model_id = model_id + self._optimal_num_frames = 160 + self._optimal_sample_rate = 16000 + + def get_optimal_num_frames(self, sample_rate: int): + """Return optimal number of frames for the given sample rate.""" + return self._optimal_num_frames + + def get_id(self): + return self._model_id + + def get_optimal_sample_rate(self): + return self._optimal_sample_rate + + +@unittest.skipUnless(HAS_AIC_SDK, "aic-sdk not installed") +class TestAICFilter(unittest.IsolatedAsyncioTestCase): + """Test suite for AICFilter audio filter using real aic_sdk types.""" + + @classmethod + def setUpClass(cls): + """Import AICFilter after confirming aic_sdk is available.""" + from pipecat.audio.filters.aic_filter import AICFilter + from pipecat.frames.frames import FilterEnableFrame + + cls.AICFilter = AICFilter + cls.FilterEnableFrame = FilterEnableFrame + + def setUp(self): + """Set up test fixtures before each test method.""" + self.mock_model = MockModel() + self.mock_processor = MockProcessor() + + def _create_filter_with_mocks(self, **kwargs): + """Create an AICFilter with mocked SDK components.""" + filter_kwargs = { + "license_key": "test-key", + "model_id": "test-model", + } + filter_kwargs.update(kwargs) + with patch(f"{AIC_FILTER_MODULE}.set_sdk_id"): + return self.AICFilter(**filter_kwargs) + + async def _start_filter_with_mocks(self, filter_instance, sample_rate=16000): + """Start a filter with mocked SDK components.""" + with ( + patch(f"{AIC_FILTER_MODULE}.Model") as mock_model_cls, + patch(f"{AIC_FILTER_MODULE}.ProcessorConfig") as mock_config_cls, + patch(f"{AIC_FILTER_MODULE}.ProcessorAsync", return_value=self.mock_processor), + ): + mock_model_cls.from_file.return_value = self.mock_model + mock_model_cls.download_async = AsyncMock(return_value="/tmp/model") + mock_config_cls.optimal.return_value = MagicMock() + await filter_instance.start(sample_rate) + + async def test_initialization_requires_model_id_or_path(self): + """Test filter initialization fails without model_id or model_path.""" + with patch(f"{AIC_FILTER_MODULE}.set_sdk_id"): + with self.assertRaises(ValueError) as context: + self.AICFilter(license_key="test-key") + + self.assertIn("model_id", str(context.exception)) + self.assertIn("model_path", str(context.exception)) + + async def test_initialization_with_model_id(self): + """Test filter initialization with model_id.""" + filter_instance = self._create_filter_with_mocks() + + self.assertEqual(filter_instance._license_key, "test-key") + self.assertEqual(filter_instance._model_id, "test-model") + self.assertIsNone(filter_instance._model_path) + self.assertFalse(filter_instance._bypass) + + async def test_initialization_with_model_path(self): + """Test filter initialization with model_path.""" + model_path = Path("/tmp/test.aicmodel") + filter_instance = self._create_filter_with_mocks(model_id=None, model_path=model_path) + + self.assertEqual(filter_instance._model_path, model_path) + self.assertIsNone(filter_instance._model_id) + + async def test_initialization_with_custom_download_dir(self): + """Test filter initialization with custom model_download_dir.""" + download_dir = Path("/custom/cache") + filter_instance = self._create_filter_with_mocks(model_download_dir=download_dir) + + self.assertEqual(filter_instance._model_download_dir, download_dir) + + async def test_start_with_model_path(self): + """Test starting filter with a local model path.""" + model_path = Path("/tmp/test.aicmodel") + filter_instance = self._create_filter_with_mocks(model_id=None, model_path=model_path) + + with ( + patch(f"{AIC_FILTER_MODULE}.Model") as mock_model_cls, + patch(f"{AIC_FILTER_MODULE}.ProcessorConfig") as mock_config_cls, + patch(f"{AIC_FILTER_MODULE}.ProcessorAsync", return_value=self.mock_processor), + ): + mock_model_cls.from_file.return_value = self.mock_model + mock_config_cls.optimal.return_value = MagicMock() + + await filter_instance.start(16000) + + mock_model_cls.from_file.assert_called_once_with(str(model_path)) + self.assertTrue(filter_instance._aic_ready) + self.assertEqual(filter_instance._sample_rate, 16000) + self.assertEqual(filter_instance._frames_per_block, 160) + + async def test_start_with_model_id_downloads(self): + """Test starting filter with model_id triggers download.""" + filter_instance = self._create_filter_with_mocks() + + with ( + patch(f"{AIC_FILTER_MODULE}.Model") as mock_model_cls, + patch(f"{AIC_FILTER_MODULE}.ProcessorConfig") as mock_config_cls, + patch(f"{AIC_FILTER_MODULE}.ProcessorAsync", return_value=self.mock_processor), + ): + mock_model_cls.from_file.return_value = self.mock_model + mock_model_cls.download_async = AsyncMock(return_value="/tmp/model") + mock_config_cls.optimal.return_value = MagicMock() + + await filter_instance.start(16000) + + mock_model_cls.download_async.assert_called_once() + mock_model_cls.from_file.assert_called_once() + self.assertTrue(filter_instance._aic_ready) + + async def test_start_creates_processor(self): + """Test that start creates processor with correct config.""" + filter_instance = self._create_filter_with_mocks() + + with ( + patch(f"{AIC_FILTER_MODULE}.Model") as mock_model_cls, + patch(f"{AIC_FILTER_MODULE}.ProcessorConfig") as mock_config_cls, + patch( + f"{AIC_FILTER_MODULE}.ProcessorAsync", return_value=self.mock_processor + ) as mock_processor_cls, + ): + mock_model_cls.from_file.return_value = self.mock_model + mock_model_cls.download_async = AsyncMock(return_value="/tmp/model") + mock_config_cls.optimal.return_value = MagicMock() + + await filter_instance.start(16000) + + mock_config_cls.optimal.assert_called_once() + mock_processor_cls.assert_called_once() + self.assertIsNotNone(filter_instance._processor_ctx) + self.assertIsNotNone(filter_instance._vad_ctx) + + async def test_start_applies_initial_bypass_parameter(self): + """Test that start applies bypass parameter.""" + filter_instance = self._create_filter_with_mocks() + await self._start_filter_with_mocks(filter_instance) + + # Check that bypass was set to 0.0 (enabled) + bypass_params = [ + (p, v) + for p, v in self.mock_processor.processor_ctx.parameters_set + if p == aic_sdk.ProcessorParameter.Bypass + ] + self.assertTrue(len(bypass_params) > 0) + self.assertEqual(bypass_params[-1][1], 0.0) + + async def test_stop_cleans_up_resources(self): + """Test that stop properly cleans up resources.""" + filter_instance = self._create_filter_with_mocks() + await self._start_filter_with_mocks(filter_instance) + + await filter_instance.stop() + + self.assertTrue(self.mock_processor.processor_ctx.reset_called) + self.assertIsNone(filter_instance._processor) + self.assertIsNone(filter_instance._processor_ctx) + self.assertIsNone(filter_instance._vad_ctx) + self.assertIsNone(filter_instance._model) + self.assertFalse(filter_instance._aic_ready) + + async def test_stop_without_start(self): + """Test that stop can be called safely without start.""" + filter_instance = self._create_filter_with_mocks() + + # Should not raise + await filter_instance.stop() + + async def test_process_frame_enable(self): + """Test processing FilterEnableFrame to enable filtering.""" + filter_instance = self._create_filter_with_mocks() + await self._start_filter_with_mocks(filter_instance) + filter_instance._bypass = True + + enable_frame = self.FilterEnableFrame(enable=True) + await filter_instance.process_frame(enable_frame) + + self.assertFalse(filter_instance._bypass) + + async def test_process_frame_disable(self): + """Test processing FilterEnableFrame to disable filtering.""" + filter_instance = self._create_filter_with_mocks() + await self._start_filter_with_mocks(filter_instance) + + disable_frame = self.FilterEnableFrame(enable=False) + await filter_instance.process_frame(disable_frame) + + self.assertTrue(filter_instance._bypass) + + async def test_filter_when_not_ready(self): + """Test that filter returns audio unchanged when not ready.""" + filter_instance = self._create_filter_with_mocks() + # Don't call start() + + input_audio = b"\x00\x01\x02\x03" + output_audio = await filter_instance.filter(input_audio) + + self.assertEqual(output_audio, input_audio) + + async def test_filter_with_incomplete_frame(self): + """Test filtering audio with incomplete frame data.""" + filter_instance = self._create_filter_with_mocks() + await self._start_filter_with_mocks(filter_instance) + + # Create audio data for less than one frame (100 samples = 200 bytes) + samples = np.random.randint(-32768, 32767, size=100, dtype=np.int16) + input_audio = samples.tobytes() + + output_audio = await filter_instance.filter(input_audio) + + # Should return empty bytes since no complete frame + self.assertEqual(output_audio, b"") + + async def test_filter_with_complete_frame(self): + """Test filtering audio with exactly one complete frame.""" + filter_instance = self._create_filter_with_mocks() + await self._start_filter_with_mocks(filter_instance) + + # Create audio data for exactly one frame (160 samples = 320 bytes) + samples = np.random.randint(-32768, 32767, size=160, dtype=np.int16) + input_audio = samples.tobytes() + + output_audio = await filter_instance.filter(input_audio) + + self.assertIsInstance(output_audio, bytes) + self.assertEqual(len(output_audio), len(input_audio)) + + async def test_filter_with_multiple_frames(self): + """Test filtering audio with multiple complete frames.""" + filter_instance = self._create_filter_with_mocks() + await self._start_filter_with_mocks(filter_instance) + + # Create audio data for 3 complete frames (480 samples = 960 bytes) + samples = np.random.randint(-32768, 32767, size=480, dtype=np.int16) + input_audio = samples.tobytes() + + output_audio = await filter_instance.filter(input_audio) + + self.assertEqual(len(output_audio), len(input_audio)) + + async def test_filter_with_buffering(self): + """Test that filter properly buffers incomplete frames.""" + filter_instance = self._create_filter_with_mocks() + await self._start_filter_with_mocks(filter_instance) + + # First call: Send 100 samples (incomplete frame) + samples1 = np.random.randint(-32768, 32767, size=100, dtype=np.int16) + input_audio1 = samples1.tobytes() + output_audio1 = await filter_instance.filter(input_audio1) + + self.assertEqual(output_audio1, b"") + self.assertEqual(len(filter_instance._audio_buffer), 200) + + # Second call: Send 60 more samples (now we have 160 total = 1 complete frame) + samples2 = np.random.randint(-32768, 32767, size=60, dtype=np.int16) + input_audio2 = samples2.tobytes() + output_audio2 = await filter_instance.filter(input_audio2) + + self.assertEqual(len(output_audio2), 320) + self.assertEqual(len(filter_instance._audio_buffer), 0) + + async def test_filter_with_partial_buffering(self): + """Test that filter keeps remainder in buffer after processing.""" + filter_instance = self._create_filter_with_mocks() + await self._start_filter_with_mocks(filter_instance) + + # Send 250 samples (1 complete frame + 90 samples remainder) + samples = np.random.randint(-32768, 32767, size=250, dtype=np.int16) + input_audio = samples.tobytes() + + output_audio = await filter_instance.filter(input_audio) + + self.assertEqual(len(output_audio), 320) # 1 frame + self.assertEqual(len(filter_instance._audio_buffer), 180) # 90 samples * 2 bytes + + async def test_get_vad_context_before_start(self): + """Test that get_vad_context raises before start.""" + filter_instance = self._create_filter_with_mocks() + + with self.assertRaises(RuntimeError) as context: + filter_instance.get_vad_context() + + self.assertIn("not initialized", str(context.exception)) + + async def test_get_vad_context_after_start(self): + """Test that get_vad_context returns context after start.""" + filter_instance = self._create_filter_with_mocks() + await self._start_filter_with_mocks(filter_instance) + + vad_ctx = filter_instance.get_vad_context() + + self.assertEqual(vad_ctx, self.mock_processor.vad_ctx) + + async def test_create_vad_analyzer(self): + """Test create_vad_analyzer returns analyzer with factory.""" + filter_instance = self._create_filter_with_mocks() + + analyzer = filter_instance.create_vad_analyzer() + + self.assertIsNotNone(analyzer) + # Factory should be set + self.assertIsNotNone(analyzer._vad_context_factory) + + async def test_create_vad_analyzer_with_params(self): + """Test create_vad_analyzer with custom parameters.""" + filter_instance = self._create_filter_with_mocks() + + analyzer = filter_instance.create_vad_analyzer( + speech_hold_duration=0.1, + minimum_speech_duration=0.05, + sensitivity=8.0, + ) + + self.assertEqual(analyzer._pending_speech_hold_duration, 0.1) + self.assertEqual(analyzer._pending_minimum_speech_duration, 0.05) + self.assertEqual(analyzer._pending_sensitivity, 8.0) + + async def test_multiple_start_stop_cycles(self): + """Test multiple start/stop cycles.""" + filter_instance = self._create_filter_with_mocks() + + for sample_rate in [16000, 24000, 48000]: + # Create fresh mock processor for each cycle + self.mock_processor = MockProcessor() + await self._start_filter_with_mocks(filter_instance, sample_rate) + self.assertTrue(filter_instance._aic_ready) + self.assertEqual(filter_instance._sample_rate, sample_rate) + + await filter_instance.stop() + self.assertFalse(filter_instance._aic_ready) + + async def test_concurrent_filter_calls(self): + """Test that concurrent filter calls are handled safely.""" + filter_instance = self._create_filter_with_mocks() + await self._start_filter_with_mocks(filter_instance) + + samples = np.random.randint(-32768, 32767, size=160, dtype=np.int16) + input_audio = samples.tobytes() + + async def filter_audio(): + return await filter_instance.filter(input_audio) + + tasks = [filter_audio() for _ in range(10)] + results = await asyncio.gather(*tasks) + + self.assertEqual(len(results), 10) + for result in results: + self.assertIsInstance(result, bytes) + + async def test_buffer_cleared_on_stop(self): + """Test that audio buffer is cleared when stopping.""" + filter_instance = self._create_filter_with_mocks() + await self._start_filter_with_mocks(filter_instance) + + # Add incomplete frame to buffer + samples = np.random.randint(-32768, 32767, size=100, dtype=np.int16) + input_audio = samples.tobytes() + await filter_instance.filter(input_audio) + + # Verify buffer has data + self.assertGreater(len(filter_instance._audio_buffer), 0) + + # Stop should clear buffer + await filter_instance.stop() + self.assertEqual(len(filter_instance._audio_buffer), 0) + + async def test_set_sdk_id_called_on_init(self): + """Test that set_sdk_id is called during initialization.""" + with patch(f"{AIC_FILTER_MODULE}.set_sdk_id") as mock_set_sdk_id: + self.AICFilter(license_key="test-key", model_id="test-model") + + mock_set_sdk_id.assert_called_once_with(6) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_aic_vad.py b/tests/test_aic_vad.py new file mode 100644 index 000000000..5028da46b --- /dev/null +++ b/tests/test_aic_vad.py @@ -0,0 +1,322 @@ +# +# Copyright (c) 2024-2026, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + +import unittest + +# Check if aic_sdk is available +try: + import aic_sdk + + HAS_AIC_SDK = True +except ImportError: + HAS_AIC_SDK = False + + +@unittest.skipUnless(HAS_AIC_SDK, "aic-sdk not installed") +class TestAICVADAnalyzer(unittest.IsolatedAsyncioTestCase): + """Test suite for AICVADAnalyzer using real aic_sdk.""" + + @classmethod + def setUpClass(cls): + """Import AICVADAnalyzer after confirming aic_sdk is available.""" + from pipecat.audio.vad.aic_vad import AICVADAnalyzer + + cls.AICVADAnalyzer = AICVADAnalyzer + + def test_initialization_without_factory(self): + """Test analyzer initialization without a factory.""" + analyzer = self.AICVADAnalyzer() + + self.assertIsNone(analyzer._vad_context_factory) + self.assertIsNone(analyzer._vad_ctx) + # Fixed params should be set + self.assertEqual(analyzer._params.confidence, 0.5) + self.assertEqual(analyzer._params.start_secs, 0.0) + self.assertEqual(analyzer._params.stop_secs, 0.0) + self.assertEqual(analyzer._params.min_volume, 0.0) + + def test_initialization_with_factory(self): + """Test analyzer initialization with a factory.""" + # Create a mock VAD context for testing + mock_vad_ctx = MockVadContext() + factory = lambda: mock_vad_ctx + analyzer = self.AICVADAnalyzer(vad_context_factory=factory) + + self.assertIsNotNone(analyzer._vad_context_factory) + + def test_initialization_with_vad_params(self): + """Test analyzer initialization with VAD parameters.""" + analyzer = self.AICVADAnalyzer( + speech_hold_duration=0.1, + minimum_speech_duration=0.05, + sensitivity=8.0, + ) + + self.assertEqual(analyzer._pending_speech_hold_duration, 0.1) + self.assertEqual(analyzer._pending_minimum_speech_duration, 0.05) + self.assertEqual(analyzer._pending_sensitivity, 8.0) + + def test_bind_vad_context_factory(self): + """Test binding a factory post-construction.""" + mock_vad_ctx = MockVadContext() + analyzer = self.AICVADAnalyzer() + factory = lambda: mock_vad_ctx + + analyzer.bind_vad_context_factory(factory) + + self.assertEqual(analyzer._vad_context_factory, factory) + # Should have attempted to initialize + self.assertEqual(analyzer._vad_ctx, mock_vad_ctx) + + def test_bind_vad_context_factory_applies_params(self): + """Test that binding factory applies pending VAD params.""" + mock_vad_ctx = MockVadContext() + analyzer = self.AICVADAnalyzer( + speech_hold_duration=0.1, + minimum_speech_duration=0.05, + sensitivity=8.0, + ) + factory = lambda: mock_vad_ctx + + analyzer.bind_vad_context_factory(factory) + + # Verify parameters were applied + self.assertIn( + (aic_sdk.VadParameter.SpeechHoldDuration, 0.1), + mock_vad_ctx.parameters_set, + ) + self.assertIn( + (aic_sdk.VadParameter.MinimumSpeechDuration, 0.05), + mock_vad_ctx.parameters_set, + ) + self.assertIn( + (aic_sdk.VadParameter.Sensitivity, 8.0), + mock_vad_ctx.parameters_set, + ) + + def test_set_sample_rate(self): + """Test setting sample rate.""" + analyzer = self.AICVADAnalyzer() + + analyzer.set_sample_rate(16000) + + self.assertEqual(analyzer._sample_rate, 16000) + + def test_set_sample_rate_with_init_sample_rate(self): + """Test that init_sample_rate takes precedence.""" + # Create analyzer and manually set _init_sample_rate + analyzer = self.AICVADAnalyzer() + analyzer._init_sample_rate = 48000 + + analyzer.set_sample_rate(16000) + + # init_sample_rate should take precedence + self.assertEqual(analyzer._sample_rate, 48000) + + def test_set_sample_rate_triggers_context_init(self): + """Test that set_sample_rate attempts context initialization.""" + mock_vad_ctx = MockVadContext() + factory = lambda: mock_vad_ctx + analyzer = self.AICVADAnalyzer(vad_context_factory=factory) + + analyzer.set_sample_rate(16000) + + self.assertEqual(analyzer._vad_ctx, mock_vad_ctx) + + def test_num_frames_required_with_sample_rate(self): + """Test num_frames_required returns correct value.""" + analyzer = self.AICVADAnalyzer() + analyzer.set_sample_rate(16000) + + frames = analyzer.num_frames_required() + + # 10ms at 16kHz = 160 frames + self.assertEqual(frames, 160) + + def test_num_frames_required_different_sample_rates(self): + """Test num_frames_required for different sample rates.""" + analyzer = self.AICVADAnalyzer() + + test_cases = [ + (8000, 80), # 10ms at 8kHz + (16000, 160), # 10ms at 16kHz + (24000, 240), # 10ms at 24kHz + (48000, 480), # 10ms at 48kHz + ] + + for sample_rate, expected_frames in test_cases: + analyzer.set_sample_rate(sample_rate) + frames = analyzer.num_frames_required() + self.assertEqual(frames, expected_frames, f"Failed for {sample_rate}Hz") + + def test_num_frames_required_no_sample_rate(self): + """Test num_frames_required returns default when no sample rate.""" + analyzer = self.AICVADAnalyzer() + + frames = analyzer.num_frames_required() + + # Default is 160 + self.assertEqual(frames, 160) + + def test_voice_confidence_no_context(self): + """Test voice_confidence returns 0.0 when no context.""" + analyzer = self.AICVADAnalyzer() + + confidence = analyzer.voice_confidence(b"\x00" * 320) + + self.assertEqual(confidence, 0.0) + + def test_voice_confidence_speech_detected(self): + """Test voice_confidence returns 1.0 when speech detected.""" + mock_vad_ctx = MockVadContext(speech_detected=True) + factory = lambda: mock_vad_ctx + analyzer = self.AICVADAnalyzer(vad_context_factory=factory) + analyzer.set_sample_rate(16000) + + confidence = analyzer.voice_confidence(b"\x00" * 320) + + self.assertEqual(confidence, 1.0) + + def test_voice_confidence_no_speech(self): + """Test voice_confidence returns 0.0 when no speech.""" + mock_vad_ctx = MockVadContext(speech_detected=False) + factory = lambda: mock_vad_ctx + analyzer = self.AICVADAnalyzer(vad_context_factory=factory) + analyzer.set_sample_rate(16000) + + confidence = analyzer.voice_confidence(b"\x00" * 320) + + self.assertEqual(confidence, 0.0) + + def test_voice_confidence_handles_exception(self): + """Test voice_confidence handles exceptions gracefully.""" + mock_vad_ctx = MockVadContext(raise_on_detect=True) + factory = lambda: mock_vad_ctx + analyzer = self.AICVADAnalyzer(vad_context_factory=factory) + analyzer.set_sample_rate(16000) + + confidence = analyzer.voice_confidence(b"\x00" * 320) + + self.assertEqual(confidence, 0.0) + + def test_lazy_initialization(self): + """Test that VAD context is lazily initialized.""" + call_count = 0 + mock_vad_ctx = MockVadContext() + + def counting_factory(): + nonlocal call_count + call_count += 1 + return mock_vad_ctx + + analyzer = self.AICVADAnalyzer(vad_context_factory=counting_factory) + + # Factory not called yet + self.assertEqual(call_count, 0) + + # First call to voice_confidence triggers initialization + analyzer.voice_confidence(b"\x00" * 320) + self.assertEqual(call_count, 1) + + # Subsequent calls don't re-initialize + analyzer.voice_confidence(b"\x00" * 320) + analyzer.voice_confidence(b"\x00" * 320) + self.assertEqual(call_count, 1) + + def test_deferred_initialization_on_factory_failure(self): + """Test that initialization is deferred when factory fails.""" + call_count = 0 + mock_vad_ctx = MockVadContext(speech_detected=True) + + def failing_then_succeeding_factory(): + nonlocal call_count + call_count += 1 + if call_count < 3: + raise RuntimeError("Not ready yet") + return mock_vad_ctx + + analyzer = self.AICVADAnalyzer(vad_context_factory=failing_then_succeeding_factory) + + # First two calls fail, should return 0.0 + self.assertEqual(analyzer.voice_confidence(b"\x00" * 320), 0.0) + self.assertEqual(analyzer.voice_confidence(b"\x00" * 320), 0.0) + + # Third call succeeds + self.assertEqual(analyzer.voice_confidence(b"\x00" * 320), 1.0) + + def test_apply_vad_params_deferred_on_failure(self): + """Test that VAD param application handles exceptions.""" + mock_vad_ctx = MockVadContext(raise_on_set_param=True) + factory = lambda: mock_vad_ctx + + analyzer = self.AICVADAnalyzer( + vad_context_factory=factory, + speech_hold_duration=0.1, + ) + + # Should not raise, just log debug message + analyzer.bind_vad_context_factory(factory) + + # Context should still be set despite param failure + self.assertEqual(analyzer._vad_ctx, mock_vad_ctx) + + def test_apply_vad_params_only_set_values(self): + """Test that only specified VAD params are applied.""" + mock_vad_ctx = MockVadContext() + factory = lambda: mock_vad_ctx + analyzer = self.AICVADAnalyzer( + vad_context_factory=factory, + speech_hold_duration=0.1, + # minimum_speech_duration and sensitivity not set + ) + + analyzer.bind_vad_context_factory(factory) + + # Only SpeechHoldDuration should be set + self.assertEqual(len(mock_vad_ctx.parameters_set), 1) + self.assertIn( + (aic_sdk.VadParameter.SpeechHoldDuration, 0.1), + mock_vad_ctx.parameters_set, + ) + + def test_fixed_vad_params(self): + """Test that VAD uses fixed parameters.""" + analyzer = self.AICVADAnalyzer() + + # These are the fixed params for AIC VAD + self.assertEqual(analyzer._params.confidence, 0.5) + self.assertEqual(analyzer._params.start_secs, 0.0) + self.assertEqual(analyzer._params.stop_secs, 0.0) + self.assertEqual(analyzer._params.min_volume, 0.0) + + +class MockVadContext: + """A lightweight mock for AIC VadContext that mimics real behavior.""" + + def __init__( + self, + speech_detected: bool = False, + raise_on_detect: bool = False, + raise_on_set_param: bool = False, + ): + self.speech_detected = speech_detected + self.raise_on_detect = raise_on_detect + self.raise_on_set_param = raise_on_set_param + self.parameters_set: list[tuple] = [] + + def is_speech_detected(self) -> bool: + if self.raise_on_detect: + raise RuntimeError("VAD error") + return self.speech_detected + + def set_parameter(self, param, value): + if self.raise_on_set_param: + raise RuntimeError("Param error") + self.parameters_set.append((param, value)) + + +if __name__ == "__main__": + unittest.main()