DailyTransport: custom audio tracks support

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
+    """A chunk of audio. Will be played by the output transport. If the
-    transport's microphone has been enabled.
+    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
+    """An image that will be shown by the transport. If the transport supports
-    enabled.
+    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)

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,