import os import json import gradio as gr import tempfile from PIL import Image, ImageDraw, ImageFont import cv2 from typing import Tuple, Optional import torch from pathlib import Path import time import torch import spaces import os # from huggingface_hub import HfApi, login # def setup_hf_auth(): # """ # Set up Hugging Face authentication using a token stored as a secret. # Returns authenticated HfApi instance if successful. # """ # try: # # First try to get token from environment variable # hf_token = os.getenv('HF_TOKEN') # if not hf_token: # raise ValueError("HF_TOKEN not found in environment variables or .env file") # # Login to Hugging Face # login(token=hf_token) # # Initialize API client # api = HfApi() # # Verify authentication by trying to get user info # user_info = api.whoami() # print(f"Successfully authenticated as: {user_info.name}") # return api # except Exception as e: # print(f"Authentication failed: {str(e)}") # return None from video_highlight_detector import ( load_model, BatchedVideoHighlightDetector, get_video_duration_seconds ) def load_examples(json_path: str) -> dict: with open(json_path, 'r') as f: return json.load(f) def format_duration(seconds: int) -> str: hours = seconds // 3600 minutes = (seconds % 3600) // 60 secs = seconds % 60 if hours > 0: return f"{hours}:{minutes:02d}:{secs:02d}" return f"{minutes}:{secs:02d}" def add_watermark(video_path: str, output_path: str): watermark_text = "🤗 SmolVLM2 Highlight" command = f"""ffmpeg -i {video_path} -vf \ "drawtext=text='{watermark_text}':fontcolor=white:fontsize=24:box=1:boxcolor=black@0.5:\ boxborderw=5:x=w-tw-10:y=h-th-10" \ -codec:a copy {output_path}""" os.system(command) @spaces.GPU def process_video( video_path: str, progress = gr.Progress() ) -> Tuple[str, str, str, str]: try: duration = get_video_duration_seconds(video_path) if duration > 1200: # 20 minutes return None, None, None, "Video must be shorter than 20 minutes" progress(0.1, desc="Loading model...") model, processor = load_model() detector = BatchedVideoHighlightDetector(model, processor) progress(0.2, desc="Analyzing video content...") video_description = detector.analyze_video_content(video_path) progress(0.3, desc="Determining highlight types...") highlight_types = detector.determine_highlights(video_description) progress(0.4, desc="Detecting and extracting highlights...") with tempfile.NamedTemporaryFile(suffix='.mp4', delete=False) as tmp_file: temp_output = tmp_file.name detector.create_highlight_video(video_path, temp_output) progress(0.9, desc="Adding watermark...") output_path = temp_output.replace('.mp4', '_watermark.mp4') add_watermark(temp_output, output_path) os.unlink(temp_output) video_description = video_description[:500] + "..." if len(video_description) > 500 else video_description highlight_types = highlight_types[:500] + "..." if len(highlight_types) > 500 else highlight_types return output_path, video_description, highlight_types, None except Exception as e: return None, None, None, f"Error processing video: {str(e)}" def create_ui(examples_path: str): examples_data = load_examples(examples_path) with gr.Blocks() as app: gr.Markdown("# Video Highlight Generator") gr.Markdown("Upload a video (max 20 minutes) and get an automated highlight reel!") with gr.Row(): gr.Markdown("## Example Results") with gr.Row(): for example in examples_data["examples"]: with gr.Column(): gr.Video( value=example["original"]["url"], label=f"Original ({format_duration(example['original']['duration_seconds'])})", interactive=False ) gr.Markdown(f"### {example['title']}") with gr.Column(): gr.Video( value=example["highlights"]["url"], label=f"Highlights ({format_duration(example['highlights']['duration_seconds'])})", interactive=False ) with gr.Accordion("Model chain of thought details", open=False): gr.Markdown(f"#Summary: {example['analysis']['video_description']}") gr.Markdown(f"#Highlights to search for: {example['analysis']['highlight_types']}") gr.Markdown("## Try It Yourself!") with gr.Row(): input_video = gr.Video( label="Upload your video (max 20 minutes)", interactive=True ) process_btn = gr.Button("Process Video", variant="primary") status = gr.Markdown(visible=True) with gr.Row() as results_row: with gr.Column(): video_description = gr.Markdown(visible=False) with gr.Column(): highlight_types = gr.Markdown(visible=False) with gr.Row() as output_row: output_video = gr.Video(label="Highlight Video", visible=False) download_btn = gr.Button("Download Highlights", visible=False) def on_process(video): if not video: return { status: "Please upload a video", video_description: gr.update(visible=False), highlight_types: gr.update(visible=False), output_video: gr.update(visible=False), download_btn: gr.update(visible=False) } status.value = "Processing video..." output_path, desc, highlights, err = process_video(video) if err: return { status: f"Error: {err}", video_description: gr.update(visible=False), highlight_types: gr.update(visible=False), output_video: gr.update(visible=False), download_btn: gr.update(visible=False) } return { status: "Processing complete!", video_description: gr.update(value=desc, visible=True), highlight_types: gr.update(value=highlights, visible=True), output_video: gr.update(value=output_path, visible=True), download_btn: gr.update(visible=True) } process_btn.click( on_process, inputs=[input_video], outputs=[status, video_description, highlight_types, output_video, download_btn] ) download_btn.click( lambda x: x, inputs=[output_video], outputs=[output_video] ) return app if __name__ == "__main__": # Initialize CUDA device = torch.device('cuda' if torch.cuda.is_available() else 'cpu') zero = torch.Tensor([0]).to(device) # hf_api = setup_hf_auth() # if not hf_api: # raise RuntimeError("Failed to authenticate with Hugging Face") app = create_ui("video_spec.json") app.launch()