Spiritual_Health_Project / src /interface /enhanced_verification_interface.py
DocUA's picture
Implement model overrides for AI clients and enhance interfaces for manual input and file upload
37184d4
# 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()