# # Copyright (c) 2024-2026, Daily # # SPDX-License-Identifier: BSD 2-Clause License # """Base input transport implementation for Pipecat. This module provides the BaseInputTransport class which handles audio and video input processing, including VAD, turn analysis, and interruption management. """ import asyncio import time from typing import Optional from loguru import logger from pipecat.audio.turn.base_turn_analyzer import ( BaseTurnAnalyzer, EndOfTurnState, ) from pipecat.audio.vad.vad_analyzer import VADAnalyzer, VADState from pipecat.frames.frames import ( BotStartedSpeakingFrame, BotStoppedSpeakingFrame, CancelFrame, EmulateUserStartedSpeakingFrame, EmulateUserStoppedSpeakingFrame, EndFrame, FilterUpdateSettingsFrame, Frame, InputAudioRawFrame, InputImageRawFrame, MetricsFrame, SpeechControlParamsFrame, StartFrame, StopFrame, SystemFrame, UserSpeakingFrame, UserStartedSpeakingFrame, UserStoppedSpeakingFrame, VADParamsUpdateFrame, VADUserStartedSpeakingFrame, VADUserStoppedSpeakingFrame, ) from pipecat.metrics.metrics import MetricsData from pipecat.processors.frame_processor import FrameDirection, FrameProcessor from pipecat.transports.base_transport import TransportParams AUDIO_INPUT_TIMEOUT_SECS = 0.5 class BaseInputTransport(FrameProcessor): """Base class for input transport implementations. Handles audio and video input processing including Voice Activity Detection, turn analysis, audio filtering, and user interaction management. Supports interruption handling and provides hooks for transport-specific implementations. """ def __init__(self, params: TransportParams, **kwargs): """Initialize the base input transport. Args: params: Transport configuration parameters. **kwargs: Additional arguments passed to parent class. """ super().__init__(**kwargs) self._params = params # Input sample rate. It will be initialized on StartFrame. self._sample_rate = 0 # Track bot speaking state for interruption logic self._bot_speaking = False # Track user speaking state for interruption logic self._user_speaking = False # Last time a UserSpeakingFrame was pushed. self._user_speaking_frame_time = 0 # How often a UserSpeakingFrame should be pushed (value should be # greater than the audio chunks to have any effect). self._user_speaking_frame_period = 0.2 # Task to process incoming audio (VAD) and push audio frames downstream # if passthrough is enabled. self._audio_task = None # If the transport is stopped with `StopFrame` we might still be # receiving frames from the transport but we really don't want to push # them downstream until we get another `StartFrame`. self._paused = False if self._params.vad_enabled: import warnings with warnings.catch_warnings(): warnings.simplefilter("always") warnings.warn( "Parameter 'vad_enabled' is deprecated, use 'audio_in_enabled' and 'vad_analyzer' instead.", DeprecationWarning, ) self._params.audio_in_enabled = True if self._params.vad_audio_passthrough: import warnings with warnings.catch_warnings(): warnings.simplefilter("always") warnings.warn( "Parameter 'vad_audio_passthrough' is deprecated, audio passthrough is now always enabled. Use 'audio_in_passthrough' to disable.", DeprecationWarning, ) self._params.audio_in_passthrough = True if self._params.camera_in_enabled: import warnings with warnings.catch_warnings(): warnings.simplefilter("always") warnings.warn( "Parameters 'camera_*' are deprecated, use 'video_*' instead.", DeprecationWarning, ) self._params.video_in_enabled = self._params.camera_in_enabled self._params.video_out_enabled = self._params.camera_out_enabled self._params.video_out_is_live = self._params.camera_out_is_live self._params.video_out_width = self._params.camera_out_width self._params.video_out_height = self._params.camera_out_height self._params.video_out_bitrate = self._params.camera_out_bitrate self._params.video_out_framerate = self._params.camera_out_framerate self._params.video_out_color_format = self._params.camera_out_color_format if self._params.turn_analyzer: import warnings with warnings.catch_warnings(): warnings.simplefilter("always") warnings.warn( "Parameter 'turn_analyzer' is deprecated, use `LLMUserAggregator`'s new " "`user_turn_strategies` parameter instead.", DeprecationWarning, ) if self._params.vad_analyzer: import warnings with warnings.catch_warnings(): warnings.simplefilter("always") warnings.warn( "Parameter 'vad_analyzer' is deprecated, use `LLMUserAggregator`'s new " "`vad_analyzer` parameter instead.", DeprecationWarning, ) def enable_audio_in_stream_on_start(self, enabled: bool) -> None: """Enable or disable audio streaming on transport start. Args: enabled: Whether to start audio streaming immediately on transport start. """ logger.debug(f"Enabling audio on start. {enabled}") self._params.audio_in_stream_on_start = enabled async def start_audio_in_streaming(self): """Start audio input streaming. Override in subclasses to implement transport-specific audio streaming. """ pass @property def sample_rate(self) -> int: """Get the current audio sample rate. Returns: The sample rate in Hz. """ return self._sample_rate @property def vad_analyzer(self) -> Optional[VADAnalyzer]: """Get the Voice Activity Detection analyzer. .. deprecated:: 0.0.101 This method is deprecated and will be removed in a future version. Use `LLMUserAggregator`'s new `vad_analyzer` parameter instead. Returns: The VAD analyzer instance if configured, None otherwise. """ import warnings with warnings.catch_warnings(): warnings.simplefilter("always") warnings.warn( "Method 'vad_analyzer' is deprecated. Use `LLMUserAggregator`'s new " "`vad_analyzer` parameter instead.", DeprecationWarning, ) return self._params.vad_analyzer @property def turn_analyzer(self) -> Optional[BaseTurnAnalyzer]: """Get the turn-taking analyzer. .. deprecated:: 0.0.99 This method is deprecated and will be removed in a future version. Use `LLMUserAggregator`'s new `user_turn_strategies` parameter instead. Returns: The turn analyzer instance if configured, None otherwise. """ import warnings with warnings.catch_warnings(): warnings.simplefilter("always") warnings.warn( "Method 'turn_analyzer' is deprecated. Use `LLMUserAggregator`'s new " "`user_turn_strategies` parameter instead.", DeprecationWarning, ) return self._params.turn_analyzer async def start(self, frame: StartFrame): """Start the input transport and initialize components. Args: frame: The start frame containing initialization parameters. """ self._paused = False self._user_speaking = False self._sample_rate = self._params.audio_in_sample_rate or frame.audio_in_sample_rate # Start audio filter. if self._params.audio_in_filter: await self._params.audio_in_filter.start(self._sample_rate) ################################################################### # DEPRECATED. # Configure VAD analyzer. if self._params.vad_analyzer: self._params.vad_analyzer.set_sample_rate(self._sample_rate) # Configure End of turn analyzer. if self._params.turn_analyzer: self._params.turn_analyzer.set_sample_rate(self._sample_rate) if self._params.vad_analyzer or self._params.turn_analyzer: vad_params = self._params.vad_analyzer.params if self._params.vad_analyzer else None turn_params = self._params.turn_analyzer.params if self._params.turn_analyzer else None await self.broadcast_frame( SpeechControlParamsFrame, vad_params=vad_params, turn_params=turn_params ) ################################################################### async def stop(self, frame: EndFrame): """Stop the input transport and cleanup resources. Args: frame: The end frame signaling transport shutdown. """ # Cancel and wait for the audio input task to finish. await self._cancel_audio_task() # Stop audio filter. if self._params.audio_in_filter: await self._params.audio_in_filter.stop() async def pause(self, frame: StopFrame): """Pause the input transport temporarily. Args: frame: The stop frame signaling transport pause. """ self._paused = True # Cancel task so we clear the queue await self._cancel_audio_task() # Retart the task self._create_audio_task() async def cancel(self, frame: CancelFrame): """Cancel the input transport and stop all processing. Args: frame: The cancel frame signaling immediate cancellation. """ # Cancel and wait for the audio input task to finish. await self._cancel_audio_task() # Stop audio filter. if self._params.audio_in_filter: await self._params.audio_in_filter.stop() async def set_transport_ready(self, frame: StartFrame): """Called when the transport is ready to stream. Args: frame: The start frame containing initialization parameters. """ # Create audio input queue and task if needed. self._create_audio_task() async def push_video_frame(self, frame: InputImageRawFrame): """Push a video frame downstream if video input is enabled. Args: frame: The input video frame to process. """ if self._params.video_in_enabled and not self._paused: await self.push_frame(frame) async def push_audio_frame(self, frame: InputAudioRawFrame): """Push an audio frame to the processing queue if audio input is enabled. Args: frame: The input audio frame to process. """ if self._params.audio_in_enabled and not self._paused: await self._audio_in_queue.put(frame) # # Frame processor # async def process_frame(self, frame: Frame, direction: FrameDirection): """Process incoming frames and handle transport-specific logic. Args: frame: The frame to process. direction: The direction of frame flow in the pipeline. """ await super().process_frame(frame, direction) # Specific system frames if isinstance(frame, StartFrame): # Push StartFrame before start(), because we want StartFrame to be # processed by every processor before any other frame is processed. await self.push_frame(frame, direction) await self.start(frame) elif isinstance(frame, CancelFrame): await self.cancel(frame) await self.push_frame(frame, direction) elif isinstance(frame, BotStartedSpeakingFrame): await self._deprecated_handle_bot_started_speaking(frame) await self.push_frame(frame, direction) elif isinstance(frame, BotStoppedSpeakingFrame): await self._deprecated_handle_bot_stopped_speaking(frame) await self.push_frame(frame, direction) elif isinstance(frame, EmulateUserStartedSpeakingFrame): logger.debug("Emulating user started speaking") await self._deprecated_handle_user_interruption(VADState.SPEAKING, emulated=True) elif isinstance(frame, EmulateUserStoppedSpeakingFrame): logger.debug("Emulating user stopped speaking") await self._deprecated_handle_user_interruption(VADState.QUIET, emulated=True) # All other system frames elif isinstance(frame, SystemFrame): await self.push_frame(frame, direction) # Control frames elif isinstance(frame, EndFrame): # Push EndFrame before stop(), because stop() waits on the task to # finish and the task finishes when EndFrame is processed. await self.push_frame(frame, direction) await self.stop(frame) elif isinstance(frame, StopFrame): await self.push_frame(frame, direction) await self.pause(frame) ################################################################### # DEPRECATED. elif isinstance(frame, VADParamsUpdateFrame): if self._params.vad_analyzer: self._params.vad_analyzer.set_params(frame.params) await self.broadcast_frame( SpeechControlParamsFrame, vad_params=frame.params, turn_params=self._params.turn_analyzer.params if self._params.turn_analyzer else None, ) ################################################################### elif isinstance(frame, FilterUpdateSettingsFrame) and self._params.audio_in_filter: await self._params.audio_in_filter.process_frame(frame) # Other frames else: await self.push_frame(frame, direction) # # Audio input # def _create_audio_task(self): """Create the audio processing task if audio input is enabled.""" if not self._audio_task and self._params.audio_in_enabled: self._audio_in_queue = asyncio.Queue() self._audio_task = self.create_task(self._audio_task_handler()) async def _cancel_audio_task(self): """Cancel and cleanup the audio processing task.""" if self._audio_task: await self.cancel_task(self._audio_task) self._audio_task = None async def _audio_task_handler(self): """Main audio processing task handler for VAD and turn analysis.""" vad_state: VADState = VADState.QUIET # Skip timeout handling until the first audio frame arrives (e.g. client # not yet connected). audio_received = False while True: try: frame: InputAudioRawFrame = await asyncio.wait_for( self._audio_in_queue.get(), timeout=AUDIO_INPUT_TIMEOUT_SECS ) # From now on, timeout should warn if there's no audio. audio_received = True # If an audio filter is available, run it before VAD. if self._params.audio_in_filter: frame.audio = await self._params.audio_in_filter.filter(frame.audio) # Skip frames with no audio data (e.g. filter is buffering). if not frame.audio: self._audio_in_queue.task_done() continue ################################################################### # DEPRECATED. # # Check VAD and push event if necessary. We just care about # changes from QUIET to SPEAKING and vice versa. previous_vad_state = vad_state if self._params.vad_analyzer: vad_state = await self._deprecated_handle_vad(frame, vad_state) if self._params.turn_analyzer: await self._deprecated_run_turn_analyzer(frame, vad_state, previous_vad_state) if self._params.vad_analyzer and vad_state == VADState.SPEAKING: await self._deprecated_user_currently_speaking() ################################################################### # Push audio downstream if passthrough is set. if self._params.audio_in_passthrough: await self.push_frame(frame) self._audio_in_queue.task_done() except asyncio.TimeoutError: if not audio_received: continue ################################################################### # DEPRECATED. if self._user_speaking: logger.warning( "Forcing VAD user stopped speaking due to timeout receiving audio frame!" ) vad_state = VADState.QUIET if self._params.turn_analyzer: self._params.turn_analyzer.clear() if self._params.turn_analyzer: await self._deprecated_handle_user_interruption(VADState.QUIET) else: stop_secs = ( self._params.vad_analyzer.params.stop_secs if self._params.vad_analyzer else 0.0 ) await self.push_frame(VADUserStoppedSpeakingFrame(stop_secs=stop_secs)) ################################################################### # # DEPRECATED. # # The functions below are deprecated and should be removed once the old # interruption strategies and turn analyzer are removed. # async def _deprecated_vad_analyze(self, audio_frame: InputAudioRawFrame) -> VADState: """Analyze audio frame for voice activity.""" state = VADState.QUIET if self._params.vad_analyzer: state = await self._params.vad_analyzer.analyze_audio(audio_frame.audio) return state async def _deprecated_new_handle_vad( self, audio_frame: InputAudioRawFrame, vad_state: VADState ) -> VADState: """Handle Voice Activity Detection results and generate appropriate frames.""" new_vad_state = await self._deprecated_vad_analyze(audio_frame) if ( new_vad_state != vad_state and new_vad_state != VADState.STARTING and new_vad_state != VADState.STOPPING ): if new_vad_state == VADState.SPEAKING: start_secs = ( self._params.vad_analyzer.params.start_secs if self._params.vad_analyzer else 0.0 ) await self.push_frame(VADUserStartedSpeakingFrame(start_secs=start_secs)) elif new_vad_state == VADState.QUIET: stop_secs = ( self._params.vad_analyzer.params.stop_secs if self._params.vad_analyzer else 0.0 ) await self.push_frame(VADUserStoppedSpeakingFrame(stop_secs=stop_secs)) vad_state = new_vad_state return vad_state async def _deprecated_handle_vad( self, audio_frame: InputAudioRawFrame, vad_state: VADState ) -> VADState: """Handle Voice Activity Detection results and generate appropriate frames.""" if self._params.turn_analyzer or self._deprecated_openaillmcontext: return await self._deprecated_old_handle_vad(audio_frame, vad_state) else: return await self._deprecated_new_handle_vad(audio_frame, vad_state) async def _deprecated_user_currently_speaking(self): """Handle user speaking frame.""" diff_time = time.time() - self._user_speaking_frame_time if diff_time >= self._user_speaking_frame_period: await self.broadcast_frame(UserSpeakingFrame) self._user_speaking_frame_time = time.time() async def _deprecated_handle_bot_started_speaking(self, frame: BotStartedSpeakingFrame): """Update bot speaking state when bot starts speaking.""" self._bot_speaking = True async def _deprecated_handle_bot_stopped_speaking(self, frame: BotStoppedSpeakingFrame): """Update bot speaking state when bot stops speaking.""" self._bot_speaking = False async def _deprecated_handle_user_interruption( self, vad_state: VADState, emulated: bool = False ): """Handle user interruption events based on speaking state.""" if vad_state == VADState.SPEAKING: logger.debug("User started speaking") self._user_speaking = True await self.broadcast_frame(UserStartedSpeakingFrame, emulated=emulated) # Only push InterruptionFrame if: # 1. No interruption config is set, OR # 2. Interruption config is set but bot is not speaking should_push_immediate_interruption = ( not self.interruption_strategies or not self._bot_speaking ) # Make sure we notify about interruptions quickly out-of-band. if should_push_immediate_interruption and self._allow_interruptions: await self.broadcast_interruption() elif self.interruption_strategies and self._bot_speaking: logger.debug( "User started speaking while bot is speaking with interruption config - " "deferring interruption to aggregator" ) elif vad_state == VADState.QUIET: logger.debug("User stopped speaking") self._user_speaking = False await self.broadcast_frame(UserStoppedSpeakingFrame, emulated=emulated) async def _deprecated_old_handle_vad( self, audio_frame: InputAudioRawFrame, vad_state: VADState ) -> VADState: """Handle Voice Activity Detection results and generate appropriate frames.""" new_vad_state = await self._deprecated_vad_analyze(audio_frame) if ( new_vad_state != vad_state and new_vad_state != VADState.STARTING and new_vad_state != VADState.STOPPING ): interruption_state = None # If the turn analyser is enabled, this will prevent: # - Creating the UserStoppedSpeakingFrame # - Creating the UserStartedSpeakingFrame multiple times can_create_user_frames = ( self._params.turn_analyzer is None or not self._params.turn_analyzer.speech_triggered ) if new_vad_state == VADState.SPEAKING: start_secs = ( self._params.vad_analyzer.params.start_secs if self._params.vad_analyzer else 0.0 ) await self.push_frame(VADUserStartedSpeakingFrame(start_secs=start_secs)) if can_create_user_frames: interruption_state = VADState.SPEAKING elif new_vad_state == VADState.QUIET: stop_secs = ( self._params.vad_analyzer.params.stop_secs if self._params.vad_analyzer else 0.0 ) await self.push_frame(VADUserStoppedSpeakingFrame(stop_secs=stop_secs)) if can_create_user_frames: interruption_state = VADState.QUIET if interruption_state: await self._deprecated_handle_user_interruption(interruption_state) vad_state = new_vad_state return vad_state async def _deprecated_handle_end_of_turn(self): """Handle end-of-turn analysis and generate prediction results.""" if self._params.turn_analyzer: state, prediction = await self._params.turn_analyzer.analyze_end_of_turn() await self._deprecated_handle_prediction_result(prediction) await self._deprecated_handle_end_of_turn_complete(state) async def _deprecated_handle_end_of_turn_complete(self, state: EndOfTurnState): """Handle completion of end-of-turn analysis.""" if state == EndOfTurnState.COMPLETE: await self._deprecated_handle_user_interruption(VADState.QUIET) async def _deprecated_handle_prediction_result(self, result: MetricsData): """Handle a prediction result event from the turn analyzer.""" await self.push_frame(MetricsFrame(data=[result])) async def _deprecated_run_turn_analyzer( self, frame: InputAudioRawFrame, vad_state: VADState, previous_vad_state: VADState ): """Run turn analysis on audio frame and handle results.""" is_speech = vad_state == VADState.SPEAKING or vad_state == VADState.STARTING # If silence exceeds threshold, we are going to receive EndOfTurnState.COMPLETE end_of_turn_state = self._params.turn_analyzer.append_audio(frame.audio, is_speech) if end_of_turn_state == EndOfTurnState.COMPLETE: await self._deprecated_handle_end_of_turn_complete(end_of_turn_state) # Otherwise we are going to trigger to check if the turn is completed based on the VAD elif vad_state == VADState.QUIET and vad_state != previous_vad_state: await self._deprecated_handle_end_of_turn()