app.py

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 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}"


@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, batch_size=16)

        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:
            output_path = tmp_file.name
        
        detector.create_highlight_video(video_path, output_path)
        
        # progress(0.9, desc="Adding watermark...")
        # output_path = temp_output.replace('.mp4', '_watermark.mp4')
        # add_watermark(temp_output, output_path)
        
        os.unlink(output_path)
        progress(1.0, desc="Complete!")

        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']}")

        # Main interface section
        gr.Markdown("## Try It Yourself!")
        with gr.Row():
            # Left column: Upload and Process
            with gr.Column(scale=1):
                input_video = gr.Video(
                    label="Upload your video (max 20 minutes)",
                    interactive=True,
                    max_length = 1200
                )
                process_btn = gr.Button("Process Video", variant="primary")
                status = gr.Markdown(visible=True)
            
            # Right column: Progress, Results and Analysis
            with gr.Column(scale=1):
                gr.Progress()
                
                # Output video (initially hidden)
                output_video = gr.Video(
                    label="Highlight Video",
                    visible=False,
                    interactive=False,
                )
                
                # Analysis accordion
                with gr.Accordion("Model chain of thought details", open=True, visible=False) as analysis_accordion:
                    video_description = gr.Markdown(visible=True)
                    highlight_types = gr.Markdown(visible=True)

        def on_process(video, progress=gr.Progress()):
            if not video:
                return {
                    status: "Please upload a video",
                    output_video: gr.update(visible=False),
                    analysis_accordion: gr.update(visible=False),
                }
            
            status.value = "Processing video..."
            output_path, desc, highlights, err = process_video(video, progress=progress)
            
            if err:
                return {
                    status: f"Error: {err}",
                    output_video: gr.update(visible=False),
                    analysis_accordion: gr.update(visible=False),
                }
            
            # Format the analysis text
            desc = f"#Summary: {desc[:500] + '...' if len(desc) > 500 else desc}"
            highlights = f"#Highlights to search for: {highlights[:500] + '...' if len(highlights) > 500 else highlights}"
            
            return {
                status: "Processing complete!",
                output_video: gr.update(value=output_path, visible=True),
                analysis_accordion: gr.update(visible=True),
                video_description: desc,
                highlight_types: highlights,
            }

        process_btn.click(
            on_process,
            inputs=[input_video],
            outputs=[
                status,
                output_video,
                analysis_accordion,
                video_description,
                highlight_types,
            ]
        )

    return app
    #     gr.Markdown("## Try It Yourself!")
    #     with gr.Row():
    #         input_video = gr.Video(
    #             label="Upload your video (max 20 minutes)",
    #             interactive=True
    #         )
        
    #     gr.Progress()
    #     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, progress=gr.Progress()):
    #         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, progress=progress)
            
    #         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)

    app = create_ui("video_spec.json")
    app.launch()