From 6d63cff1bf175cc909b7c3b6d74e7071994e277f Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Wed, 16 Apr 2025 18:17:30 -0700 Subject: [PATCH] DailyTransport: custom audio tracks support --- src/pipecat/frames/frames.py | 49 +++++++++++++++++------ src/pipecat/transports/services/daily.py | 51 ++++++++++++++++++++++++ 2 files changed, 88 insertions(+), 12 deletions(-) diff --git a/src/pipecat/frames/frames.py b/src/pipecat/frames/frames.py index 0d7d90d78..f7c805e83 100644 --- a/src/pipecat/frames/frames.py +++ b/src/pipecat/frames/frames.py @@ -60,12 +60,16 @@ class Frame: name: str = field(init=False) pts: Optional[int] = field(init=False) metadata: Dict[str, Any] = field(init=False) + source: Optional[str] = field(init=False) + destination: Optional[str] = field(init=False) def __post_init__(self): self.id: int = obj_id() self.name: str = f"{self.__class__.__name__}#{obj_count(self)}" self.pts: Optional[int] = None self.metadata: Dict[str, Any] = {} + self.source: Optional[str] = None + self.destination: Optional[str] = None def __str__(self): return self.name @@ -136,8 +140,9 @@ class ImageRawFrame: @dataclass class OutputAudioRawFrame(DataFrame, AudioRawFrame): - """A chunk of audio. Will be played by the output transport if the - transport's microphone has been enabled. + """A chunk of audio. Will be played by the output transport. If the + transport supports multiple audio destinations (e.g. multiple audio tracks) the + destination name can be specified. """ @@ -152,8 +157,9 @@ class OutputAudioRawFrame(DataFrame, AudioRawFrame): @dataclass class OutputImageRawFrame(DataFrame, ImageRawFrame): - """An image that will be shown by the transport if the transport's camera is - enabled. + """An image that will be shown by the transport. If the transport supports + multiple video destinations (e.g. multiple video tracks) the destination + name can be specified. """ @@ -176,7 +182,7 @@ class URLImageRawFrame(OutputImageRawFrame): """ - url: Optional[str] + url: Optional[str] = None def __str__(self): pts = format_pts(self.pts) @@ -716,7 +722,11 @@ class UserImageRequestFrame(SystemFrame): @dataclass class InputAudioRawFrame(SystemFrame, AudioRawFrame): - """A chunk of audio usually coming from an input transport.""" + """A chunk of audio usually coming from an input transport. If the transport + supports multiple audio sources (e.g. multiple audio tracks) the source name + will be specified. + + """ def __post_init__(self): super().__post_init__() @@ -724,35 +734,50 @@ class InputAudioRawFrame(SystemFrame, AudioRawFrame): def __str__(self): pts = format_pts(self.pts) - return f"{self.name}(pts: {pts}, size: {len(self.audio)}, frames: {self.num_frames}, sample_rate: {self.sample_rate}, channels: {self.num_channels})" + return f"{self.name}(pts: {pts}, source: {self.source}, size: {len(self.audio)}, frames: {self.num_frames}, sample_rate: {self.sample_rate}, channels: {self.num_channels})" @dataclass class InputImageRawFrame(SystemFrame, ImageRawFrame): - """An image usually coming from an input transport.""" + """An image usually coming from an input transport. If the transport + supports multiple video sources (e.g. multiple video tracks) the source name + will be specified. + + """ def __str__(self): pts = format_pts(self.pts) - return f"{self.name}(pts: {pts}, size: {self.size}, format: {self.format})" + return f"{self.name}(pts: {pts}, source: {self.source}, size: {self.size}, format: {self.format})" + + +@dataclass +class UserAudioRawFrame(InputAudioRawFrame): + """A chunk of audio, usually coming from an input transport, associated to a user.""" + + user_id: str = "" + + def __str__(self): + pts = format_pts(self.pts) + return f"{self.name}(pts: {pts}, user: {self.user_id}, source: {self.source}, size: {len(self.audio)}, frames: {self.num_frames}, sample_rate: {self.sample_rate}, channels: {self.num_channels})" @dataclass class UserImageRawFrame(InputImageRawFrame): """An image associated to a user.""" - user_id: str + user_id: str = "" request: Optional[UserImageRequestFrame] = None def __str__(self): pts = format_pts(self.pts) - return f"{self.name}(pts: {pts}, user: {self.user_id}, size: {self.size}, format: {self.format}, request: {self.request})" + return f"{self.name}(pts: {pts}, user: {self.user_id}, source: {self.source}, size: {self.size}, format: {self.format}, request: {self.request})" @dataclass class VisionImageRawFrame(InputImageRawFrame): """An image with an associated text to ask for a description of it.""" - text: Optional[str] + text: Optional[str] = None def __str__(self): pts = format_pts(self.pts) diff --git a/src/pipecat/transports/services/daily.py b/src/pipecat/transports/services/daily.py index 23aef56ac..ed1721626 100644 --- a/src/pipecat/transports/services/daily.py +++ b/src/pipecat/transports/services/daily.py @@ -12,6 +12,7 @@ from typing import Any, Awaitable, Callable, Mapping, Optional import aiohttp from daily import ( + AudioData, VirtualCameraDevice, VirtualMicrophoneDevice, VirtualSpeakerDevice, @@ -34,6 +35,7 @@ from pipecat.frames.frames import ( TranscriptionFrame, TransportMessageFrame, TransportMessageUrgentFrame, + UserAudioRawFrame, UserImageRawFrame, UserImageRequestFrame, ) @@ -275,6 +277,7 @@ class DailyTransportClient(EventHandler): self._transport_name = transport_name self._participant_id: str = "" + self._audio_renderers = {} self._video_renderers = {} self._transcription_ids = [] self._transcription_status = None @@ -648,6 +651,20 @@ class DailyTransportClient(EventHandler): if self._joined and self._transcription_status: await self.update_transcription(self._transcription_ids) + async def capture_participant_audio( + self, + participant_id: str, + callback: Callable, + audio_source: str = "camera", + ): + self._audio_renderers[participant_id] = callback + + self._client.set_audio_renderer( + participant_id, + self._audio_data_received, + audio_source=audio_source, + ) + async def capture_participant_video( self, participant_id: str, @@ -773,6 +790,10 @@ class DailyTransportClient(EventHandler): # Daily (CallClient callbacks) # + def _audio_data_received(self, participant_id, audio_data): + callback = self._audio_renderers[participant_id] + self._call_async_callback(callback, participant_id, audio_data) + def _video_frame_received(self, participant_id, video_frame): callback = self._video_renderers[participant_id] self._call_async_callback( @@ -828,6 +849,7 @@ class DailyInputTransport(BaseInputTransport): self._client = client self._params = params + self._audio_renderers = {} self._video_renderers = {} # Whether we have seen a StartFrame already. @@ -916,6 +938,27 @@ class DailyInputTransport(BaseInputTransport): # Audio in # + async def capture_participant_audio( + self, + participant_id: str, + audio_source: str = "camera", + ): + self._audio_renderers[participant_id] = {"audio_source": audio_source} + + await self._client.capture_participant_audio( + participant_id, self._on_participant_audio_data, audio_source + ) + + async def _on_participant_audio_data(self, participant_id: str, audio: AudioData): + frame = UserAudioRawFrame( + user_id=participant_id, + audio_source=self._audio_renderers[participant_id]["audio_source"], + audio=audio.audio_frames, + sample_rate=audio.sample_rate, + num_channels=audio.num_channels, + ) + await self.push_frame(frame) + async def _audio_in_task_handler(self): while True: frame = await self._client.read_next_audio_frame() @@ -1204,6 +1247,14 @@ class DailyTransport(BaseTransport): async def capture_participant_transcription(self, participant_id: str): await self._client.capture_participant_transcription(participant_id) + async def capture_participant_audio( + self, + participant_id: str, + audio_source: str = "microphone", + ): + if self._input: + await self._input.capture_participant_audio(participant_id, audio_source) + async def capture_participant_video( self, participant_id: str,