"""Video Intelligence Chat Platform — main Gradio application.""" import logging import os import gradio as gr from dotenv import load_dotenv from utils.context_builder import build_chat_system_prompt, build_video_context from utils.llm_chat import chat_with_video from utils.transcriber import transcribe_audio from utils.video_processor import extract_audio, extract_frames, get_video_metadata from utils.vision_analyzer import analyze_frames_batch load_dotenv() logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(name)s: %(message)s") logger = logging.getLogger(__name__) MAX_VIDEO_DURATION = int(os.environ.get("MAX_VIDEO_DURATION", 600)) FRAME_INTERVAL = int(os.environ.get("FRAME_INTERVAL", 5)) _API_KEY_WARNING = ( "**NVIDIA_API_KEY is not set.** " "Get a free key at [build.nvidia.com](https://build.nvidia.com) " "and add it as a Space secret named `NVIDIA_API_KEY`." ) def _has_api_key() -> bool: return bool(os.environ.get("NVIDIA_API_KEY")) def _fmt_time(seconds: float) -> str: m = int(seconds // 60) s = int(seconds % 60) return f"{m:02d}:{s:02d}" def process_video(video_path: str, progress=gr.Progress(track_tqdm=True)): """Run the full video analysis pipeline and return UI update values.""" if not video_path: raise gr.Error("Please upload a video file before analyzing.") if not _has_api_key(): raise gr.Error("NVIDIA_API_KEY is missing. See the banner above for instructions.") try: # Step 2: metadata progress(0.1, desc="Getting video metadata...") yield _step_update("Getting video metadata...", None, None, None, None, None, True) metadata = get_video_metadata(video_path) if metadata["duration_seconds"] > MAX_VIDEO_DURATION: logger.warning("Video is %.0fs, exceeds limit of %ds", metadata["duration_seconds"], MAX_VIDEO_DURATION) yield _step_update( f"Warning: video is {metadata['duration_seconds']:.0f}s " f"(limit is {MAX_VIDEO_DURATION}s). Processing anyway — consider a shorter clip.", None, None, None, None, None, True, ) # Step 3: frames progress(0.2, desc="Extracting frames...") yield _step_update("Extracting frames...", None, None, None, None, None, True) frames = extract_frames(video_path, FRAME_INTERVAL) # Step 4: audio progress(0.4, desc="Extracting audio...") yield _step_update("Extracting audio...", None, None, None, None, None, True) audio_path = extract_audio(video_path) # Step 5: transcription progress(0.6, desc="Transcribing audio...") yield _step_update("Transcribing audio with Whisper...", None, None, None, None, None, True) transcript = transcribe_audio(audio_path) # Step 6: vision analysis (progress updates per frame) frame_analyses = [] n = len(frames) for i, frame in enumerate(frames): pct = 0.6 + 0.2 * ((i + 1) / max(n, 1)) progress(pct, desc=f"Analyzing frame {i + 1}/{n} with AI...") yield _step_update( f"Analyzing frames with AI... ({i + 1}/{n})", None, None, None, None, None, True, ) results = analyze_frames_batch([frame]) frame_analyses.extend(results) # Step 7: build context progress(1.0, desc="Building context...") yield _step_update("Building video context...", None, None, None, None, None, True) video_context = build_video_context(metadata, frame_analyses, transcript) system_prompt = build_chat_system_prompt(video_context) # Step 8: done word_count = sum(len(s["text"].split()) for s in transcript) meta_summary = ( f"Duration: {_fmt_time(metadata['duration_seconds'])} | " f"Resolution: {metadata['width']}x{metadata['height']} | " f"FPS: {metadata['fps']:.1f}\n" f"Frames analyzed: {len(frame_analyses)} | " f"Transcript words: {word_count}" ) context_preview = video_context[:500] + ("..." if len(video_context) > 500 else "") yield _step_update( "Ready to chat!", video_context, system_prompt, True, meta_summary, context_preview, False, ) except gr.Error: raise except Exception as exc: logger.exception("Processing pipeline failed") raise gr.Error(f"Processing failed: {exc}") from exc def _step_update(status, video_ctx, sys_prompt, processing, meta_text, ctx_preview, analyze_btn_interactive): return ( status, video_ctx, sys_prompt, processing, meta_text or "", ctx_preview or "", gr.update(interactive=not analyze_btn_interactive), gr.update(interactive=bool(processing is True and video_ctx is None) is False and processing is True), ) def send_message(message: str, history: list, video_context: str, system_prompt: str, processing_complete: bool): """Stream a chat response given the current conversation state.""" if not processing_complete: raise gr.Error("Please analyze a video first before chatting.") if not message.strip(): return "", history history = history or [] history.append([message, ""]) full_response = "" try: for chunk in chat_with_video(message, history[:-1], system_prompt): full_response += chunk history[-1][1] = full_response yield "", history except Exception as exc: logger.exception("Chat error") history[-1][1] = f"Error: {exc}" yield "", history def use_starter(question: str): """Return a starter question as the textbox value.""" return question def build_app() -> gr.Blocks: api_key_present = _has_api_key() with gr.Blocks(title="Video Intelligence Chat", theme=gr.themes.Soft()) as demo: # State video_context_state = gr.State(None) system_prompt_state = gr.State("") processing_complete_state = gr.State(False) # Warning banner if not api_key_present: gr.Markdown(f"> {_API_KEY_WARNING}") gr.Markdown("# Video Intelligence Chat") gr.Markdown("Upload a video, analyze it, then ask anything about its content.") with gr.Row(): # Left panel with gr.Column(scale=1): gr.Markdown("### Upload & Analyze") video_input = gr.Video(label="Upload Video (MP4, MOV, AVI, WebM — max ~100MB)") analyze_btn = gr.Button("Analyze Video", variant="primary") status_box = gr.Textbox(label="Status", interactive=False, lines=2) with gr.Accordion("Video Details", open=False): meta_text = gr.Textbox(label="Metadata", interactive=False, lines=3) ctx_preview = gr.Textbox(label="Context Preview (first 500 chars)", interactive=False, lines=6) # Right panel with gr.Column(scale=1): gr.Markdown("### Chat About the Video") chatbot = gr.Chatbot(height=500, bubble_full_width=False) with gr.Row(): msg_input = gr.Textbox( placeholder="Ask anything about the video...", label="Your message", scale=4, interactive=False, ) send_btn = gr.Button("Send", variant="primary", scale=1, interactive=False) clear_btn = gr.Button("Clear Chat") gr.Markdown("**Suggested questions:**") with gr.Row(): q1 = gr.Button("What is this video about?", size="sm") q2 = gr.Button("Who appears in this video?", size="sm") with gr.Row(): q3 = gr.Button("Summarize the key moments", size="sm") q4 = gr.Button("What is being said?", size="sm") # Wire up processing pipeline analyze_btn.click( fn=process_video, inputs=[video_input], outputs=[ status_box, video_context_state, system_prompt_state, processing_complete_state, meta_text, ctx_preview, analyze_btn, send_btn, ], ) # Enable msg_input when processing completes processing_complete_state.change( fn=lambda done: gr.update(interactive=done), inputs=[processing_complete_state], outputs=[msg_input], ) # Chat submit send_btn.click( fn=send_message, inputs=[msg_input, chatbot, video_context_state, system_prompt_state, processing_complete_state], outputs=[msg_input, chatbot], ) msg_input.submit( fn=send_message, inputs=[msg_input, chatbot, video_context_state, system_prompt_state, processing_complete_state], outputs=[msg_input, chatbot], ) # Clear chat clear_btn.click(fn=lambda: [], outputs=[chatbot]) # Starter questions for btn, question in [(q1, "What is this video about?"), (q2, "Who appears in this video?"), (q3, "Summarize the key moments"), (q4, "What is being said?")]: btn.click(fn=lambda q=question: q, outputs=[msg_input]) return demo if __name__ == "__main__": app = build_app() app.launch()