Spaces:
Runtime error
Runtime error
| """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() | |