#!/usr/bin/env python3 """ NeuroAnim Gradio Web Interface A comprehensive web UI for generating educational STEM animations with: - Topic input and configuration - Real-time progress tracking - Video preview and download - Generated content display (narration, code, quiz) - Error handling and logging """ import asyncio import logging import os from datetime import datetime from pathlib import Path from typing import Any, Dict, Optional, Tuple import gradio as gr from dotenv import load_dotenv from orchestrator import NeuroAnimOrchestrator load_dotenv() # Set up logging logging.basicConfig( level=logging.INFO, format="%(asctime)s - %(name)s - %(levelname)s - %(message)s" ) logger = logging.getLogger(__name__) def format_quiz_markdown(quiz_text: str) -> str: """Format quiz text into a nice markdown display.""" if not quiz_text or quiz_text == "Not available": return "ā No quiz generated yet." # If it's already formatted or looks good, return as is with some styling formatted = f"## š Assessment Questions\n\n{quiz_text}" # Try to add some structure if it's plain text lines = quiz_text.split("\n") formatted_lines = [] question_num = 0 for line in lines: line = line.strip() if not line: formatted_lines.append("") continue # Detect question patterns if line.lower().startswith(("q:", "question", "q.", f"{question_num + 1}.")): question_num += 1 formatted_lines.append(f"\n### Question {question_num}") # Remove the question prefix clean_line = line.split(":", 1)[-1].strip() if ":" in line else line formatted_lines.append(f"**{clean_line}**\n") elif line.lower().startswith(("a)", "b)", "c)", "d)", "a.", "b.", "c.", "d.")): # Format multiple choice options formatted_lines.append(f"- {line}") elif line.lower().startswith(("answer:", "a:", "correct:")): # Format answers formatted_lines.append(f"\n> ā {line}\n") else: formatted_lines.append(line) # If we detected structure, use the formatted version if question_num > 0: return "## š Assessment Questions\n\n" + "\n".join(formatted_lines) # Otherwise return with basic formatting return formatted class NeuroAnimApp: """Main application class for Gradio interface.""" def __init__(self): self.orchestrator: Optional[NeuroAnimOrchestrator] = None self.current_task: Optional[asyncio.Task] = None self.is_generating = False self.event_loop: Optional[asyncio.AbstractEventLoop] = None self.current_progress = None # Store progress callback for dynamic updates async def initialize_orchestrator(self): """Initialize the orchestrator if not already done.""" if self.orchestrator is None: self.orchestrator = NeuroAnimOrchestrator() await self.orchestrator.initialize() logger.info("Orchestrator initialized successfully") async def cleanup_orchestrator(self): """Clean up orchestrator resources.""" if self.orchestrator is not None: await self.orchestrator.cleanup() self.orchestrator = None logger.info("Orchestrator cleaned up") def cleanup_event_loop(self): """Clean up the event loop on application shutdown.""" if self.event_loop is not None and not self.event_loop.is_closed(): self.event_loop.close() self.event_loop = None logger.info("Event loop closed") async def generate_animation_async( self, topic: str, audience: str, duration: float, quality: str, progress=gr.Progress() ) -> Dict[str, Any]: """ Generate animation with progress tracking. Args: topic: STEM topic to animate audience: Target audience level duration: Animation duration in minutes quality: Video quality (low, medium, high, production_quality) progress: Gradio progress tracker Returns: Results dictionary with generated content """ try: self.is_generating = True # Validate inputs if not topic or len(topic.strip()) < 3: return { "success": False, "error": "Please provide a valid topic (at least 3 characters)", } if duration < 0.5 or duration > 10: return { "success": False, "error": "Duration must be between 0.5 and 10 minutes", } # Initialize orchestrator progress(0.05, desc="Initializing system...") await self.initialize_orchestrator() # Generate unique filename timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") safe_topic = "".join(c if c.isalnum() else "_" for c in topic)[:30] output_filename = f"{safe_topic}_{timestamp}.mp4" # Map quality from UI to orchestrator format quality_map = { "Low (480p, faster)": "low", "Medium (720p, balanced)": "medium", "High (1080p, slower)": "high", "Production (4K, slowest)": "production_quality", } quality_param = quality_map.get(quality, "medium") # Map audience from UI to orchestrator format audience_map = { "elementary": "elementary", "middle_school": "middle_school", "high_school": "high_school", "undergraduate": "college", # Map to 'college' for LLM compatibility "phd": "graduate", # Map to 'graduate' for LLM compatibility "general": "general", } audience_param = audience_map.get(audience, audience) # Dynamic progress tracking with step-based updates step_times = {} # Track step start times step_index = [0] # Current step index steps = [ (0.1, "Planning concept"), (0.25, "Generating narration script"), (0.40, "Creating Manim animation code"), (0.55, "Rendering animation video"), (0.75, "Generating audio narration"), (0.90, "Merging video and audio"), (0.95, "Creating quiz questions"), ] import time def progress_callback(step_name: str, step_progress: float): """Callback for orchestrator to report progress.""" # Find matching step for idx, (prog, desc) in enumerate(steps): if desc.lower() in step_name.lower(): step_index[0] = idx # Track timing current_time = time.time() if step_name not in step_times: step_times[step_name] = current_time elapsed = current_time - step_times[step_name] # Add timing info for long steps if elapsed > 30: # Show message if step takes more than 30s desc_with_time = f"{desc} (taking longer than usual, please wait...)" else: desc_with_time = f"{desc}..." progress(prog, desc=desc_with_time) return # If no match, use the provided progress directly progress(step_progress, desc=f"{step_name}...") # Start generation with dynamic progress result = await self.orchestrator.generate_animation( topic=topic, target_audience=audience_param, animation_length_minutes=duration, output_filename=output_filename, quality=quality_param, progress_callback=progress_callback, ) progress(1.0, desc="Complete!") logger.info("Async generation completed, returning result") return result except Exception as e: logger.error(f"Generation failed: {e}", exc_info=True) return {"success": False, "error": str(e)} finally: self.is_generating = False def generate_animation_sync( self, topic: str, audience: str, duration: float, quality: str, progress=gr.Progress() ) -> Tuple[str, str, str, str, str, str]: """ Synchronous wrapper for Gradio interface. Returns: Tuple of (video_path, status, narration, code, quiz, concept_plan) """ try: # Reuse existing event loop or create a persistent one if self.event_loop is None or self.event_loop.is_closed(): self.event_loop = asyncio.new_event_loop() asyncio.set_event_loop(self.event_loop) logger.info("Created new persistent event loop") else: asyncio.set_event_loop(self.event_loop) logger.info("Reusing existing event loop") logger.info("Starting event loop execution...") result = self.event_loop.run_until_complete( self.generate_animation_async(topic, audience, duration, quality, progress) ) logger.info("Event loop execution completed") # DO NOT close the loop - keep it for subsequent generations if result["success"]: logger.info("Processing successful result...") video_path = result["output_file"] status = f"ā **Animation Generated Successfully!**\n\n**Topic:** {result['topic']}\n**Audience:** {result['target_audience']}\n**Output:** {os.path.basename(video_path)}" narration = result.get("narration", "Not available") code = result.get("manim_code", "Not available") quiz_raw = result.get("quiz", "Not available") quiz = format_quiz_markdown(quiz_raw) concept = result.get("concept_plan", "Not available") logger.info(f"Returning result to Gradio: {video_path}") return video_path, video_path, status, narration, code, quiz, concept else: error_msg = result.get("error", "Unknown error") status = f"ā **Generation Failed**\n\n{error_msg}" return None, None, status, "", "", "", "" except Exception as e: logger.error(f"Sync wrapper error: {e}", exc_info=True) status = f"š„ **Unexpected Error**\n\n{str(e)}" return None, None, status, "", "", "", "" def create_interface() -> gr.Blocks: """Create the Gradio interface.""" app = NeuroAnimApp() # Custom CSS for better styling custom_css = """ .main-title { text-align: center; color: #2563eb; font-size: 2.5em; font-weight: bold; margin-bottom: 0.5em; } .subtitle { text-align: center; color: #64748b; font-size: 1.2em; margin-bottom: 2em; } .status-box { padding: 1em; border-radius: 8px; margin: 1em 0; } .gradio-container { max-width: 1400px !important; } /* Video player styling */ video { border-radius: 8px; box-shadow: 0 4px 6px rgba(0, 0, 0, 0.1); } /* Quiz and content styling */ .markdown-text h2 { color: #1e40af; border-bottom: 2px solid #3b82f6; padding-bottom: 0.5em; margin-top: 1em; } .markdown-text h3 { color: #1e293b; margin-top: 1em; } .markdown-text blockquote { background-color: #f0fdf4; border-left: 4px solid #22c55e; padding: 0.5em 1em; margin: 1em 0; } /* Button styling */ .primary { background: linear-gradient(135deg, #2563eb 0%, #1d4ed8 100%); } /* Code block styling */ .code-container { border-radius: 8px; margin: 1em 0; } """ with gr.Blocks(title="NeuroAnim - STEM Animation Generator") as interface: # Apply custom CSS interface.css = custom_css # Header gr.HTML("""