wx44wx/pipecat
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge pull request #3831 from pipecat-ai/aleix/custom-video-tracks

Browse Source 
Replace VirtualCameraDevice with CustomVideoTrack + custom video track support
This commit is contained in:
Aleix Conchillo Flaqué
2026-03-10 11:44:29 -07:00
committed by GitHub
parent 6c4a46dc79 86597cc9ec
commit 743e2ac277
4 changed files with 385 additions and 24 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
changelog/3831.added.md Normal file
View File

@@ -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.

1
changelog/3831.changed.md Normal file
View File

@@ -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`.

210
examples/foundational/56-custom-video-track.py Normal file
View File

@@ -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()

197
src/pipecat/transports/daily/transport.py
View File

@@ -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.