""" 📊 Real-time Progress Tracker for CourseCrafter AI Advanced progress tracking with visual feedback and status updates. """ import gradio as gr import asyncio import time from typing import Dict, List, Any, Optional, Callable, Generator from dataclasses import dataclass from enum import Enum import logging import threading from datetime import datetime logger = logging.getLogger(__name__) class ProgressStatus(Enum): """Progress status enumeration""" PENDING = "pending" ACTIVE = "active" COMPLETE = "complete" ERROR = "error" SKIPPED = "skipped" @dataclass class ProgressStep: """Individual progress step""" id: str name: str description: str emoji: str status: ProgressStatus = ProgressStatus.PENDING progress: float = 0.0 start_time: Optional[datetime] = None end_time: Optional[datetime] = None error_message: Optional[str] = None substeps: List['ProgressStep'] = None def __post_init__(self): if self.substeps is None: self.substeps = [] class RealTimeProgressTracker: """Real-time progress tracking with visual updates""" def __init__(self): self.steps: List[ProgressStep] = [] self.current_step_index = 0 self.overall_progress = 0.0 self.is_running = False self.start_time: Optional[datetime] = None self.end_time: Optional[datetime] = None self.log_entries: List[str] = [] self.update_callbacks: List[Callable] = [] # Initialize default steps self._initialize_default_steps() def _initialize_default_steps(self): """Initialize the default course generation steps""" self.steps = [ ProgressStep( id="research", name="Research & Analysis", description="Gathering information and analyzing the topic", emoji="🔍", substeps=[ ProgressStep("research_web", "Web Search", "Searching for relevant information", "🌐"), ProgressStep("research_academic", "Academic Sources", "Finding scholarly content", "📚"), ProgressStep("research_analysis", "Content Analysis", "Analyzing gathered information", "🧠"), ] ), ProgressStep( id="planning", name="Course Planning", description="Creating the course structure and outline", emoji="📋", substeps=[ ProgressStep("plan_structure", "Course Structure", "Designing lesson flow", "🏗️"), ProgressStep("plan_objectives", "Learning Objectives", "Defining learning goals", "🎯"), ProgressStep("plan_assessment", "Assessment Strategy", "Planning quizzes and activities", "📝"), ] ), ProgressStep( id="content", name="Content Generation", description="Creating engaging lesson content", emoji="✍️", substeps=[ ProgressStep("content_lessons", "Lesson Content", "Writing lesson materials", "📖"), ProgressStep("content_examples", "Examples & Exercises", "Creating practical examples", "💡"), ProgressStep("content_review", "Content Review", "Reviewing and refining content", "🔍"), ] ), ProgressStep( id="assessments", name="Assessment Creation", description="Generating quizzes and flashcards", emoji="🎯", substeps=[ ProgressStep("assess_flashcards", "Flashcards", "Creating study flashcards", "🃏"), ProgressStep("assess_quizzes", "Quizzes", "Generating quiz questions", "❓"), ProgressStep("assess_validation", "Validation", "Validating assessment quality", "✅"), ] ), ProgressStep( id="images", name="Visual Content", description="Generating images and diagrams", emoji="🎨", substeps=[ ProgressStep("images_cover", "Course Cover", "Creating course cover image", "🖼️"), ProgressStep("images_lessons", "Lesson Images", "Generating lesson illustrations", "🎭"), ProgressStep("images_diagrams", "Diagrams", "Creating concept diagrams", "📊"), ] ), ProgressStep( id="finalize", name="Finalization", description="Packaging and finalizing the course", emoji="📦", substeps=[ ProgressStep("final_package", "Course Package", "Assembling course materials", "📁"), ProgressStep("final_metadata", "Metadata", "Adding course metadata", "🏷️"), ProgressStep("final_validation", "Final Validation", "Final quality check", "🔍"), ] ) ] def start_generation(self) -> None: """Start the course generation process""" self.is_running = True self.start_time = datetime.now() self.current_step_index = 0 self.overall_progress = 0.0 self.log_entries = [] self.add_log_entry("🚀 Starting course generation...") self.add_log_entry(f"⏰ Started at {self.start_time.strftime('%H:%M:%S')}") def update_step_progress(self, step_id: str, progress: float, message: str = "") -> None: """Update progress for a specific step""" step = self._find_step_by_id(step_id) if not step: logger.warning(f"Step {step_id} not found") return step.progress = min(100.0, max(0.0, progress)) if step.status == ProgressStatus.PENDING and progress > 0: step.status = ProgressStatus.ACTIVE step.start_time = datetime.now() self.add_log_entry(f"{step.emoji} Started: {step.name}") if progress >= 100.0: step.status = ProgressStatus.COMPLETE step.end_time = datetime.now() duration = (step.end_time - step.start_time).total_seconds() if step.start_time else 0 self.add_log_entry(f"✅ Completed: {step.name} ({duration:.1f}s)") if message: self.add_log_entry(f" {message}") self._update_overall_progress() self._notify_callbacks() def update_substep_progress(self, step_id: str, substep_id: str, progress: float, message: str = "") -> None: """Update progress for a substep""" step = self._find_step_by_id(step_id) if not step: return substep = self._find_substep_by_id(step, substep_id) if not substep: return substep.progress = min(100.0, max(0.0, progress)) if substep.status == ProgressStatus.PENDING and progress > 0: substep.status = ProgressStatus.ACTIVE substep.start_time = datetime.now() self.add_log_entry(f" {substep.emoji} {substep.name}...") if progress >= 100.0: substep.status = ProgressStatus.COMPLETE substep.end_time = datetime.now() if message: self.add_log_entry(f" {message}") # Update parent step progress based on substeps self._update_step_from_substeps(step) self._notify_callbacks() def mark_step_error(self, step_id: str, error_message: str) -> None: """Mark a step as having an error""" step = self._find_step_by_id(step_id) if not step: return step.status = ProgressStatus.ERROR step.error_message = error_message step.end_time = datetime.now() self.add_log_entry(f"❌ Error in {step.name}: {error_message}") self._notify_callbacks() def skip_step(self, step_id: str, reason: str = "") -> None: """Skip a step""" step = self._find_step_by_id(step_id) if not step: return step.status = ProgressStatus.SKIPPED step.progress = 100.0 step.end_time = datetime.now() skip_msg = f"⏭️ Skipped: {step.name}" if reason: skip_msg += f" ({reason})" self.add_log_entry(skip_msg) self._update_overall_progress() self._notify_callbacks() def complete_generation(self, success: bool = True) -> None: """Complete the course generation process""" self.is_running = False self.end_time = datetime.now() if success: self.overall_progress = 100.0 total_time = (self.end_time - self.start_time).total_seconds() if self.start_time else 0 self.add_log_entry(f"🎉 Course generation completed successfully!") self.add_log_entry(f"⏱️ Total time: {total_time:.1f} seconds") else: self.add_log_entry("❌ Course generation failed") self._notify_callbacks() def add_log_entry(self, message: str) -> None: """Add an entry to the progress log""" timestamp = datetime.now().strftime("%H:%M:%S") log_entry = f"[{timestamp}] {message}" self.log_entries.append(log_entry) # Keep only last 100 entries if len(self.log_entries) > 100: self.log_entries = self.log_entries[-100:] def get_progress_display(self) -> Dict[str, Any]: """Get current progress display data""" return { "overall_progress": self.overall_progress, "current_step": self.get_current_step_display(), "step_indicators": self.get_step_indicators(), "progress_log": "\n".join(self.log_entries[-20:]), # Last 20 entries "is_running": self.is_running, "elapsed_time": self.get_elapsed_time() } def get_current_step_display(self) -> str: """Get current step display HTML""" if not self.is_running: if self.overall_progress >= 100: return """
🎉
Course Generation Complete!
Your course is ready to explore
""" else: return """
🚀
Ready to Generate Course
Click the generate button to begin
""" current_step = self._get_current_active_step() if not current_step: return "
Processing...
" status_class = current_step.status.value return f"""
{current_step.emoji}
{current_step.name}
{current_step.description}
{current_step.progress:.0f}%
""" def get_step_indicators(self) -> str: """Get step indicators HTML""" indicators_html = "
" for step in self.steps: status_class = step.status.value indicators_html += f"""
{step.emoji}
{step.name}
{step.progress:.0f}%
""" indicators_html += "
" return indicators_html def get_elapsed_time(self) -> str: """Get elapsed time string""" if not self.start_time: return "00:00" end_time = self.end_time or datetime.now() elapsed = (end_time - self.start_time).total_seconds() minutes = int(elapsed // 60) seconds = int(elapsed % 60) return f"{minutes:02d}:{seconds:02d}" def add_update_callback(self, callback: Callable) -> None: """Add a callback for progress updates""" self.update_callbacks.append(callback) def _find_step_by_id(self, step_id: str) -> Optional[ProgressStep]: """Find a step by ID""" for step in self.steps: if step.id == step_id: return step return None def _find_substep_by_id(self, step: ProgressStep, substep_id: str) -> Optional[ProgressStep]: """Find a substep by ID""" for substep in step.substeps: if substep.id == substep_id: return substep return None def _update_step_from_substeps(self, step: ProgressStep) -> None: """Update step progress based on substeps""" if not step.substeps: return total_progress = sum(substep.progress for substep in step.substeps) step.progress = total_progress / len(step.substeps) # Update step status based on substeps if all(substep.status == ProgressStatus.COMPLETE for substep in step.substeps): step.status = ProgressStatus.COMPLETE if not step.end_time: step.end_time = datetime.now() elif any(substep.status == ProgressStatus.ACTIVE for substep in step.substeps): if step.status == ProgressStatus.PENDING: step.status = ProgressStatus.ACTIVE step.start_time = datetime.now() elif any(substep.status == ProgressStatus.ERROR for substep in step.substeps): step.status = ProgressStatus.ERROR def _update_overall_progress(self) -> None: """Update overall progress based on all steps""" total_progress = sum(step.progress for step in self.steps) self.overall_progress = total_progress / len(self.steps) if self.steps else 0.0 def _get_current_active_step(self) -> Optional[ProgressStep]: """Get the currently active step""" for step in self.steps: if step.status == ProgressStatus.ACTIVE: return step # If no active step, return the first pending step for step in self.steps: if step.status == ProgressStatus.PENDING: return step return None def _notify_callbacks(self) -> None: """Notify all registered callbacks""" for callback in self.update_callbacks: try: callback() except Exception as e: logger.error(f"Error in progress callback: {e}") class ProgressSimulator: """Simulates realistic progress for demonstration""" def __init__(self, tracker: RealTimeProgressTracker): self.tracker = tracker self.is_running = False async def simulate_course_generation(self) -> None: """Simulate a realistic course generation process""" self.is_running = True self.tracker.start_generation() try: # Research phase await self._simulate_step("research", [ ("research_web", "Searching web for relevant content...", 3.0), ("research_academic", "Finding academic sources...", 2.0), ("research_analysis", "Analyzing gathered information...", 2.5), ]) # Planning phase await self._simulate_step("planning", [ ("plan_structure", "Designing course structure...", 2.0), ("plan_objectives", "Defining learning objectives...", 1.5), ("plan_assessment", "Planning assessments...", 1.0), ]) # Content generation phase await self._simulate_step("content", [ ("content_lessons", "Generating lesson content...", 4.0), ("content_examples", "Creating examples and exercises...", 3.0), ("content_review", "Reviewing content quality...", 1.5), ]) # Assessment creation phase await self._simulate_step("assessments", [ ("assess_flashcards", "Creating flashcards...", 2.0), ("assess_quizzes", "Generating quiz questions...", 2.5), ("assess_validation", "Validating assessments...", 1.0), ]) # Image generation phase await self._simulate_step("images", [ ("images_cover", "Creating course cover image...", 3.0), ("images_lessons", "Generating lesson illustrations...", 4.0), ("images_diagrams", "Creating concept diagrams...", 2.0), ]) # Finalization phase await self._simulate_step("finalize", [ ("final_package", "Packaging course materials...", 1.5), ("final_metadata", "Adding metadata...", 0.5), ("final_validation", "Final quality check...", 1.0), ]) self.tracker.complete_generation(success=True) except Exception as e: self.tracker.add_log_entry(f"❌ Simulation error: {str(e)}") self.tracker.complete_generation(success=False) finally: self.is_running = False async def _simulate_step(self, step_id: str, substeps: List[tuple]) -> None: """Simulate a step with substeps""" for substep_id, message, duration in substeps: if not self.is_running: break self.tracker.add_log_entry(f"Starting {message}") # Simulate gradual progress steps = 20 for i in range(steps + 1): if not self.is_running: break progress = (i / steps) * 100 self.tracker.update_substep_progress(step_id, substep_id, progress) await asyncio.sleep(duration / steps) # Add some realistic variation await asyncio.sleep(0.2) def stop_simulation(self) -> None: """Stop the simulation""" self.is_running = False