From 18e99123af3f171b43623f621101470fc0e24f89 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Mon, 9 Mar 2026 23:55:24 -0700 Subject: [PATCH] Replace VirtualCameraDevice with CustomVideoTrack and add custom video track support Use CustomVideoSource/CustomVideoTrack for the default camera output instead of VirtualCameraDevice, mirroring how audio already uses CustomAudioSource/CustomAudioTrack. Add support for custom video destinations (register_video_destination, add/remove custom video tracks, routing in write_video_frame) so multiple video tracks can be published simultaneously. --- src/pipecat/transports/daily/transport.py | 197 +++++++++++++++++++--- 1 file changed, 173 insertions(+), 24 deletions(-) diff --git a/src/pipecat/transports/daily/transport.py b/src/pipecat/transports/daily/transport.py index 421e5db8e..1ded5ded7 100644 --- a/src/pipecat/transports/daily/transport.py +++ b/src/pipecat/transports/daily/transport.py @@ -57,10 +57,11 @@ try: CallClient, CustomAudioSource, CustomAudioTrack, + CustomVideoSource, + CustomVideoTrack, Daily, EventHandler, VideoFrame, - VirtualCameraDevice, VirtualSpeakerDevice, ) from daily import LogLevel as DailyLogLevel @@ -304,6 +305,44 @@ class DailyTranscriptionSettings(BaseModel): extra: Mapping[str, Any] = {"interim_results": True} +class DailyCustomVideoTrackParams(BaseModel): + """Configuration for a custom video track. + + If ``send_settings`` is not provided, the track will use the default video + publishing settings (framerate, bitrate, codec, etc.). + + Parameters: + width: Video width in pixels. + height: Video height in pixels. + color_format: Video color format (e.g., "RGB", "RGBA", "BGRA"). + send_settings: Optional Daily sendSettings dict for this track. + See https://reference-python.daily.co/types.html#videopublishingsettings + """ + + width: int = 1024 + height: int = 768 + color_format: str = "RGB" + send_settings: Optional[Dict[str, Any]] = None + + +class DailyCustomAudioTrackParams(BaseModel): + """Configuration for a custom audio track. + + If ``send_settings`` is not provided, the track will use the default audio + publishing settings (bitrate, channel config, etc.). + + Parameters: + sample_rate: Audio sample rate in Hz. Defaults to transport's output sample rate. + channels: Number of audio channels. + send_settings: Optional Daily sendSettings dict for this track. + See https://reference-python.daily.co/types.html#audiopublishingsettings + """ + + sample_rate: Optional[int] = None + channels: int = 1 + send_settings: Optional[Dict[str, Any]] = None + + class DailyParams(TransportParams): """Configuration parameters for Daily transport. @@ -311,8 +350,10 @@ class DailyParams(TransportParams): api_url: Daily API base URL. api_key: Daily API authentication key. audio_in_user_tracks: Receive users' audio in separate tracks - dialin_settings: Optional settings for dial-in functionality. camera_out_enabled: Whether to enable the main camera output track. + custom_audio_track_params: Per-destination configuration for custom audio tracks. + custom_video_track_params: Per-destination configuration for custom video tracks. + dialin_settings: Optional settings for dial-in functionality. microphone_out_enabled: Whether to enable the main microphone track. transcription_enabled: Whether to enable speech transcription. transcription_settings: Configuration for transcription service. @@ -321,8 +362,10 @@ class DailyParams(TransportParams): api_url: str = "https://api.daily.co/v1" api_key: str = "" audio_in_user_tracks: bool = True - dialin_settings: Optional[DailyDialinSettings] = None camera_out_enabled: bool = True + custom_audio_track_params: Optional[Mapping[str, DailyCustomAudioTrackParams]] = None + custom_video_track_params: Optional[Mapping[str, DailyCustomVideoTrackParams]] = None + dialin_settings: Optional[DailyDialinSettings] = None microphone_out_enabled: bool = True transcription_enabled: bool = False transcription_settings: DailyTranscriptionSettings = DailyTranscriptionSettings() @@ -430,6 +473,19 @@ class DailyAudioTrack: track: CustomAudioTrack +@dataclass +class DailyVideoTrack: + """Container for Daily video track components. + + Parameters: + source: The custom video source for the track. + track: The custom video track instance. + """ + + source: CustomVideoSource + track: CustomVideoTrack + + # This is just a type alias for the errors returned by daily-python. Right now # they are just a string. CallClientError = str @@ -519,14 +575,11 @@ class DailyTransportClient(EventHandler): self._in_sample_rate = 0 self._out_sample_rate = 0 - self._camera: Optional[VirtualCameraDevice] = None self._speaker: Optional[VirtualSpeakerDevice] = None + self._camera_track: Optional[DailyVideoTrack] = None self._microphone_track: Optional[DailyAudioTrack] = None self._custom_audio_tracks: Dict[str, DailyAudioTrack] = {} - - def _camera_name(self): - """Generate a unique virtual camera name for this client instance.""" - return f"camera-{self}" + self._custom_video_tracks: Dict[str, DailyVideoTrack] = {} def _speaker_name(self): """Generate a unique virtual speaker name for this client instance.""" @@ -625,8 +678,29 @@ class DailyTransportClient(EventHandler): Args: destination: The destination identifier to register. """ - self._custom_audio_tracks[destination] = await self.add_custom_audio_track(destination) - self._client.update_publishing({"customAudio": {destination: True}}) + params = (self._params.custom_audio_track_params or {}).get(destination) + self._custom_audio_tracks[destination] = await self.add_custom_audio_track( + destination, params=params + ) + publishing: Dict[str, Any] = {"customAudio": {destination: True}} + if params and params.send_settings: + publishing["customAudio"][destination] = {"sendSettings": params.send_settings} + self._client.update_publishing(publishing) + + async def register_video_destination(self, destination: str): + """Register a custom video destination for multi-track output. + + Args: + destination: The destination identifier to register. + """ + params = (self._params.custom_video_track_params or {}).get(destination) + self._custom_video_tracks[destination] = await self.add_custom_video_track( + destination, params=params + ) + publishing: Dict[str, Any] = {"customVideo": {destination: True}} + if params and params.send_settings: + publishing["customVideo"][destination] = {"sendSettings": params.send_settings} + self._client.update_publishing(publishing) async def write_audio_frame(self, frame: OutputAudioRawFrame) -> bool: """Write an audio frame to the appropriate audio track. @@ -657,7 +731,7 @@ class DailyTransportClient(EventHandler): return num_frames > 0 async def write_video_frame(self, frame: OutputImageRawFrame) -> bool: - """Write a video frame to the camera device. + """Write a video frame to the appropriate video track. Args: frame: The image frame to write. @@ -665,10 +739,20 @@ class DailyTransportClient(EventHandler): Returns: True if the video frame was written successfully, False otherwise. """ - if not frame.transport_destination and self._camera: - self._camera.write_frame(frame.image) + destination = frame.transport_destination + video_source: Optional[CustomVideoSource] = None + if not destination and self._camera_track: + video_source = self._camera_track.source + elif destination and destination in self._custom_video_tracks: + track = self._custom_video_tracks[destination] + video_source = track.source + + if video_source: + video_source.write_frame(frame.image) return True - return False + else: + logger.warning(f"{self} unable to write video frames to destination [{destination}]") + return False async def setup(self, setup: FrameProcessorSetup): """Setup the client with task manager and event queues. @@ -733,13 +817,14 @@ class DailyTransportClient(EventHandler): self._callback_task_handler(self._video_queue), f"{self}::video_callback_task", ) - if self._params.video_out_enabled and not self._camera: - self._camera = Daily.create_camera_device( - self._camera_name(), - width=self._params.video_out_width, - height=self._params.video_out_height, - color_format=self._params.video_out_color_format, + if self._params.video_out_enabled and not self._camera_track: + video_source = CustomVideoSource( + self._params.video_out_width, + self._params.video_out_height, + self._params.video_out_color_format, ) + video_track = CustomVideoTrack(video_source) + self._camera_track = DailyVideoTrack(source=video_source, track=video_track) if self._params.audio_out_enabled and not self._microphone_track: audio_source = CustomAudioSource(self._out_sample_rate, self._params.audio_out_channels) @@ -809,7 +894,11 @@ class DailyTransportClient(EventHandler): "camera": { "isEnabled": camera_enabled, "settings": { - "deviceId": self._camera_name(), + "customTrack": { + "id": self._camera_track.track.id + if self._camera_track + else "no-camera-track" + } }, }, "microphone": { @@ -874,6 +963,8 @@ class DailyTransportClient(EventHandler): # Remove any custom tracks, if any. for track_name, _ in self._custom_audio_tracks.items(): await self.remove_custom_audio_track(track_name) + for track_name, _ in self._custom_video_tracks.items(): + await self.remove_custom_video_track(track_name) error = await self._leave() if not error: @@ -1173,18 +1264,26 @@ class DailyTransportClient(EventHandler): color_format=color_format, ) - async def add_custom_audio_track(self, track_name: str) -> DailyAudioTrack: + async def add_custom_audio_track( + self, + track_name: str, + params: Optional[DailyCustomAudioTrackParams] = None, + ) -> DailyAudioTrack: """Add a custom audio track for multi-stream output. Args: track_name: Name for the custom audio track. + params: Optional per-track configuration for sample rate, channels, and sendSettings. Returns: The created DailyAudioTrack instance. """ future = self._get_event_loop().create_future() - audio_source = CustomAudioSource(self._out_sample_rate, 1) + sample_rate = params.sample_rate if params and params.sample_rate else self._out_sample_rate + channels = params.channels if params else 1 + + audio_source = CustomAudioSource(sample_rate, channels) audio_track = CustomAudioTrack(audio_source) @@ -1217,6 +1316,56 @@ class DailyTransportClient(EventHandler): ) return await future + async def add_custom_video_track( + self, + track_name: str, + params: Optional[DailyCustomVideoTrackParams] = None, + ) -> DailyVideoTrack: + """Add a custom video track for multi-stream output. + + Args: + track_name: Name for the custom video track. + params: Optional per-track configuration for dimensions, color format, and sendSettings. + + Returns: + The created DailyVideoTrack instance. + """ + future = self._get_event_loop().create_future() + + width = params.width if params else self._params.video_out_width + height = params.height if params else self._params.video_out_height + color_format = params.color_format if params else self._params.video_out_color_format + + video_source = CustomVideoSource(width, height, color_format) + + video_track = CustomVideoTrack(video_source) + + self._client.add_custom_video_track( + track_name=track_name, + video_track=video_track, + completion=completion_callback(future), + ) + + await future + + return DailyVideoTrack(source=video_source, track=video_track) + + async def remove_custom_video_track(self, track_name: str) -> Optional[CallClientError]: + """Remove a custom video track. + + Args: + track_name: Name of the custom video track to remove. + + Returns: + error: An error description or None. + """ + future = self._get_event_loop().create_future() + self._client.remove_custom_video_track( + track_name=track_name, + completion=completion_callback(future), + ) + return await future + async def update_transcription( self, participants=None, instance_id=None ) -> Optional[CallClientError]: @@ -1985,7 +2134,7 @@ class DailyOutputTransport(BaseOutputTransport): Args: destination: The destination identifier to register. """ - logger.warning(f"{self} registering video destinations is not supported yet") + await self._client.register_video_destination(destination) async def register_audio_destination(self, destination: str): """Register an audio output destination.