# enhanced_progress_components.py """ Enhanced Progress UI Components for Verification Modes. Provides Gradio components for real-time progress tracking, statistics display, and session management across all verification modes. Requirements: 9.1, 9.2, 9.3, 9.4, 9.5 """ from __future__ import annotations import gradio as gr from typing import Tuple, Dict, Any, Optional from datetime import datetime, timedelta from src.core.enhanced_progress_tracker import ( EnhancedProgressTracker, VerificationMode, ProgressDisplayFormatter ) class EnhancedProgressComponents: """Enhanced progress tracking UI components.""" @staticmethod def create_progress_panel() -> Tuple[gr.Component, gr.Component, gr.Component, gr.Component, gr.Component]: """ Create comprehensive progress tracking panel. Returns: Tuple of (progress_display, accuracy_display, speed_display, error_display, time_display) """ # Main progress display with bar and position progress_display = gr.HTML( value=EnhancedProgressComponents._get_initial_progress_html(), label="Progress Tracking" ) # Running accuracy display accuracy_display = gr.Markdown( value="🎯 Current Accuracy: No verifications yet", label="Accuracy" ) # Processing speed display (for batch mode) speed_display = gr.Markdown( value="", label="Processing Speed", visible=False ) # Error count and status display error_display = gr.Markdown( value="", label="Error Status", visible=False ) # Time tracking display time_display = gr.Markdown( value="⏱️ Time: Ready to start", label="Session Time" ) return progress_display, accuracy_display, speed_display, error_display, time_display @staticmethod def create_compact_progress_panel() -> gr.Component: """ Create compact progress panel for smaller interfaces. Returns: Single HTML component with comprehensive progress info """ return gr.HTML( value=EnhancedProgressComponents._get_initial_progress_html(), label="Session Progress" ) @staticmethod def create_session_controls() -> Tuple[gr.Component, gr.Component, gr.Component]: """ Create session control buttons. Returns: Tuple of (pause_btn, resume_btn, reset_btn) """ pause_btn = gr.Button( "⏸️ Pause Session", variant="secondary", size="sm", visible=False ) resume_btn = gr.Button( "▢️ Resume Session", variant="primary", size="sm", visible=False ) reset_btn = gr.Button( "πŸ”„ Reset Progress", variant="stop", size="sm" ) return pause_btn, resume_btn, reset_btn @staticmethod def update_progress_displays( tracker: EnhancedProgressTracker, use_compact: bool = False ) -> Tuple[str, str, str, str, str, bool, bool]: """ Update all progress displays based on tracker state. Args: tracker: Progress tracker instance use_compact: Whether to use compact display format Returns: Tuple of display values and visibility states """ if use_compact: # Return single HTML component progress_html = ProgressDisplayFormatter.create_progress_panel_html(tracker) return ( progress_html, # progress_display "", # accuracy_display (unused in compact) "", # speed_display (unused in compact) "", # error_display (unused in compact) "", # time_display (unused in compact) False, # speed_visible False # error_visible ) else: # Return individual components progress_html = ProgressDisplayFormatter.create_progress_panel_html(tracker) accuracy_display = tracker.get_accuracy_display() speed_display = tracker.get_processing_speed_display() error_display = tracker.get_error_display() time_display = tracker.get_time_tracking_display() # Determine visibility speed_visible = tracker.mode == VerificationMode.FILE_UPLOAD and tracker.stats.processing_speed > 0 error_visible = tracker.error_tracker.error_count > 0 return ( progress_html, accuracy_display, speed_display, error_display, time_display, speed_visible, error_visible ) @staticmethod def update_session_controls( tracker: EnhancedProgressTracker ) -> Tuple[bool, bool, bool]: """ Update session control button visibility. Args: tracker: Progress tracker instance Returns: Tuple of (pause_visible, resume_visible, reset_visible) """ session_active = tracker.stats.start_time is not None is_paused = tracker.is_paused pause_visible = session_active and not is_paused resume_visible = session_active and is_paused reset_visible = session_active return pause_visible, resume_visible, reset_visible @staticmethod def create_statistics_summary() -> gr.Component: """ Create detailed statistics summary component. Returns: Gradio component for statistics display """ return gr.Markdown( value=EnhancedProgressComponents._get_initial_stats_summary(), label="Session Statistics" ) @staticmethod def update_statistics_summary(tracker: EnhancedProgressTracker) -> str: """ Update statistics summary display. Args: tracker: Progress tracker instance Returns: Formatted statistics summary """ stats = tracker.get_comprehensive_stats() # Calculate additional metrics total_verified = stats["correct_count"] + stats["incorrect_count"] summary = f""" ### πŸ“Š Session Statistics **Progress Overview:** - Messages Processed: {stats['processed_messages']}/{stats['total_messages']} ({stats['completion_percentage']:.1f}%) - Verifications Complete: {total_verified} - Current Accuracy: {stats['accuracy']:.1f}% **Performance Metrics:** - Correct Classifications: {stats['correct_count']} - Incorrect Classifications: {stats['incorrect_count']} """ if tracker.mode == VerificationMode.FILE_UPLOAD: summary += f"- Processing Speed: {stats['processing_speed']:.1f} messages/min\n" if stats["average_processing_time"] > 0: summary += f"- Average Time per Message: {stats['average_processing_time']:.1f}s\n" summary += f""" **Session Timing:** - Elapsed Time: {EnhancedProgressComponents._format_duration(stats['elapsed_time'])} """ if stats["estimated_remaining"]: remaining_str = EnhancedProgressComponents._format_duration(stats["estimated_remaining"]) summary += f"- Estimated Remaining: {remaining_str}\n" if stats["is_paused"]: summary += "- Status: ⏸️ **Paused**\n" if stats["error_count"] > 0: summary += f""" **Error Information:** - Total Errors: {stats['error_count']} - Can Continue: {'βœ… Yes' if stats['can_continue'] else '❌ No'} """ return summary @staticmethod def create_error_details_panel() -> gr.Component: """ Create detailed error information panel. Returns: Gradio component for error details """ return gr.Markdown( value="", label="Error Details", visible=False ) @staticmethod def update_error_details(tracker: EnhancedProgressTracker) -> Tuple[str, bool]: """ Update error details panel. Args: tracker: Progress tracker instance Returns: Tuple of (error_details, visible) """ if tracker.error_tracker.error_count == 0: return "", False recent_errors = tracker.error_tracker.get_recent_errors(5) details = f""" ### ⚠️ Error Details **Total Errors:** {tracker.error_tracker.error_count} **Can Continue Processing:** {'βœ… Yes' if tracker.error_tracker.can_continue else '❌ No'} **Recent Errors:** """ for i, (error_msg, timestamp) in enumerate(recent_errors, 1): time_str = timestamp.strftime("%H:%M:%S") details += f"{i}. `{time_str}` - {error_msg}\n" if tracker.error_tracker.error_count > len(recent_errors): details += f"\n*... and {tracker.error_tracker.error_count - len(recent_errors)} more errors*" return details, True @staticmethod def _get_initial_progress_html() -> str: """Get initial progress HTML.""" return """
πŸ“Š Ready to Start
Select a dataset or enter messages to begin
""" @staticmethod def _get_initial_stats_summary() -> str: """Get initial statistics summary.""" return """ ### πŸ“Š Session Statistics **Progress Overview:** - Messages Processed: 0/0 (0%) - Verifications Complete: 0 - Current Accuracy: 0% **Performance Metrics:** - Correct Classifications: 0 - Incorrect Classifications: 0 **Session Timing:** - Elapsed Time: 0s - Status: Ready to start """ @staticmethod def _format_duration(seconds: float) -> str: """Format duration in seconds to human-readable string.""" if seconds is None or seconds <= 0: return "0s" total_seconds = int(seconds) if total_seconds < 60: return f"{total_seconds}s" elif total_seconds < 3600: minutes = total_seconds // 60 seconds = total_seconds % 60 return f"{minutes}m {seconds}s" else: hours = total_seconds // 3600 minutes = (total_seconds % 3600) // 60 return f"{hours}h {minutes}m" class ProgressTrackingMixin: """Mixin class for adding progress tracking to verification interfaces.""" def __init__(self, mode: VerificationMode): """Initialize progress tracking.""" self.progress_tracker = EnhancedProgressTracker(mode) self.progress_components = None def setup_progress_tracking(self, total_messages: int = 0) -> None: """ Setup progress tracking for a session. Args: total_messages: Total number of messages to process """ self.progress_tracker = EnhancedProgressTracker(self.progress_tracker.mode, total_messages) self.progress_tracker.start_session() def record_verification_with_timing(self, is_correct: bool, start_time: datetime = None) -> None: """ Record verification with automatic timing. Args: is_correct: Whether verification was correct start_time: When processing started (for timing calculation) """ processing_time = None if start_time: processing_time = (datetime.now() - start_time).total_seconds() self.progress_tracker.record_verification(is_correct, processing_time) def get_progress_updates(self, use_compact: bool = False) -> Tuple: """ Get all progress display updates. Args: use_compact: Whether to use compact display Returns: Tuple of display updates """ return EnhancedProgressComponents.update_progress_displays( self.progress_tracker, use_compact ) def handle_session_pause(self) -> Tuple[bool, bool, bool]: """ Handle session pause and return control states. Returns: Tuple of control button visibility states """ self.progress_tracker.pause_session() return EnhancedProgressComponents.update_session_controls(self.progress_tracker) def handle_session_resume(self) -> Tuple[bool, bool, bool]: """ Handle session resume and return control states. Returns: Tuple of control button visibility states """ self.progress_tracker.resume_session() return EnhancedProgressComponents.update_session_controls(self.progress_tracker)