diff --git a/CHANGELOG.md b/CHANGELOG.md index bef74dc06..9f9cd7e0c 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,15 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Added +- Added a new `audio_in_stream_on_start` field to `TransportParams`. + +- Added a new method `start_audio_in_streaming` in the `BaseInputTransport`. + - This method should be used to start receiving the input audio in case the field `audio_in_stream_on_start` is set to `false`. + +- Added support for the `RTVIProcessor` to handle buffered audio in `base64` format, converting it into InputAudioRawFrame for transport. + +- Added support for the `RTVIProcessor` to trigger `start_audio_in_streaming` only after the `client-ready` message. + - Added new `MUTE_UNTIL_FIRST_BOT_COMPLETE` strategy to `STTMuteStrategy`. This strategy starts muted and remains muted until the first bot speech completes, ensuring the bot's first response cannot be interrupted. This complements the @@ -36,6 +45,12 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Changed +- Updated `DailyTransport` to respect the `audio_in_stream_on_start` field, ensuring it only starts receiving the audio input if it is enabled. + +- Updated `FastAPIWebsocketOutputTransport` to send `TransportMessageFrame` and `TransportMessageUrgentFrame` to the serializer. + +- Updated `WebsocketServerOutputTransport` to send `TransportMessageFrame` and `TransportMessageUrgentFrame` to the serializer. + - Enhanced `STTMuteConfig` to validate strategy combinations, preventing `MUTE_UNTIL_FIRST_BOT_COMPLETE` and `FIRST_SPEECH` from being used together as they handle first bot speech differently. diff --git a/src/pipecat/transports/base_input.py b/src/pipecat/transports/base_input.py index a44332a65..6bfe86001 100644 --- a/src/pipecat/transports/base_input.py +++ b/src/pipecat/transports/base_input.py @@ -47,6 +47,13 @@ class BaseInputTransport(FrameProcessor): # if passthrough is enabled. self._audio_task = None + def enable_audio_in_stream_on_start(self, enabled: bool) -> None: + logger.debug(f"Enabling audio on start. {enabled}") + self._params.audio_in_stream_on_start = enabled + + def start_audio_in_streaming(self): + pass + @property def sample_rate(self) -> int: return self._sample_rate diff --git a/src/pipecat/transports/base_transport.py b/src/pipecat/transports/base_transport.py index 436e78b94..e07b41c22 100644 --- a/src/pipecat/transports/base_transport.py +++ b/src/pipecat/transports/base_transport.py @@ -39,6 +39,7 @@ class TransportParams(BaseModel): audio_in_sample_rate: Optional[int] = None audio_in_channels: int = 1 audio_in_filter: Optional[BaseAudioFilter] = None + audio_in_stream_on_start: bool = True vad_enabled: bool = False vad_audio_passthrough: bool = False vad_analyzer: Optional[VADAnalyzer] = None diff --git a/src/pipecat/transports/services/daily.py b/src/pipecat/transports/services/daily.py index 99071cf5c..2274b3e8f 100644 --- a/src/pipecat/transports/services/daily.py +++ b/src/pipecat/transports/services/daily.py @@ -5,7 +5,6 @@ # import asyncio -import base64 import time import warnings from concurrent.futures import ThreadPoolExecutor @@ -839,6 +838,13 @@ class DailyInputTransport(BaseInputTransport): def vad_analyzer(self) -> Optional[VADAnalyzer]: return self._vad_analyzer + def start_audio_in_streaming(self): + # Create audio task. It reads audio frames from Daily and push them + # internally for VAD processing. + if self._params.audio_in_enabled or self._params.vad_enabled: + logger.debug(f"Start receiving audio") + self._audio_in_task = self.create_task(self._audio_in_task_handler()) + async def start(self, frame: StartFrame): # Parent start. await super().start(frame) @@ -849,10 +855,8 @@ class DailyInputTransport(BaseInputTransport): # Inialize WebRTC VAD if needed. if self._params.vad_enabled and not self._params.vad_analyzer: self._vad_analyzer = WebRTCVADAnalyzer(sample_rate=self.sample_rate) - # Create audio task. It reads audio frames from Daily and push them - # internally for VAD processing. - if self._params.audio_in_enabled or self._params.vad_enabled: - self._audio_in_task = self.create_task(self._audio_in_task_handler()) + if self._params.audio_in_stream_on_start: + self.start_audio_in_streaming() async def stop(self, frame: EndFrame): # Parent stop. @@ -1200,22 +1204,7 @@ class DailyTransport(BaseTransport): async def _on_app_message(self, message: Any, sender: str): if self._input: - if message["type"] in {"raw-audio", "raw-audio-batch"}: - data = message["data"] - audio_list = data.get( - "base64AudioBatch", [data.get("base64Audio")] - ) # Ensure a list - - for base64_audio in filter(None, audio_list): # Filter out None values - pcm_bytes = base64.b64decode(base64_audio) - frame = InputAudioRawFrame( - audio=pcm_bytes, - sample_rate=data["sampleRate"], - num_channels=data["numChannels"], - ) - await self._input.push_audio_frame(frame) - else: - await self._input.push_app_message(message, sender) + await self._input.push_app_message(message, sender) await self._call_event_handler("on_app_message", message, sender) async def _on_call_state_updated(self, state: str):