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

Replace VirtualCameraDevice with CustomVideoTrack and add custom video track support

Browse Source 
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.
This commit is contained in:
Aleix Conchillo Flaqué
2026-03-09 23:55:24 -07:00
parent 6c4a46dc79
commit 18e99123af
1 changed files with 173 additions and 24 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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.