From 14dd028b8f0a751135bfcd490ad5af83398766e7 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Mon, 9 Mar 2026 23:55:28 -0700 Subject: [PATCH] Add custom video track example with per-track params --- .../foundational/56-custom-video-track.py | 210 ++++++++++++++++++ 1 file changed, 210 insertions(+) create mode 100644 examples/foundational/56-custom-video-track.py 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()