|
|
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) |
|
|
|
|
|
print("🎉 Dummy stream finished!") |
|
|
|
|
|
|
|
|
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) |
