Spaces:
Sleeping
Sleeping
| # enhanced_verification_interface.py | |
| """ | |
| Enhanced Verification Interface Integration. | |
| Integrates the enhanced verification modes with the existing Gradio application. | |
| Provides mode selection, session resumption, and progress preservation. | |
| Requirements: 1.1, 1.2, 1.3, 1.4, 1.5, 6.1 | |
| """ | |
| import gradio as gr | |
| from typing import List, Dict, Tuple, Optional, Any, Union | |
| from datetime import datetime | |
| import uuid | |
| from src.core.verification_models import ( | |
| EnhancedVerificationSession, | |
| VerificationRecord, | |
| TestMessage, | |
| TestDataset, | |
| ) | |
| from src.core.verification_store import JSONVerificationStore | |
| from src.core.test_datasets import TestDatasetManager | |
| from src.interface.enhanced_verification_ui import EnhancedVerificationUIComponents | |
| # Import configuration with fallback defaults | |
| try: | |
| from app_config import ( | |
| ENHANCED_VERIFICATION_CONFIG, | |
| FEATURE_FLAGS, | |
| is_feature_enabled | |
| ) | |
| except ImportError: | |
| ENHANCED_VERIFICATION_CONFIG = {"enabled": True, "default_mode": None} | |
| FEATURE_FLAGS = { | |
| "manual_input_mode_enabled": True, | |
| "file_upload_mode_enabled": True, | |
| "dataset_editing_enabled": True, | |
| "show_incomplete_session_prompts": True, | |
| } | |
| def is_feature_enabled(feature_name: str) -> bool: | |
| return FEATURE_FLAGS.get(feature_name, False) | |
| class EnhancedVerificationInterface: | |
| """Main interface controller for enhanced verification modes.""" | |
| def __init__(self, store: JSONVerificationStore = None, config: dict = None): | |
| """ | |
| Initialize the enhanced verification interface. | |
| Args: | |
| store: Verification data store (optional, creates default if not provided) | |
| config: Configuration dictionary (optional, uses ENHANCED_VERIFICATION_CONFIG if not provided) | |
| """ | |
| self.store = store or JSONVerificationStore() | |
| self.config = config or ENHANCED_VERIFICATION_CONFIG | |
| self.current_mode = self.config.get("default_mode", None) | |
| self.current_session = None | |
| self.incomplete_sessions = [] | |
| # Feature flags for mode availability | |
| self.manual_input_enabled = is_feature_enabled("manual_input_mode_enabled") | |
| self.file_upload_enabled = is_feature_enabled("file_upload_mode_enabled") | |
| self.dataset_editing_enabled = is_feature_enabled("dataset_editing_enabled") | |
| self.show_incomplete_prompts = is_feature_enabled("show_incomplete_session_prompts") | |
| def create_interface(self) -> gr.Blocks: | |
| """ | |
| Create the complete enhanced verification interface. | |
| Returns: | |
| Gradio Blocks component with mode selection and all verification modes | |
| """ | |
| with gr.Blocks(title="Enhanced Verification Modes") as interface: | |
| # Application state | |
| current_mode_state = gr.State(value=None) | |
| current_session_state = gr.State(value=None) | |
| incomplete_sessions_state = gr.State(value=[]) | |
| pending_mode_switch_state = gr.State(value=None) | |
| selected_session_state = gr.State(value=None) | |
| # Model overrides (populated by the main app's AI Model Configuration, if wired) | |
| model_overrides_state = gr.State(value={}) | |
| # Main container | |
| with gr.Column(): | |
| # Header | |
| gr.Markdown("# 🔍 Enhanced Verification Modes") | |
| gr.Markdown("Choose your verification approach based on your testing needs and data source.") | |
| # Status message | |
| status_message = gr.Markdown("", visible=True, label="Status") | |
| # Incomplete sessions section | |
| incomplete_sessions_section = gr.Row(visible=False) | |
| with incomplete_sessions_section: | |
| with gr.Column(): | |
| gr.Markdown("## 📋 Resume Previous Sessions") | |
| gr.Markdown("You have incomplete verification sessions. You can resume where you left off or start a new session.") | |
| incomplete_sessions_display = gr.HTML( | |
| value="", | |
| label="Incomplete Sessions" | |
| ) | |
| with gr.Row(): | |
| resume_session_btn = gr.Button( | |
| "▶️ Resume Selected Session", | |
| variant="primary", | |
| scale=2 | |
| ) | |
| clear_sessions_btn = gr.Button( | |
| "🗑️ Clear All Sessions", | |
| variant="secondary", | |
| scale=1 | |
| ) | |
| # Mode selection section | |
| mode_selection_section = gr.Row(visible=True) | |
| with mode_selection_section: | |
| with gr.Column(): | |
| gr.Markdown("## 🎯 Select Verification Mode") | |
| with gr.Row(): | |
| # Standard Verification Mode (replaced Enhanced Dataset) | |
| # NOTE: Per UI simplification request, we keep the implementation | |
| # but hide this tile by default. | |
| show_standard_verification_tile = False | |
| with gr.Column(scale=1, visible=show_standard_verification_tile): | |
| gr.Markdown("### ✓ Standard Verification") | |
| gr.Markdown("Use predefined test datasets to verify classifier accuracy. Quick and straightforward verification workflow.") | |
| gr.Markdown("**Features:**") | |
| gr.Markdown("• Pre-classified test datasets") | |
| gr.Markdown("• Step-by-step message review") | |
| gr.Markdown("• Accuracy tracking and statistics") | |
| gr.Markdown("• Export verification results") | |
| gr.Markdown("• Session progress preservation") | |
| standard_verification_btn = gr.Button( | |
| "✓ Start Standard Verification", | |
| variant="primary", | |
| size="lg" | |
| ) | |
| # Manual Input Mode | |
| with gr.Column(scale=1): | |
| mode_info = EnhancedVerificationUIComponents.MODE_OPTIONS["manual_input"] | |
| gr.Markdown(f"### {mode_info['icon']} {mode_info['title']}") | |
| gr.Markdown(mode_info["description"]) | |
| gr.Markdown("**Features:**") | |
| for feature in mode_info["features"]: | |
| gr.Markdown(f"• {feature}") | |
| manual_input_btn = gr.Button( | |
| f"{mode_info['icon']} Start Manual Input Mode", | |
| variant="primary", | |
| size="lg" | |
| ) | |
| # File Upload Mode | |
| with gr.Column(scale=1): | |
| mode_info = EnhancedVerificationUIComponents.MODE_OPTIONS["file_upload"] | |
| gr.Markdown(f"### {mode_info['icon']} {mode_info['title']}") | |
| gr.Markdown(mode_info["description"]) | |
| gr.Markdown("**Features:**") | |
| for feature in mode_info["features"]: | |
| gr.Markdown(f"• {feature}") | |
| file_upload_btn = gr.Button( | |
| f"{mode_info['icon']} Start File Upload Mode", | |
| variant="primary", | |
| size="lg" | |
| ) | |
| # Mode switch confirmation dialog | |
| mode_switch_dialog = gr.Row(visible=False) | |
| with mode_switch_dialog: | |
| with gr.Column(): | |
| gr.Markdown("### ⚠️ Switch Mode Confirmation") | |
| switch_warning_text = gr.Markdown( | |
| "You have unsaved progress in the current mode. What would you like to do?", | |
| label="Warning" | |
| ) | |
| with gr.Row(): | |
| save_and_switch_btn = gr.Button( | |
| "💾 Save Progress & Switch", | |
| variant="primary", | |
| scale=2 | |
| ) | |
| discard_and_switch_btn = gr.Button( | |
| "🗑️ Discard & Switch", | |
| variant="secondary", | |
| scale=1 | |
| ) | |
| cancel_switch_btn = gr.Button( | |
| "❌ Cancel", | |
| scale=1 | |
| ) | |
| # Individual mode interfaces (initially hidden) | |
| # Standard Verification interface (replaced Enhanced Dataset) | |
| standard_verification_interface = gr.Row(visible=False) | |
| with standard_verification_interface: | |
| with gr.Column(): | |
| gr.Markdown("# ✓ Standard Verification Mode") | |
| gr.Markdown("Review classified messages and provide feedback to improve the spiritual distress classifier.") | |
| back_from_standard_btn = gr.Button("← Back to Mode Selection", size="sm") | |
| gr.Markdown("---") | |
| # Create dataset interface components | |
| from src.interface.enhanced_dataset_interface import EnhancedDatasetInterfaceController | |
| dataset_controller = EnhancedDatasetInterfaceController() | |
| # Pre-load dataset choices for initialization | |
| try: | |
| initial_choices, _, _, _ = dataset_controller.initialize_interface() | |
| except Exception: | |
| initial_choices = [] | |
| # Dataset selection interface | |
| with gr.Row(): | |
| with gr.Column(scale=2): | |
| gr.Markdown("## 📊 Select Dataset") | |
| dataset_selector = gr.Dropdown( | |
| choices=initial_choices, | |
| value=initial_choices[0] if initial_choices else None, | |
| label="Available Datasets", | |
| info="Choose a dataset to verify", | |
| interactive=True | |
| ) | |
| load_dataset_btn = gr.Button("📥 Load Dataset", variant="primary") | |
| with gr.Column(scale=1): | |
| gr.Markdown("## 📊 Dataset Information") | |
| dataset_info_display = gr.Markdown( | |
| "Select a dataset to view details", | |
| label="Dataset Details" | |
| ) | |
| # Verification setup section (initially hidden) | |
| verification_section = gr.Row(visible=False) | |
| with verification_section: | |
| with gr.Column(): | |
| gr.Markdown("## 🔍 Dataset Verification") | |
| # Verification controls | |
| with gr.Row(): | |
| with gr.Column(scale=2): | |
| verifier_name_input = gr.Textbox( | |
| label="Verifier Name", | |
| placeholder="Enter your name...", | |
| interactive=True | |
| ) | |
| with gr.Column(scale=1): | |
| start_verification_btn = gr.Button( | |
| "🚀 Start Verification", | |
| variant="primary", | |
| size="lg" | |
| ) | |
| # Progress display | |
| verification_progress = gr.Markdown( | |
| "Ready to start verification", | |
| label="Progress" | |
| ) | |
| # Message review section (initially hidden) | |
| message_review_section = gr.Row(visible=False) | |
| with message_review_section: | |
| with gr.Column(scale=2): | |
| gr.Markdown("## 📝 Message Review") | |
| # Current message display | |
| current_message_display = gr.Textbox( | |
| label="Patient Message", | |
| interactive=False, | |
| lines=4 | |
| ) | |
| # Classification result | |
| with gr.Row(): | |
| classifier_decision_display = gr.Markdown( | |
| "🔄 Loading...", | |
| label="Classifier Decision" | |
| ) | |
| classifier_confidence_display = gr.Markdown( | |
| "", | |
| label="Confidence" | |
| ) | |
| # Feedback buttons | |
| gr.Markdown("### Is this classification correct?") | |
| with gr.Row(): | |
| correct_btn = gr.Button("✓ Correct", variant="primary", scale=1) | |
| incorrect_btn = gr.Button("✗ Incorrect", variant="stop", scale=1) | |
| # Correction section (initially hidden) | |
| correction_section = gr.Row(visible=False) | |
| with correction_section: | |
| with gr.Column(): | |
| gr.Markdown("### Select correct classification:") | |
| correction_selector = gr.Radio( | |
| choices=["green", "yellow", "red"], | |
| label="Correct Classification", | |
| value=None | |
| ) | |
| correction_notes = gr.Textbox( | |
| label="Notes (Optional)", | |
| placeholder="Why is this incorrect?", | |
| lines=2 | |
| ) | |
| with gr.Row(): | |
| submit_correction_btn = gr.Button("✓ Submit Correction", variant="primary") | |
| cancel_correction_btn = gr.Button("✗ Cancel", variant="secondary") | |
| with gr.Column(scale=1): | |
| gr.Markdown("## 📊 Session Statistics") | |
| session_stats_display = gr.Markdown( | |
| "**Progress:** 0/0\n**Correct:** 0\n**Incorrect:** 0\n**Accuracy:** 0%" | |
| ) | |
| # Export section | |
| gr.Markdown("## 💾 Export Results") | |
| with gr.Column(): | |
| export_csv_btn = gr.Button("📄 Export CSV", variant="secondary", size="sm") | |
| export_json_btn = gr.Button("📋 Export JSON", variant="secondary", size="sm") | |
| export_xlsx_btn = gr.Button("📊 Export Excel", variant="secondary", size="sm") | |
| # Navigation | |
| gr.Markdown("## 🎮 Session Control") | |
| with gr.Row(): | |
| skip_btn = gr.Button("⏭️ Skip", variant="secondary") | |
| finish_btn = gr.Button("🏁 Finish Session", variant="primary") | |
| # Status message for dataset interface | |
| dataset_status_message = gr.Markdown("", visible=True) | |
| # Dataset state | |
| current_dataset_state = gr.State(value=None) | |
| verification_session_state = gr.State(value=None) | |
| current_message_index_state = gr.State(value=0) | |
| current_classification_state = gr.State(value=None) | |
| manual_input_interface = gr.Row(visible=False) | |
| with manual_input_interface: | |
| with gr.Column(): | |
| gr.Markdown("# ✏️ Manual Input Mode") | |
| gr.Markdown("Enter individual messages for immediate classification and verification.") | |
| back_from_manual_btn = gr.Button("← Back to Mode Selection", size="sm") | |
| gr.Markdown("---") | |
| # Embed the actual interface | |
| manual_input_ui = EnhancedVerificationUIComponents.create_manual_input_interface(model_overrides_state) | |
| file_upload_interface = gr.Row(visible=False) | |
| with file_upload_interface: | |
| with gr.Column(): | |
| gr.Markdown("# 📁 File Upload Mode") | |
| gr.Markdown("Upload CSV or XLSX files for batch processing.") | |
| back_from_file_btn = gr.Button("← Back to Mode Selection", size="sm") | |
| gr.Markdown("---") | |
| # Embed the actual interface | |
| file_upload_ui = EnhancedVerificationUIComponents.create_file_upload_interface(model_overrides_state) | |
| # Event handlers | |
| def initialize_interface(): | |
| """Initialize the interface and check for incomplete sessions.""" | |
| try: | |
| has_incomplete, sessions, display_html = EnhancedVerificationUIComponents.check_for_incomplete_sessions(self.store) | |
| if has_incomplete: | |
| return ( | |
| gr.Row(visible=True), # Show incomplete sessions section | |
| display_html, # Display sessions HTML | |
| sessions, # Store sessions in state | |
| "✨ Welcome back! You have incomplete sessions. You can resume where you left off or start a new session." | |
| ) | |
| else: | |
| return ( | |
| gr.Row(visible=False), # Hide incomplete sessions section | |
| "", # Empty display | |
| [], # Empty sessions list | |
| "✨ Welcome to Enhanced Verification Modes! Choose a mode to get started." | |
| ) | |
| except Exception as e: | |
| return ( | |
| gr.Row(visible=False), | |
| "", | |
| [], | |
| f"❌ Error initializing interface: {str(e)}" | |
| ) | |
| def switch_to_mode( | |
| mode_type: str, | |
| current_mode_val: Optional[str], | |
| current_session_val: Optional[EnhancedVerificationSession] | |
| ): | |
| """Handle mode switching with progress preservation.""" | |
| try: | |
| # Check if we need to show progress preservation warning | |
| has_progress = ( | |
| current_session_val is not None and | |
| not current_session_val.is_complete and | |
| current_session_val.verified_count > 0 | |
| ) | |
| if has_progress and current_mode_val != mode_type: | |
| # Show confirmation dialog | |
| warning_msg, show_dialog = EnhancedVerificationUIComponents.create_mode_switch_confirmation( | |
| current_mode_val, mode_type, has_progress | |
| ) | |
| return ( | |
| gr.Row(visible=True), # Show confirmation dialog | |
| warning_msg, # Warning message | |
| mode_type, # Store pending mode switch | |
| current_mode_val, # Keep current mode | |
| current_session_val, # Keep current session | |
| f"⚠️ Confirm mode switch to {EnhancedVerificationUIComponents.MODE_OPTIONS[mode_type]['title']}" | |
| ) | |
| else: | |
| # Direct switch (no progress to preserve) | |
| # Return only the 6 values expected by the button click handler | |
| result = perform_mode_switch(mode_type, current_session_val) | |
| # Extract only the values needed for the 6-output handler | |
| return ( | |
| result[4], # mode_switch_dialog visibility | |
| result[5], # switch_warning_text | |
| result[6], # pending_mode_switch_state | |
| result[7], # current_mode_state | |
| result[8], # current_session_state | |
| result[9] # status_message | |
| ) | |
| except Exception as e: | |
| return ( | |
| gr.Row(visible=False), # Hide confirmation dialog | |
| "", # Clear warning | |
| None, # Clear pending switch | |
| current_mode_val, # Keep current mode | |
| current_session_val, # Keep current session | |
| f"❌ Error switching modes: {str(e)}" | |
| ) | |
| def perform_mode_switch( | |
| mode_type: str, | |
| session_to_save: Optional[EnhancedVerificationSession] = None, | |
| save_progress: bool = True | |
| ): | |
| """Perform the actual mode switch.""" | |
| try: | |
| # Save current session if exists and requested | |
| if session_to_save and not session_to_save.is_complete and save_progress: | |
| self.store.save_session(session_to_save) | |
| # Update interface visibility | |
| mode_selection_visible = mode_type is None | |
| standard_verification_visible = mode_type == "standard_verification" | |
| manual_input_visible = mode_type == "manual_input" | |
| file_upload_visible = mode_type == "file_upload" | |
| # Create status message | |
| if mode_type: | |
| mode_titles = { | |
| "standard_verification": "Standard Verification", | |
| "manual_input": "Manual Input", | |
| "file_upload": "File Upload" | |
| } | |
| mode_title = mode_titles.get(mode_type, 'Unknown') | |
| status_msg = f"✅ Switched to {mode_title} mode" | |
| else: | |
| status_msg = "✅ Returned to mode selection" | |
| return ( | |
| gr.Row(visible=mode_selection_visible), # Mode selection section | |
| gr.Row(visible=standard_verification_visible), # Standard verification interface | |
| gr.Row(visible=manual_input_visible), # Manual input interface | |
| gr.Row(visible=file_upload_visible), # File upload interface | |
| gr.Row(visible=False), # Hide confirmation dialog | |
| "", # Clear warning message | |
| None, # Clear pending mode switch | |
| mode_type, # Set current mode | |
| None, # Clear current session (will be set by mode interface) | |
| status_msg # Status message | |
| ) | |
| except Exception as e: | |
| return ( | |
| gr.Row(visible=True), # Show mode selection on error | |
| gr.Row(visible=False), # Hide standard verification | |
| gr.Row(visible=False), # Hide manual input | |
| gr.Row(visible=False), # Hide file upload | |
| gr.Row(visible=False), # Hide confirmation dialog | |
| "", # Clear warning | |
| None, # Clear pending switch | |
| None, # Clear current mode | |
| None, # Clear current session | |
| f"❌ Error performing mode switch: {str(e)}" | |
| ) | |
| def resume_selected_session( | |
| sessions: List[EnhancedVerificationSession], | |
| selected_session: Optional[EnhancedVerificationSession] | |
| ): | |
| """Resume a selected session.""" | |
| try: | |
| if not selected_session: | |
| return ( | |
| None, # Current mode | |
| None, # Current session | |
| "⚠️ No session selected. Please select a session first." | |
| ) | |
| # Switch to the appropriate mode for this session | |
| mode_type = selected_session.mode_type | |
| # Update current session | |
| self.current_session = selected_session | |
| # Perform mode switch to resume session | |
| return perform_mode_switch(mode_type, None, False) + (selected_session,) | |
| except Exception as e: | |
| return ( | |
| None, | |
| None, | |
| f"❌ Error resuming session: {str(e)}" | |
| ) | |
| def clear_all_sessions(sessions: List[EnhancedVerificationSession]): | |
| """Clear all incomplete sessions.""" | |
| try: | |
| cleared_count = 0 | |
| for session in sessions: | |
| if self.store.delete_session(session.session_id): | |
| cleared_count += 1 | |
| return ( | |
| gr.Row(visible=False), # Hide incomplete sessions section | |
| "", # Clear display | |
| [], # Clear sessions list | |
| f"✅ Cleared {cleared_count} incomplete session{'s' if cleared_count != 1 else ''}" | |
| ) | |
| except Exception as e: | |
| return ( | |
| gr.Row(visible=True), # Keep section visible | |
| "Error clearing sessions", # Error display | |
| sessions, # Keep sessions | |
| f"❌ Error clearing sessions: {str(e)}" | |
| ) | |
| # Bind initialization | |
| interface.load( | |
| initialize_interface, | |
| outputs=[ | |
| incomplete_sessions_section, | |
| incomplete_sessions_display, | |
| incomplete_sessions_state, | |
| status_message | |
| ] | |
| ) | |
| # Helper function for direct mode switch (no confirmation needed) | |
| def direct_mode_switch(mode_type: str): | |
| """Directly switch to a mode without confirmation.""" | |
| mode_selection_visible = False | |
| standard_verification_visible = mode_type == "standard_verification" | |
| manual_input_visible = mode_type == "manual_input" | |
| file_upload_visible = mode_type == "file_upload" | |
| mode_titles = { | |
| "standard_verification": "Standard Verification", | |
| "manual_input": "Manual Input", | |
| "file_upload": "File Upload" | |
| } | |
| mode_title = mode_titles.get(mode_type, 'Unknown') | |
| status_msg = f"✅ Switched to {mode_title} mode" | |
| return ( | |
| gr.Row(visible=mode_selection_visible), # mode_selection_section | |
| gr.Row(visible=standard_verification_visible), # standard_verification_interface | |
| gr.Row(visible=manual_input_visible), # manual_input_interface | |
| gr.Row(visible=file_upload_visible), # file_upload_interface | |
| mode_type, # current_mode_state | |
| status_msg # status_message | |
| ) | |
| # Special function for standard verification mode initialization | |
| def switch_to_standard_verification(): | |
| """Switch to standard verification mode with proper initialization.""" | |
| try: | |
| # Get first dataset info for display | |
| dataset_choices, _, _, _ = dataset_controller.initialize_interface() | |
| first_dataset = None | |
| dataset_info = "Select a dataset to view details" | |
| if dataset_choices: | |
| first_info, first_dataset = dataset_controller.get_dataset_info(dataset_choices[0]) | |
| dataset_info = first_info | |
| return ( | |
| gr.Row(visible=False), # mode_selection_section | |
| gr.Row(visible=True), # standard_verification_interface | |
| gr.Row(visible=False), # manual_input_interface | |
| gr.Row(visible=False), # file_upload_interface | |
| "standard_verification", # current_mode_state | |
| "✅ Switched to Standard Verification mode", # status_message | |
| dataset_info, # dataset_info_display | |
| first_dataset # current_dataset_state | |
| ) | |
| except Exception as e: | |
| return ( | |
| gr.Row(visible=False), # mode_selection_section | |
| gr.Row(visible=True), # standard_verification_interface | |
| gr.Row(visible=False), # manual_input_interface | |
| gr.Row(visible=False), # file_upload_interface | |
| "standard_verification", # current_mode_state | |
| f"❌ Error initializing verification mode: {str(e)}", # status_message | |
| "Error loading datasets", # dataset_info_display | |
| None # current_dataset_state | |
| ) | |
| # Bind mode selection buttons | |
| standard_verification_btn.click( | |
| switch_to_standard_verification, | |
| inputs=[], | |
| outputs=[ | |
| mode_selection_section, | |
| standard_verification_interface, | |
| manual_input_interface, | |
| file_upload_interface, | |
| current_mode_state, | |
| status_message, | |
| dataset_info_display, | |
| current_dataset_state | |
| ] | |
| ) | |
| manual_input_btn.click( | |
| lambda: direct_mode_switch("manual_input"), | |
| inputs=[], | |
| outputs=[ | |
| mode_selection_section, | |
| standard_verification_interface, | |
| manual_input_interface, | |
| file_upload_interface, | |
| current_mode_state, | |
| status_message | |
| ] | |
| ) | |
| file_upload_btn.click( | |
| lambda: direct_mode_switch("file_upload"), | |
| inputs=[], | |
| outputs=[ | |
| mode_selection_section, | |
| standard_verification_interface, | |
| manual_input_interface, | |
| file_upload_interface, | |
| current_mode_state, | |
| status_message | |
| ] | |
| ) | |
| # Bind confirmation dialog buttons | |
| save_and_switch_btn.click( | |
| lambda pms, cs: perform_mode_switch(pms, cs, True), | |
| inputs=[pending_mode_switch_state, current_session_state], | |
| outputs=[ | |
| mode_selection_section, | |
| standard_verification_interface, | |
| manual_input_interface, | |
| file_upload_interface, | |
| mode_switch_dialog, | |
| switch_warning_text, | |
| pending_mode_switch_state, | |
| current_mode_state, | |
| current_session_state, | |
| status_message | |
| ] | |
| ) | |
| discard_and_switch_btn.click( | |
| lambda pms, cs: perform_mode_switch(pms, cs, False), | |
| inputs=[pending_mode_switch_state, current_session_state], | |
| outputs=[ | |
| mode_selection_section, | |
| standard_verification_interface, | |
| manual_input_interface, | |
| file_upload_interface, | |
| mode_switch_dialog, | |
| switch_warning_text, | |
| pending_mode_switch_state, | |
| current_mode_state, | |
| current_session_state, | |
| status_message | |
| ] | |
| ) | |
| cancel_switch_btn.click( | |
| lambda: ( | |
| gr.Row(visible=False), # Hide dialog | |
| "", # Clear warning | |
| None, # Clear pending switch | |
| "❌ Mode switch cancelled" | |
| ), | |
| outputs=[ | |
| mode_switch_dialog, | |
| switch_warning_text, | |
| pending_mode_switch_state, | |
| status_message | |
| ] | |
| ) | |
| # Helper function to go back to mode selection | |
| def go_back_to_mode_selection(): | |
| """Return to mode selection screen.""" | |
| return ( | |
| gr.Row(visible=True), # mode_selection_section | |
| gr.Row(visible=False), # standard_verification_interface | |
| gr.Row(visible=False), # manual_input_interface | |
| gr.Row(visible=False), # file_upload_interface | |
| None, # current_mode_state | |
| "✅ Returned to mode selection" # status_message | |
| ) | |
| # Bind Back buttons for each mode | |
| back_from_standard_btn.click( | |
| go_back_to_mode_selection, | |
| outputs=[ | |
| mode_selection_section, | |
| standard_verification_interface, | |
| manual_input_interface, | |
| file_upload_interface, | |
| current_mode_state, | |
| status_message | |
| ] | |
| ) | |
| back_from_manual_btn.click( | |
| go_back_to_mode_selection, | |
| outputs=[ | |
| mode_selection_section, | |
| standard_verification_interface, | |
| manual_input_interface, | |
| file_upload_interface, | |
| current_mode_state, | |
| status_message | |
| ] | |
| ) | |
| back_from_file_btn.click( | |
| go_back_to_mode_selection, | |
| outputs=[ | |
| mode_selection_section, | |
| standard_verification_interface, | |
| manual_input_interface, | |
| file_upload_interface, | |
| current_mode_state, | |
| status_message | |
| ] | |
| ) | |
| # Bind session resumption buttons | |
| resume_session_btn.click( | |
| resume_selected_session, | |
| inputs=[incomplete_sessions_state, selected_session_state], | |
| outputs=[ | |
| current_mode_state, | |
| current_session_state, | |
| status_message | |
| ] | |
| ) | |
| clear_sessions_btn.click( | |
| clear_all_sessions, | |
| inputs=[incomplete_sessions_state], | |
| outputs=[ | |
| incomplete_sessions_section, | |
| incomplete_sessions_display, | |
| incomplete_sessions_state, | |
| status_message | |
| ] | |
| ) | |
| # Dataset interface event handlers | |
| def on_dataset_selection_change(dataset_selection): | |
| """Handle dataset selection change.""" | |
| try: | |
| dataset_info, dataset_obj = dataset_controller.get_dataset_info(dataset_selection) | |
| return ( | |
| dataset_info, # dataset_info_display | |
| dataset_obj # current_dataset_state | |
| ) | |
| except Exception as e: | |
| return ( | |
| f"❌ Error loading dataset info: {str(e)}", # dataset_info_display | |
| None # current_dataset_state | |
| ) | |
| def on_load_dataset(current_dataset): | |
| """Handle load dataset for verification.""" | |
| try: | |
| if not current_dataset: | |
| return ( | |
| gr.Row(visible=False), # verification_section | |
| "❌ No dataset selected" # dataset_status_message | |
| ) | |
| return ( | |
| gr.Row(visible=True), # verification_section | |
| f"✅ Dataset '{current_dataset.name}' loaded for verification" # dataset_status_message | |
| ) | |
| except Exception as e: | |
| return ( | |
| gr.Row(visible=False), # verification_section | |
| f"❌ Error loading dataset: {str(e)}" # dataset_status_message | |
| ) | |
| def on_start_verification(current_dataset, verifier_name): | |
| """Handle starting verification session.""" | |
| try: | |
| if not current_dataset: | |
| return ( | |
| None, # verification_session_state | |
| gr.Row(visible=False), # message_review_section | |
| "", # current_message_display | |
| "", # classifier_decision_display | |
| "", # classifier_confidence_display | |
| "", # session_stats_display | |
| 0, # current_message_index_state | |
| None, # current_classification_state | |
| "❌ No dataset selected" # dataset_status_message | |
| ) | |
| if not verifier_name or not verifier_name.strip(): | |
| return ( | |
| None, # verification_session_state | |
| gr.Row(visible=False), # message_review_section | |
| "", # current_message_display | |
| "", # classifier_decision_display | |
| "", # classifier_confidence_display | |
| "", # session_stats_display | |
| 0, # current_message_index_state | |
| None, # current_classification_state | |
| "❌ Please enter your name" # dataset_status_message | |
| ) | |
| success, message, session = dataset_controller.start_verification_session( | |
| current_dataset, verifier_name | |
| ) | |
| if success: | |
| # Get first message for verification | |
| first_message, classification_result = dataset_controller.get_current_message_for_verification() | |
| if first_message: | |
| # Format classification display | |
| decision = classification_result.get('decision', 'unknown').upper() | |
| confidence = classification_result.get('confidence', 0) * 100 | |
| decision_colors = {'GREEN': '🟢', 'YELLOW': '🟡', 'RED': '🔴'} | |
| decision_badge = f"{decision_colors.get(decision, '❓')} **{decision}**" | |
| confidence_text = f"Confidence: **{confidence:.1f}%**" | |
| stats_text = f"""**Progress:** 1/{len(current_dataset.messages)} | |
| **Correct:** 0 | |
| **Incorrect:** 0 | |
| **Accuracy:** N/A""" | |
| return ( | |
| session, # verification_session_state | |
| gr.Row(visible=True), # message_review_section | |
| first_message.text, # current_message_display | |
| decision_badge, # classifier_decision_display | |
| confidence_text, # classifier_confidence_display | |
| stats_text, # session_stats_display | |
| 0, # current_message_index_state | |
| classification_result, # current_classification_state | |
| message # dataset_status_message | |
| ) | |
| else: | |
| return ( | |
| session, # verification_session_state | |
| gr.Row(visible=False), # message_review_section | |
| "", # current_message_display | |
| "", # classifier_decision_display | |
| "", # classifier_confidence_display | |
| "", # session_stats_display | |
| 0, # current_message_index_state | |
| None, # current_classification_state | |
| "❌ No messages in dataset" # dataset_status_message | |
| ) | |
| else: | |
| return ( | |
| None, # verification_session_state | |
| gr.Row(visible=False), # message_review_section | |
| "", # current_message_display | |
| "", # classifier_decision_display | |
| "", # classifier_confidence_display | |
| "", # session_stats_display | |
| 0, # current_message_index_state | |
| None, # current_classification_state | |
| message # dataset_status_message | |
| ) | |
| except Exception as e: | |
| return ( | |
| None, # verification_session_state | |
| gr.Row(visible=False), # message_review_section | |
| "", # current_message_display | |
| "", # classifier_decision_display | |
| "", # classifier_confidence_display | |
| "", # session_stats_display | |
| 0, # current_message_index_state | |
| None, # current_classification_state | |
| f"❌ Error starting verification: {str(e)}" # dataset_status_message | |
| ) | |
| def on_correct_feedback(current_dataset, session, msg_index, classification): | |
| """Handle correct classification feedback.""" | |
| try: | |
| success, message, stats = dataset_controller.submit_verification_feedback(True) | |
| if success and not stats.get('is_complete', False): | |
| # Get next message | |
| next_message, next_classification = dataset_controller.get_current_message_for_verification() | |
| if next_message: | |
| decision = next_classification.get('decision', 'unknown').upper() | |
| confidence = next_classification.get('confidence', 0) * 100 | |
| decision_colors = {'GREEN': '🟢', 'YELLOW': '🟡', 'RED': '🔴'} | |
| decision_badge = f"{decision_colors.get(decision, '❓')} **{decision}**" | |
| confidence_text = f"Confidence: **{confidence:.1f}%**" | |
| stats_text = f"""**Progress:** {stats['processed'] + 1}/{stats['total']} | |
| **Correct:** {stats['correct']} | |
| **Incorrect:** {stats['incorrect']} | |
| **Accuracy:** {stats['accuracy']:.1f}%""" | |
| return ( | |
| next_message.text, # current_message_display | |
| decision_badge, # classifier_decision_display | |
| confidence_text, # classifier_confidence_display | |
| stats_text, # session_stats_display | |
| msg_index + 1, # current_message_index_state | |
| next_classification, # current_classification_state | |
| gr.Row(visible=False), # correction_section | |
| message # dataset_status_message | |
| ) | |
| # Session complete | |
| stats_text = f"""**🎉 Session Complete!** | |
| **Total:** {stats['processed']} | |
| **Correct:** {stats['correct']} | |
| **Incorrect:** {stats['incorrect']} | |
| **Final Accuracy:** {stats['accuracy']:.1f}%""" | |
| return ( | |
| "✅ Verification session complete!", # current_message_display | |
| "🎉 **COMPLETE**", # classifier_decision_display | |
| "", # classifier_confidence_display | |
| stats_text, # session_stats_display | |
| msg_index, # current_message_index_state | |
| None, # current_classification_state | |
| gr.Row(visible=False), # correction_section | |
| f"✅ Session complete! Accuracy: {stats['accuracy']:.1f}%" # dataset_status_message | |
| ) | |
| except Exception as e: | |
| return ( | |
| gr.update(), # current_message_display | |
| gr.update(), # classifier_decision_display | |
| gr.update(), # classifier_confidence_display | |
| gr.update(), # session_stats_display | |
| msg_index, # current_message_index_state | |
| classification, # current_classification_state | |
| gr.Row(visible=False), # correction_section | |
| f"❌ Error: {str(e)}" # dataset_status_message | |
| ) | |
| def on_incorrect_feedback(): | |
| """Show correction section.""" | |
| return ( | |
| gr.Row(visible=True), # correction_section | |
| "⚠️ Please select the correct classification" # dataset_status_message | |
| ) | |
| def on_cancel_correction(): | |
| """Cancel correction.""" | |
| return ( | |
| gr.Row(visible=False), # correction_section | |
| "❌ Correction cancelled" # dataset_status_message | |
| ) | |
| def on_submit_correction(current_dataset, session, msg_index, classification, correction, notes): | |
| """Submit correction feedback.""" | |
| try: | |
| if not correction: | |
| return ( | |
| gr.update(), # current_message_display | |
| gr.update(), # classifier_decision_display | |
| gr.update(), # classifier_confidence_display | |
| gr.update(), # session_stats_display | |
| msg_index, # current_message_index_state | |
| classification, # current_classification_state | |
| gr.Row(visible=True), # correction_section | |
| "❌ Please select a classification" # dataset_status_message | |
| ) | |
| success, message, stats = dataset_controller.submit_verification_feedback(False, correction, notes) | |
| if success and not stats.get('is_complete', False): | |
| # Get next message | |
| next_message, next_classification = dataset_controller.get_current_message_for_verification() | |
| if next_message: | |
| decision = next_classification.get('decision', 'unknown').upper() | |
| confidence = next_classification.get('confidence', 0) * 100 | |
| decision_colors = {'GREEN': '🟢', 'YELLOW': '🟡', 'RED': '🔴'} | |
| decision_badge = f"{decision_colors.get(decision, '❓')} **{decision}**" | |
| confidence_text = f"Confidence: **{confidence:.1f}%**" | |
| stats_text = f"""**Progress:** {stats['processed'] + 1}/{stats['total']} | |
| **Correct:** {stats['correct']} | |
| **Incorrect:** {stats['incorrect']} | |
| **Accuracy:** {stats['accuracy']:.1f}%""" | |
| return ( | |
| next_message.text, # current_message_display | |
| decision_badge, # classifier_decision_display | |
| confidence_text, # classifier_confidence_display | |
| stats_text, # session_stats_display | |
| msg_index + 1, # current_message_index_state | |
| next_classification, # current_classification_state | |
| gr.Row(visible=False), # correction_section | |
| message # dataset_status_message | |
| ) | |
| # Session complete | |
| stats_text = f"""**🎉 Session Complete!** | |
| **Total:** {stats['processed']} | |
| **Correct:** {stats['correct']} | |
| **Incorrect:** {stats['incorrect']} | |
| **Final Accuracy:** {stats['accuracy']:.1f}%""" | |
| return ( | |
| "✅ Verification session complete!", # current_message_display | |
| "🎉 **COMPLETE**", # classifier_decision_display | |
| "", # classifier_confidence_display | |
| stats_text, # session_stats_display | |
| msg_index, # current_message_index_state | |
| None, # current_classification_state | |
| gr.Row(visible=False), # correction_section | |
| f"✅ Session complete! Accuracy: {stats['accuracy']:.1f}%" # dataset_status_message | |
| ) | |
| except Exception as e: | |
| return ( | |
| gr.update(), # current_message_display | |
| gr.update(), # classifier_decision_display | |
| gr.update(), # classifier_confidence_display | |
| gr.update(), # session_stats_display | |
| msg_index, # current_message_index_state | |
| classification, # current_classification_state | |
| gr.Row(visible=True), # correction_section | |
| f"❌ Error: {str(e)}" # dataset_status_message | |
| ) | |
| def on_finish_session(session): | |
| """Finish verification session.""" | |
| return ( | |
| gr.Row(visible=False), # message_review_section | |
| "✅ Session finished. You can start a new verification or select another dataset." # dataset_status_message | |
| ) | |
| def on_export_results(format_type, session): | |
| """Export verification results.""" | |
| try: | |
| if not session: | |
| return "❌ No active session to export" | |
| success, message, file_path = dataset_controller.export_session_results(format_type) | |
| if success: | |
| return f"✅ Results exported to {format_type.upper()} format" | |
| else: | |
| return message | |
| except Exception as e: | |
| return f"❌ Export error: {str(e)}" | |
| def on_skip_message(current_dataset, session, msg_index, classification): | |
| """Skip current message without feedback.""" | |
| try: | |
| if not session or not current_dataset: | |
| return ( | |
| gr.update(), # current_message_display | |
| gr.update(), # classifier_decision_display | |
| gr.update(), # classifier_confidence_display | |
| gr.update(), # session_stats_display | |
| msg_index, # current_message_index_state | |
| classification, # current_classification_state | |
| "❌ No active session" # dataset_status_message | |
| ) | |
| # Move to next message without recording feedback | |
| dataset_controller.current_message_index += 1 | |
| if dataset_controller.current_message_index >= len(current_dataset.messages): | |
| # Session complete | |
| stats_text = f"""**🎉 Session Complete!** | |
| **Total:** {session.verified_count} | |
| **Correct:** {session.correct_count} | |
| **Incorrect:** {session.incorrect_count} | |
| **Skipped:** {dataset_controller.current_message_index - session.verified_count} | |
| **Final Accuracy:** {(session.correct_count / session.verified_count * 100) if session.verified_count > 0 else 0:.1f}%""" | |
| return ( | |
| "✅ Verification session complete!", # current_message_display | |
| "🎉 **COMPLETE**", # classifier_decision_display | |
| "", # classifier_confidence_display | |
| stats_text, # session_stats_display | |
| dataset_controller.current_message_index, # current_message_index_state | |
| None, # current_classification_state | |
| "✅ Session complete!" # dataset_status_message | |
| ) | |
| # Get next message | |
| next_message, next_classification = dataset_controller.get_current_message_for_verification() | |
| if next_message: | |
| decision = next_classification.get('decision', 'unknown').upper() | |
| confidence = next_classification.get('confidence', 0) * 100 | |
| decision_colors = {'GREEN': '🟢', 'YELLOW': '🟡', 'RED': '🔴'} | |
| decision_badge = f"{decision_colors.get(decision, '❓')} **{decision}**" | |
| confidence_text = f"Confidence: **{confidence:.1f}%**" | |
| skipped_count = dataset_controller.current_message_index - session.verified_count | |
| stats_text = f"""**Progress:** {dataset_controller.current_message_index + 1}/{len(current_dataset.messages)} | |
| **Correct:** {session.correct_count} | |
| **Incorrect:** {session.incorrect_count} | |
| **Skipped:** {skipped_count} | |
| **Accuracy:** {(session.correct_count / session.verified_count * 100) if session.verified_count > 0 else 0:.1f}%""" | |
| return ( | |
| next_message.text, # current_message_display | |
| decision_badge, # classifier_decision_display | |
| confidence_text, # classifier_confidence_display | |
| stats_text, # session_stats_display | |
| dataset_controller.current_message_index, # current_message_index_state | |
| next_classification, # current_classification_state | |
| "⏭️ Message skipped" # dataset_status_message | |
| ) | |
| else: | |
| return ( | |
| "No more messages", # current_message_display | |
| "", # classifier_decision_display | |
| "", # classifier_confidence_display | |
| "All messages processed", # session_stats_display | |
| msg_index, # current_message_index_state | |
| None, # current_classification_state | |
| "✅ All messages processed" # dataset_status_message | |
| ) | |
| except Exception as e: | |
| return ( | |
| gr.update(), # current_message_display | |
| gr.update(), # classifier_decision_display | |
| gr.update(), # classifier_confidence_display | |
| gr.update(), # session_stats_display | |
| msg_index, # current_message_index_state | |
| classification, # current_classification_state | |
| f"❌ Error skipping: {str(e)}" # dataset_status_message | |
| ) | |
| # Bind dataset interface events | |
| dataset_selector.change( | |
| on_dataset_selection_change, | |
| inputs=[dataset_selector], | |
| outputs=[dataset_info_display, current_dataset_state] | |
| ) | |
| load_dataset_btn.click( | |
| on_load_dataset, | |
| inputs=[current_dataset_state], | |
| outputs=[verification_section, dataset_status_message] | |
| ) | |
| start_verification_btn.click( | |
| on_start_verification, | |
| inputs=[current_dataset_state, verifier_name_input], | |
| outputs=[ | |
| verification_session_state, | |
| message_review_section, | |
| current_message_display, | |
| classifier_decision_display, | |
| classifier_confidence_display, | |
| session_stats_display, | |
| current_message_index_state, | |
| current_classification_state, | |
| dataset_status_message | |
| ] | |
| ) | |
| correct_btn.click( | |
| on_correct_feedback, | |
| inputs=[current_dataset_state, verification_session_state, current_message_index_state, current_classification_state], | |
| outputs=[ | |
| current_message_display, | |
| classifier_decision_display, | |
| classifier_confidence_display, | |
| session_stats_display, | |
| current_message_index_state, | |
| current_classification_state, | |
| correction_section, | |
| dataset_status_message | |
| ] | |
| ) | |
| incorrect_btn.click( | |
| on_incorrect_feedback, | |
| outputs=[correction_section, dataset_status_message] | |
| ) | |
| cancel_correction_btn.click( | |
| on_cancel_correction, | |
| outputs=[correction_section, dataset_status_message] | |
| ) | |
| submit_correction_btn.click( | |
| on_submit_correction, | |
| inputs=[ | |
| current_dataset_state, | |
| verification_session_state, | |
| current_message_index_state, | |
| current_classification_state, | |
| correction_selector, | |
| correction_notes | |
| ], | |
| outputs=[ | |
| current_message_display, | |
| classifier_decision_display, | |
| classifier_confidence_display, | |
| session_stats_display, | |
| current_message_index_state, | |
| current_classification_state, | |
| correction_section, | |
| dataset_status_message | |
| ] | |
| ) | |
| skip_btn.click( | |
| on_skip_message, | |
| inputs=[current_dataset_state, verification_session_state, current_message_index_state, current_classification_state], | |
| outputs=[ | |
| current_message_display, | |
| classifier_decision_display, | |
| classifier_confidence_display, | |
| session_stats_display, | |
| current_message_index_state, | |
| current_classification_state, | |
| dataset_status_message | |
| ] | |
| ) | |
| finish_btn.click( | |
| on_finish_session, | |
| inputs=[verification_session_state], | |
| outputs=[message_review_section, dataset_status_message] | |
| ) | |
| # Export buttons | |
| export_csv_btn.click( | |
| lambda session: on_export_results("csv", session), | |
| inputs=[verification_session_state], | |
| outputs=[dataset_status_message] | |
| ) | |
| export_json_btn.click( | |
| lambda session: on_export_results("json", session), | |
| inputs=[verification_session_state], | |
| outputs=[dataset_status_message] | |
| ) | |
| export_xlsx_btn.click( | |
| lambda session: on_export_results("xlsx", session), | |
| inputs=[verification_session_state], | |
| outputs=[dataset_status_message] | |
| ) | |
| return interface | |
| def create_enhanced_verification_tab() -> gr.Blocks: | |
| """ | |
| Create enhanced verification tab for integration with existing application. | |
| Returns: | |
| Gradio Blocks component for enhanced verification modes | |
| """ | |
| interface_controller = EnhancedVerificationInterface() | |
| return interface_controller.create_interface() |