diff --git a/changelog/3831.added.md b/changelog/3831.added.md new file mode 100644 index 000000000..7c3f7f4df --- /dev/null +++ b/changelog/3831.added.md @@ -0,0 +1 @@ +- Added custom video track support to Daily transport. Use `video_out_destinations` in `DailyParams` to publish multiple video tracks simultaneously, mirroring the existing `audio_out_destinations` feature. diff --git a/changelog/3831.changed.md b/changelog/3831.changed.md new file mode 100644 index 000000000..bb4d7df83 --- /dev/null +++ b/changelog/3831.changed.md @@ -0,0 +1 @@ +- Daily transport now uses `CustomVideoSource`/`CustomVideoTrack` instead of `VirtualCameraDevice` for the default camera output, mirroring how audio already works with `CustomAudioSource`/`CustomAudioTrack`. diff --git a/examples/foundational/56-custom-video-track.py b/examples/foundational/56-custom-video-track.py new file mode 100644 index 000000000..274ab6a5e --- /dev/null +++ b/examples/foundational/56-custom-video-track.py @@ -0,0 +1,210 @@ +# +# Copyright (c) 2024-2026, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + +"""Example demonstrating custom video tracks output with Daily transport. + +This example outputs two video track simultaneously: + - The default camera track with an animated color gradient pattern. + - A custom "blue" track with the same pattern but with a blue tint applied. + +The pattern generator pushes frames to the default camera. A second processor +(BlueTintProcessor) duplicates each frame, applies a blue tint, and pushes it +to the "blue" custom video destination. + +Run with: python examples/foundational/56-custom-video-track.py -t daily +""" + +import asyncio +import math +import time + +import numpy as np +from loguru import logger + +from pipecat.frames.frames import ( + CancelFrame, + EndFrame, + Frame, + OutputImageRawFrame, + StartFrame, + SystemFrame, +) +from pipecat.pipeline.pipeline import Pipeline +from pipecat.pipeline.runner import PipelineRunner +from pipecat.pipeline.task import PipelineTask +from pipecat.processors.frame_processor import FrameDirection, FrameProcessor +from pipecat.runner.types import RunnerArguments +from pipecat.runner.utils import create_transport +from pipecat.transports.base_transport import BaseTransport +from pipecat.transports.daily.transport import DailyCustomVideoTrackParams, DailyParams + +WIDTH = 320 +HEIGHT = 240 +FPS = 30 + +transport_params = { + "daily": lambda: DailyParams( + video_out_enabled=True, + video_out_width=WIDTH, + video_out_height=HEIGHT, + video_out_framerate=FPS, + video_out_destinations=["blue"], + custom_video_track_params={ + "blue": DailyCustomVideoTrackParams( + width=WIDTH, + height=HEIGHT, + send_settings={ + "maxQuality": "low", + "encodings": { + "low": { + "maxBitrate": 500_000, + "maxFramerate": FPS, + } + }, + }, + ), + }, + ), +} + + +def generate_gradient_frame(width: int, height: int, t: float) -> np.ndarray: + """Generate an animated gradient pattern. + + Creates a smooth color gradient that shifts over time using sine waves + for each RGB channel at different frequencies. + """ + x = np.linspace(0, 1, width) + y = np.linspace(0, 1, height) + xv, yv = np.meshgrid(x, y) + + r = ((np.sin(2 * math.pi * (xv + t * 0.3)) + 1) / 2 * 255).astype(np.uint8) + g = ((np.sin(2 * math.pi * (yv + t * 0.5)) + 1) / 2 * 255).astype(np.uint8) + b = ((np.sin(2 * math.pi * (xv + yv + t * 0.7)) + 1) / 2 * 255).astype(np.uint8) + + return np.stack([r, g, b], axis=-1) + + +class VideoPatternGenerator(FrameProcessor): + """Generates an animated gradient pattern and pushes it as video frames.""" + + def __init__(self, width: int, height: int, fps: int): + super().__init__() + self._width = width + self._height = height + self._fps = fps + self._generate_task = None + + async def process_frame(self, frame: Frame, direction: FrameDirection): + await super().process_frame(frame, direction) + + if isinstance(frame, StartFrame): + await self.push_frame(frame, direction) + await self._start() + elif isinstance(frame, (EndFrame, CancelFrame)): + await self._stop() + await self.push_frame(frame, direction) + else: + await self.push_frame(frame, direction) + + async def _start(self): + self._generate_task = self.create_task(self._generate_loop(), "video_generate_loop") + + async def _stop(self): + if self._generate_task: + await self.cancel_task(self._generate_task) + self._generate_task = None + + async def _generate_loop(self): + interval = 1.0 / self._fps + start = time.monotonic() + + while True: + t = time.monotonic() - start + + pattern = generate_gradient_frame(self._width, self._height, t) + + frame = OutputImageRawFrame( + image=pattern.tobytes(), + size=(self._width, self._height), + format="RGB", + ) + await self.push_frame(frame) + + elapsed = time.monotonic() - start - t + await asyncio.sleep(max(0, interval - elapsed)) + + +class BlueTintProcessor(FrameProcessor): + """Duplicates OutputImageRawFrames with a blue tint for a custom video destination.""" + + def __init__(self, destination: str): + super().__init__() + self._destination = destination + + async def process_frame(self, frame: Frame, direction: FrameDirection): + await super().process_frame(frame, direction) + + if isinstance(frame, OutputImageRawFrame): + # Pass through the original frame. + await self.push_frame(frame, direction) + + # Create a blue-tinted copy for the custom destination. + img = np.frombuffer(frame.image, dtype=np.uint8).reshape( + (frame.size[1], frame.size[0], 3) + ) + tinted = img.copy() + tinted[:, :, 0] = (tinted[:, :, 0] * 0.3).astype(np.uint8) # R + tinted[:, :, 1] = (tinted[:, :, 1] * 0.3).astype(np.uint8) # G + tinted[:, :, 2] = np.clip(tinted[:, :, 2].astype(np.uint16) + 80, 0, 255).astype( + np.uint8 + ) # B + + blue_frame = OutputImageRawFrame( + image=tinted.tobytes(), + size=frame.size, + format=frame.format, + ) + blue_frame.transport_destination = self._destination + await self.push_frame(blue_frame) + else: + await self.push_frame(frame, direction) + + +async def run_bot(transport: BaseTransport, runner_args: RunnerArguments): + logger.info("Starting dual video track bot") + + generator = VideoPatternGenerator(WIDTH, HEIGHT, FPS) + blue_tint = BlueTintProcessor(destination="blue") + + task = PipelineTask( + Pipeline([generator, blue_tint, transport.output()]), + idle_timeout_secs=runner_args.pipeline_idle_timeout_secs, + ) + + @transport.event_handler("on_client_connected") + async def on_client_connected(transport, client): + logger.info("Client connected") + + @transport.event_handler("on_client_disconnected") + async def on_client_disconnected(transport, client): + logger.info("Client disconnected") + await task.queue_frame(EndFrame()) + + runner = PipelineRunner(handle_sigint=runner_args.handle_sigint) + await runner.run(task) + + +async def bot(runner_args: RunnerArguments): + """Main bot entry point compatible with Pipecat Cloud.""" + transport = await create_transport(runner_args, transport_params) + await run_bot(transport, runner_args) + + +if __name__ == "__main__": + from pipecat.runner.run import main + + main() 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.