import random import time import numpy as np import gradio as gr import colorsys from fastrtc import WebRTC, get_turn_credentials def video_generation_handler(prompt): num_frames = 120 width, height = 832, 480 for frame_idx in range(num_frames): hue = (frame_idx * 3) % 360 saturation = 0.7 + 0.3 * np.sin(frame_idx * 0.1) value = 0.8 + 0.2 * np.sin(frame_idx * 0.2) r, g, b = colorsys.hsv_to_rgb(hue/360, saturation, value) rgb_frame = np.full((height, width, 3), [r*255, g*255, b*255], dtype=np.uint8) noise = np.random.randint(-15, 15, (height, width, 3), dtype=np.int16) rgb_frame = np.clip(rgb_frame.astype(np.int16) + noise, 0, 255).astype(np.uint8) x_grad = np.linspace(0, 1, width) y_grad = np.linspace(0, 1, height) X, Y = np.meshgrid(x_grad, y_grad) wave = np.sin(X * 4 + frame_idx * 0.2) * np.sin(Y * 4 + frame_idx * 0.3) wave_normalized = ((wave + 1) / 2 * 40).astype(np.uint8) rgb_frame[:, :, (frame_idx // 20) % 3] = np.clip( rgb_frame[:, :, (frame_idx // 20) % 3].astype(np.int16) + wave_normalized, 0, 255 ).astype(np.uint8) bgr_frame = rgb_frame[:, :, ::-1] yield bgr_frame time.sleep(0.033) # 30 fps print("🎉 Dummy stream finished!") # --- Gradio UI Layout --- with gr.Blocks(theme=gr.themes.Soft(), title="Dummy FastRTC Demo") as demo: gr.Markdown("# 🚀 Dummy Video Generation with FastRTC Streaming") gr.Markdown("*This is a demo version that generates random colored frames instead of actual video content.*") with gr.Row(): with gr.Column(scale=2): gr.Markdown("### 1. Configure Generation") with gr.Group(): prompt = gr.Textbox( label="Prompt", placeholder="Enter any prompt to change the random seed", lines=3, value="A colorful animated demo" ) start = gr.Button("🚀 Start Generation", variant="primary") with gr.Column(scale=3): gr.Markdown("### 2. View Live Stream") gr.Markdown("Click 'Start Generation' to begin streaming frames.") webrtc_output = WebRTC( label="Video Stream", modality="video", mode="receive", height=480, rtc_configuration=get_turn_credentials(), ) webrtc_output.stream( fn=video_generation_handler, inputs=[prompt], outputs=[webrtc_output], time_limit=120, trigger=start.click ) if __name__ == "__main__": demo.queue().launch(share=True)