Spaces:
Running
Running
| import gradio as gr | |
| import os | |
| import httpx | |
| from typing import Dict, Any | |
| # --- Backend Client Functions --- | |
| # These functions call the Modal/backend endpoints. | |
| async def call_video_analysis_backend(video_url: str) -> Dict[str, Any]: | |
| """Calls the backend to analyze a single video.""" | |
| # Default to a placeholder if the env var is not set, to avoid crashing. | |
| backend_url = os.getenv("BACKEND_VIDEO_URL", "https://your-backend-hf-space-for-video/process_video_analysis") | |
| if not video_url: | |
| return {"status": "error", "message": "Video URL cannot be empty."} | |
| print(f"Sending request to backend for video: {video_url}") | |
| payload = {"video_url": video_url} | |
| try: | |
| async with httpx.AsyncClient(timeout=1800.0) as client: | |
| response = await client.post(backend_url, json=payload) | |
| response.raise_for_status() | |
| return response.json() | |
| except httpx.HTTPStatusError as e: | |
| return {"status": "error", "message": f"Backend Error: {e.response.status_code}", "details": e.response.text} | |
| except Exception as e: | |
| return {"status": "error", "message": "Failed to connect to backend", "details": str(e)} | |
| async def call_topic_analysis_backend(topic: str, max_videos: int) -> Dict[str, Any]: | |
| """Calls the backend to analyze videos for a topic.""" | |
| backend_url = os.getenv("BACKEND_TOPIC_URL", "https://your-backend-hf-space-for-topic/analyze_topic") | |
| if not topic: | |
| return {"status": "error", "message": "Topic cannot be empty."} | |
| print(f"Sending request to backend for topic: {topic} ({max_videos} videos)") | |
| payload = {"topic": topic, "max_videos": max_videos} | |
| try: | |
| async with httpx.AsyncClient(timeout=3600.0) as client: | |
| response = await client.post(backend_url, json=payload) | |
| response.raise_for_status() | |
| return response.json() | |
| except httpx.HTTPStatusError as e: | |
| return {"status": "error", "message": f"Backend Error: {e.response.status_code}", "details": e.response.text} | |
| except Exception as e: | |
| return {"status": "error", "message": "Failed to connect to backend", "details": str(e)} | |
| # --- Gradio Tool Functions (Wrappers for MCP) --- | |
| async def analyze_video(video_url: str): | |
| """ | |
| Triggers a comprehensive analysis of a single video from a URL. | |
| This tool calls a backend service to perform multiple analyses: | |
| - Transcribes audio to text. | |
| - Generates a descriptive caption for the video content. | |
| - Recognizes main actions in the video. | |
| - Detects objects in keyframes. | |
| :param video_url: The public URL of the video to be processed (e.g., a YouTube link). | |
| :return: A JSON object containing the full analysis results from the backend. | |
| """ | |
| status_update = f"Analyzing video: {video_url}..." | |
| results = await call_video_analysis_backend(video_url) | |
| if isinstance(results, dict) and results.get("analysis") is None: | |
| status_update = f"Error analyzing video: {results.get('error', 'Unknown error')}" | |
| else: | |
| status_update = "Video analysis complete." | |
| return status_update, results | |
| async def analyze_topic(topic: str, max_videos: int): | |
| """ | |
| Finds and analyzes multiple videos based on a given topic. | |
| This tool calls a backend service that searches for videos related to the topic, | |
| then runs a comprehensive analysis on each video found. | |
| :param topic: The topic to search for (e.g., 'latest AI advancements'). | |
| :param max_videos: The maximum number of videos to find and analyze (1-5). | |
| :return: A JSON object with the aggregated analysis results for all videos. | |
| """ | |
| status_update = f"Analyzing topic '{topic}' with {max_videos} videos... this can take a very long time." | |
| results = await call_topic_analysis_backend(topic, max_videos) | |
| if isinstance(results, dict) and results.get("results") is None: | |
| status_update = f"Error analyzing topic: {results.get('error', 'Unknown error')}" | |
| else: | |
| status_update = "Topic analysis complete." | |
| return status_update, results | |
| # --- Gradio UI --- | |
| with gr.Blocks(theme=gr.themes.Soft()) as demo: | |
| gr.Markdown("# LLM Video Interpretation MCP") | |
| gr.Markdown("This Hugging Face Space provides tools for processing video context for AI agents. Use the tools below to analyze videos by URL or by topic.") | |
| with gr.Tab("Single Video Analysis"): | |
| gr.Markdown("## Analyze a single video from a URL") | |
| with gr.Row(): | |
| video_url_input = gr.Textbox(label="Video URL", placeholder="Enter a YouTube or direct video URL...", scale=4) | |
| submit_button = gr.Button("Analyze Video", variant="primary") | |
| status_text = gr.Textbox(label="Status", interactive=False) | |
| json_output = gr.JSON(label="Analysis Results") | |
| submit_button.click( | |
| analyze_video, | |
| inputs=[video_url_input], | |
| outputs=[status_text, json_output], | |
| api_name="analyze_video" | |
| ) | |
| gr.Examples( | |
| examples=["https://www.youtube.com/watch?v=3wLg_t_H2Xw", "https://www.youtube.com/watch?v=h42dDpgE7g8"], | |
| inputs=video_url_input, | |
| fn=analyze_video, | |
| outputs=[status_text, json_output] | |
| ) | |
| with gr.Tab("Topic Video Analysis"): | |
| gr.Markdown("## Analyze multiple videos based on a topic") | |
| with gr.Row(): | |
| topic_input = gr.Textbox(label="Enter a topic", placeholder="e.g., 'Apple Vision Pro review'", scale=3) | |
| max_videos_slider = gr.Slider(minimum=1, maximum=5, value=2, step=1, label="Number of Videos to Analyze") | |
| topic_submit_button = gr.Button("Analyze Topic", variant="primary") | |
| topic_status_text = gr.Textbox(label="Status", interactive=False) | |
| topic_json_output = gr.JSON(label="Analysis Results") | |
| topic_submit_button.click( | |
| analyze_topic, | |
| inputs=[topic_input, max_videos_slider], | |
| outputs=[topic_status_text, topic_json_output], | |
| api_name="analyze_topic" | |
| ) | |
| gr.Examples( | |
| examples=[["self-driving car technology", 2], ["open source large language models", 3]], | |
| inputs=[topic_input, max_videos_slider], | |
| fn=analyze_topic, | |
| outputs=[topic_status_text, topic_json_output] | |
| ) | |
| # Final launch of the Gradio app | |
| demo.launch() | |