diff --git "a/src/interface/simplified_gradio_app.py" "b/src/interface/simplified_gradio_app.py" --- "a/src/interface/simplified_gradio_app.py" +++ "b/src/interface/simplified_gradio_app.py" @@ -22,21 +22,19 @@ from dotenv import load_dotenv load_dotenv() import gradio as gr -import uuid -from datetime import datetime -from typing import Dict, Any, Optional, List + +# Import modularized components +from src.interface.session_manager import SimplifiedSessionData +from src.interface import stats_handlers +from src.interface import chat_handlers +from src.interface import prompt_handlers +from src.interface import verification_handlers +from src.interface import profile_handlers +from src.interface import model_handlers from src.core.simplified_medical_app import SimplifiedMedicalApp from src.core.spiritual_state import SpiritualState -from src.interface.verification_ui import VerificationUIComponents -from src.interface.chaplain_feedback_ui import ChaplainFeedbackUIComponents from src.interface.enhanced_verification_interface import create_enhanced_verification_tab -from src.core.test_datasets import TestDatasetManager -from src.core.verification_models import VerificationSession, VerificationRecord, TestMessage -from src.core.verification_store import JSONVerificationStore -from src.core.verification_csv_exporter import VerificationCSVExporter -from src.core.chaplain_models import ClassificationFlowResult, DistressIndicator, FollowUpQuestion -from src.core.error_pattern_analyzer import ErrorPatternAnalyzer from src.interface.help_content import HELP_CONTENT try: @@ -58,36 +56,7 @@ except ImportError: return FEATURE_FLAGS.get(feature_name, False) -class SimplifiedSessionData: - """Container for user session data.""" - - def __init__(self, session_id: str = None): - self.session_id = session_id or str(uuid.uuid4()) - self.app_instance = SimplifiedMedicalApp() - self.created_at = datetime.now().isoformat() - self.last_activity = datetime.now().isoformat() - - # Set default patient info from profile - self.app_instance.set_patient_info(name="Serhii", phone="(555) 123-4567") - - # Update clinical_background to match default profile - from src.core.core_classes import ClinicalBackground - self.app_instance.clinical_background = ClinicalBackground( - patient_name="Serhii", - age=52, - conditions=["Atrial fibrillation", "Deep vein thrombosis", "Obesity", "Hypertension"], - primary_goal="Weight reduction and cardiovascular fitness improvement", - exercise_preferences=["Swimming", "Walking", "Light cardio"], - exercise_limitations=["Anticoagulation therapy", "Post-thrombotic recovery"] - ) - - # Update conversation logger patient name - if hasattr(self.app_instance, 'conversation_logger'): - self.app_instance.conversation_logger.patient_name = "Serhii" - - def update_activity(self): - """Update last activity timestamp.""" - self.last_activity = datetime.now().isoformat() + def create_simplified_interface(): @@ -607,1478 +576,8 @@ def create_simplified_interface(): # Event handlers - def handle_message(message: str, history, session: SimplifiedSessionData): - """Handle user message.""" - if session is None: - session = SimplifiedSessionData() - - session.update_activity() - - # Apply per-session model overrides (if configured in Model Settings) - custom_models = getattr(session, 'custom_models', None) - if custom_models: - session.app_instance.set_model_overrides(custom_models) - else: - session.app_instance.set_model_overrides({}) - - # Apply per-session prompt overrides (if configured in Edit Prompts) - custom_prompts = getattr(session, 'custom_prompts', None) - if custom_prompts: - session.app_instance.set_prompt_overrides(custom_prompts) - else: - session.app_instance.set_prompt_overrides({}) - new_history, status = session.app_instance.process_message(message, history) - - # Get updated conversation stats - stats = get_conversation_stats(session) - - # Check for provider summary (RED flag case) - provider_summary_text = "" - spiritual_care_msg = "" - show_provider_panel = False - last_summary = session.app_instance.get_last_provider_summary() - - # Debug logging - print(f"DEBUG: last_summary exists: {last_summary is not None}") - if last_summary: - print(f"DEBUG: summary patient: {last_summary.patient_name}") - print(f"DEBUG: summary indicators: {last_summary.indicators}") - provider_summary_text = session.app_instance.provider_summary_generator.format_for_display(last_summary) - - # Generate LLM-based spiritual care message - try: - spiritual_care_msg = session.app_instance.generate_spiritual_care_message( - language="English", - session_id=session.session_id - ) - if not spiritual_care_msg: - spiritual_care_msg = "" - print(f"DEBUG: spiritual care message generated: {len(spiritual_care_msg)} chars") - except Exception as e: - print(f"DEBUG: Error generating spiritual care message: {e}") - spiritual_care_msg = "" - - show_provider_panel = True - print(f"DEBUG: formatted summary length: {len(provider_summary_text)}") - print(f"DEBUG: show_provider_panel: {show_provider_panel}") - else: - print("DEBUG: No provider summary found") - - # Debug: print what we're returning - print(f"DEBUG RETURN: show_panel={show_provider_panel}, text_len={len(provider_summary_text)}") - if provider_summary_text: - print(f"DEBUG RETURN: first 100 chars: {provider_summary_text[:100]}") - - # Format as HTML for display - if provider_summary_text: - import html - escaped_text = html.escape(provider_summary_text) - html_content = f"
{escaped_text}
" - else: - html_content = "
No summary available
" - - # Use gr.update for both visibility and value - if not provider_summary_text: - provider_summary_text = "" - html_content = "" - - # Generate status message for provider summary - if show_provider_panel and provider_summary_text: - status_msg = f"""**šŸ”“ Provider Summary Generated** - -**Patient:** {session.app_instance.patient_info.get('name', 'Test Patient')} -**Classification:** RED FLAG -**Indicators:** {len(session.app_instance.get_last_provider_summary().indicators) if session.app_instance.get_last_provider_summary() else 0} distress indicators detected -**Summary Length:** {len(provider_summary_text)} characters - -Use the **Download Summary** button below to access the complete provider summary for the spiritual care team.""" - else: - status_msg = "No provider summary available" - - return ( - new_history, - status, - session, - "", - stats, - gr.update(visible=show_provider_panel), # provider_summary_content visibility - status_msg, # provider_summary_status content - gr.update(value=html_content, visible=True) if show_provider_panel else gr.update(visible=False), # provider_summary_display content - spiritual_care_msg # spiritual_care_message content - ) - - def handle_clear(session: SimplifiedSessionData): - """ - Handle clear chat button. - - Resets entire session including: - - Chat history - - Spiritual monitoring state - - Provider summary panel (hides and clears content) - - Conversation statistics - """ - if session is None: - session = SimplifiedSessionData() - - session.update_activity() - new_history, status = session.app_instance.reset_session() - - # Hide and clear provider summary panel - return ( - new_history, # Clear chat history - status, # Reset status - session, # Updated session - gr.update(visible=False), # Hide provider_summary_content group - "No provider summary available", # Clear provider_summary_status - "", # Clear provider_summary_display HTML - "" # Clear spiritual_care_message - ) - - def get_status(session: SimplifiedSessionData): - """Get current status and stats.""" - if session is None: - return "āŒ Session not initialized", "No stats", gr.update(visible=False), "No provider summary available", "", "" - - session.update_activity() - status_info = session.app_instance._get_status_info() - - # Get stats - stats_text = get_conversation_stats(session) - - # Check for provider summary - last_summary = session.app_instance.get_last_provider_summary() - show_provider_panel = last_summary is not None - - provider_summary_text = "" - spiritual_care_msg = "" - - if last_summary: - provider_summary_text = session.app_instance.provider_summary_generator.format_for_display(last_summary) - - # Generate spiritual care message - try: - spiritual_care_msg = session.app_instance.generate_spiritual_care_message( - language="English", - session_id=session.session_id - ) - if not spiritual_care_msg: - spiritual_care_msg = "" - except Exception as e: - print(f"Error generating spiritual care message in get_status: {e}") - spiritual_care_msg = "" - - if provider_summary_text: - import html - escaped_text = html.escape(provider_summary_text) - html_content = f"
{escaped_text}
" - - status_msg = f"""**šŸ”“ Provider Summary Generated** - -**Patient:** {session.app_instance.patient_info.get('name', 'Test Patient')} -**Classification:** RED FLAG -**Indicators:** {len(last_summary.indicators)} distress indicators detected -**Summary Length:** {len(provider_summary_text)} characters - -Use the **Download Summary** button below to access the complete provider summary for the spiritual care team.""" - else: - html_content = "" - status_msg = "No provider summary available" - - return ( - status_info, - stats_text, - gr.update(visible=show_provider_panel), - status_msg, - html_content, - spiritual_care_msg - ) - - def send_example(example_text: str, history, session: SimplifiedSessionData): - """Send example message.""" - return handle_message(example_text, history, session) - - def get_conversation_stats(session: SimplifiedSessionData): - """Get conversation statistics.""" - if session is None or not hasattr(session.app_instance, 'conversation_logger'): - return "No conversation yet" - - try: - summary = session.app_instance.get_conversation_summary() - if not summary or summary.get('total_exchanges', 0) == 0: - return "No conversation yet" - - stats_text = f"""**šŸ“Š Conversation Statistics** - -**Messages:** {summary['total_exchanges']} exchanges -**Duration:** {summary['session_duration_minutes']} minutes + # Handlers moved to modular modules -**Classifications:** -šŸŸ¢ Green: {summary['classification_counts']['green']} ({summary['classification_percentages']['green']}%) -šŸŸ” Yellow: {summary['classification_counts']['yellow']} ({summary['classification_percentages']['yellow']}%) -šŸ”“ Red: {summary['classification_counts']['red']} ({summary['classification_percentages']['red']}%) - -**Average Confidence:** {summary['average_confidence']} - -**Top Indicators:**""" - - for indicator, count in list(summary['top_indicators'].items())[:3]: - stats_text += f"\nā€¢ {indicator}: {count}" - - return stats_text - - except Exception as e: - return f"Error getting stats: {str(e)}" - - def download_conversation_json(session: SimplifiedSessionData): - """Download conversation as JSON.""" - if session is None or not hasattr(session.app_instance, 'conversation_logger'): - return None - - try: - log_path = session.app_instance.get_conversation_log_path() - return log_path - except Exception as e: - print(f"Error downloading JSON: {e}") - return None - - def download_conversation_csv(session: SimplifiedSessionData): - """Download conversation as CSV.""" - if session is None or not hasattr(session.app_instance, 'conversation_logger'): - return None - - try: - csv_path = session.app_instance.export_conversation_csv() - return csv_path - except Exception as e: - print(f"Error downloading CSV: {e}") - return None - - def open_verification_window(session: SimplifiedSessionData): - """Open verification window for current conversation.""" - if session is None or not hasattr(session.app_instance, 'conversation_logger'): - return """
- āŒ No conversation to verify
- Start a conversation first -
""" - - try: - # Check if conversation has any entries - if not session.app_instance.conversation_logger.entries: - return """
- āš ļø No conversation exchanges to verify
- Send some messages in the chat first -
""" - - print(f"šŸ” Opening verification for {len(session.app_instance.conversation_logger.entries)} exchanges...") - - # Create verification session - from src.core.conversation_verification import ConversationVerificationManager - import json - import os - from datetime import datetime - - manager = ConversationVerificationManager() - verification_session = manager.create_verification_session( - session.app_instance.conversation_logger, - "Medical Professional" - ) - - print(f"āœ… Created verification session: {verification_session.session_id}") - - # HF Spaces / Gradio limitation: - # Launching a *second* Gradio server from inside a running Gradio app is unreliable - # and is currently causing the Button._id error in Spaces. - # Instead, export the verification session to a JSON file that the user can download. - - export_dir = os.path.join(os.getcwd(), "verification_sessions") - os.makedirs(export_dir, exist_ok=True) - - export_filename = f"verification_session_{datetime.utcnow().strftime('%Y%m%d_%H%M%S')}_{verification_session.session_id}.json" - export_path = os.path.join(export_dir, export_filename) - - # Serialize to JSON in a resilient way (dataclasses / pydantic / plain python). - def _to_dict(obj): - if hasattr(obj, "model_dump"): - return obj.model_dump() - if hasattr(obj, "dict") and callable(getattr(obj, "dict")): - return obj.dict() - if hasattr(obj, "__dict__"): - return obj.__dict__ - return str(obj) - - payload = { - "session_id": verification_session.session_id, - "patient_name": verification_session.patient_name, - "verifier_name": verification_session.verifier_name, - "start_time": verification_session.start_time.isoformat() if hasattr(verification_session, "start_time") else None, - "verification_records": [ - { - # Conversation verification records use `exchange_id`. - # Keep a `record_id` alias for backward compatibility with older exports. - "exchange_id": getattr(r, "exchange_id", None), - "record_id": getattr(r, "exchange_id", None), - "timestamp": r.timestamp.isoformat() if hasattr(r, "timestamp") else None, - "user_message": r.user_message, - "assistant_response": r.assistant_response, - "original_classification": r.original_classification, - "original_confidence": r.original_confidence, - "original_indicators": r.original_indicators, - "original_reasoning": r.original_reasoning, - "is_correct": r.is_correct, - "correct_classification": r.correct_classification, - "correction_reason": r.correction_reason, - "verifier_notes": r.verifier_notes, - } - for r in verification_session.verification_records - ], - } - - with open(export_path, "w", encoding="utf-8") as f: - json.dump(payload, f, ensure_ascii=False, indent=2, default=_to_dict) - - print(f"āœ… Verification session exported: {export_path}") - - return f"""
- āœ… Verification session exported
- Exchanges: {len(verification_session.verification_records)}
- Download JSON from the app's files panel (or add a dedicated download button). -
""" - - except Exception as e: - print(f"āŒ Error opening verification: {str(e)}") - import traceback - traceback.print_exc() - - return f"""
- āŒ Error opening verification
- {str(e)} -
""" - - # Prompt editing handlers - def format_prompt_with_html(prompt_text: str) -> str: - """Format prompt with HTML tags for better visualization.""" - import re - import html - - # Escape HTML first to prevent injection - formatted = html.escape(prompt_text) - - # Highlight XML-like opening tags - formatted = re.sub( - r'<([a-z_]+)(?:\s+[^>]*)?>', - r'<\1>', - formatted, - flags=re.IGNORECASE - ) - - # Highlight XML-like closing tags - formatted = re.sub( - r'</([a-z_]+)>', - r'</\1>', - formatted, - flags=re.IGNORECASE - ) - - # Highlight XML attributes - formatted = re.sub( - r'([a-z_]+)="([^"]+)"', - r'\1="\2"', - formatted, - flags=re.IGNORECASE - ) - - # Highlight bullet points - formatted = re.sub( - r'^([-ā€¢]\s+.+)$', - r'
\1
', - formatted, - flags=re.MULTILINE - ) - - # Highlight numbered lists - formatted = re.sub( - r'^(\d+\.\s+.+)$', - r'
\1
', - formatted, - flags=re.MULTILINE - ) - - # Highlight important keywords - keywords = [ - 'IMPORTANT', 'CRITICAL', 'REQUIRED', 'MUST', 'SHALL', - 'WARNING', 'NOTE', 'TASK', 'GOAL', 'OUTPUT', 'ANY' - ] - for keyword in keywords: - formatted = re.sub( - f'\\b({keyword})\\b', - r'\1', - formatted, - flags=re.IGNORECASE - ) - - # Highlight JSON/code blocks - formatted = re.sub( - r'(\{[^}]+\})', - r'\1', - formatted - ) - - # Convert newlines to
for proper display - formatted = formatted.replace('\n', '
') - - # Wrap in container with monospace font for code-like appearance - formatted = f'
{formatted}
' - - return formatted - - def _prompt_name_to_agent(prompt_name: str) -> str: - """Map UI prompt selection to internal agent/prompt key.""" - mapping = { - "šŸ” Spiritual Monitor (Classifier)": "SpiritualDistressAnalyzer", - "šŸŸ” Soft Spiritual Triage": "SoftSpiritualTriage", - "šŸ“Š Triage Response Evaluator": "TriageResponseEvaluator", - "šŸ„ Medical Assistant": "MedicalAssistant", - "šŸ©ŗ Soft Medical Triage": "SoftMedicalTriage", - } - return mapping.get(prompt_name, prompt_name) - - def load_prompt(prompt_name: str, session: Optional[SimplifiedSessionData] = None): - """Load selected prompt for editing using enhanced prompt editor.""" - try: - from src.interface.enhanced_prompt_editor import EnhancedPromptEditor - - # Initialize enhanced editor - editor = EnhancedPromptEditor() - - # Get session ID - session_id = getattr(session, 'session_id', 'default_session') if session else 'default_session' - - # Use enhanced editor to load prompt - prompt_content, info_html, status_html = editor.load_prompt_for_editing(prompt_name, session_id) - - return prompt_content, info_html, status_html - - except Exception as e: - # Fallback to old system if enhanced editor fails - logger.warning(f"Enhanced prompt editor failed, using fallback: {e}") - - from src.core.spiritual_monitor import SYSTEM_PROMPT_SPIRITUAL_MONITOR - from src.core.soft_triage_manager import ( - SYSTEM_PROMPT_TRIAGE_QUESTION, - SYSTEM_PROMPT_TRIAGE_EVALUATE - ) - from src.config.prompts import ( - SYSTEM_PROMPT_MEDICAL_ASSISTANT, - SYSTEM_PROMPT_SOFT_MEDICAL_TRIAGE - ) - - prompts = { - "šŸ” Spiritual Monitor (Classifier)": SYSTEM_PROMPT_SPIRITUAL_MONITOR, - "šŸŸ” Soft Spiritual Triage": SYSTEM_PROMPT_TRIAGE_QUESTION, - "šŸ“Š Triage Response Evaluator": SYSTEM_PROMPT_TRIAGE_EVALUATE, - "šŸ„ Medical Assistant": SYSTEM_PROMPT_MEDICAL_ASSISTANT, - "šŸ©ŗ Soft Medical Triage": SYSTEM_PROMPT_SOFT_MEDICAL_TRIAGE - } - - prompt_text = prompts.get(prompt_name, "") - - info = f"""**Loaded:** {prompt_name} -**Length:** {len(prompt_text)} characters -**Status:** Fallback mode (enhanced editor unavailable)""" - - status = """
-

āš ļø Fallback Mode

-

Using basic prompt editor. Enhanced features unavailable.

-
""" - - return prompt_text, info, status - - def apply_prompt_changes(prompt_name: str, prompt_text: str, session: SimplifiedSessionData): - """Apply custom prompt changes using enhanced prompt editor.""" - try: - from src.interface.enhanced_prompt_editor import EnhancedPromptEditor - - if session is None: - session = SimplifiedSessionData() - - # Initialize enhanced editor - editor = EnhancedPromptEditor() - - # Get session ID - session_id = getattr(session, 'session_id', 'default_session') - - # Use enhanced editor to apply changes - status_html, success = editor.apply_prompt_changes(prompt_name, prompt_text, session_id) - - if success: - # Also store in session for backward compatibility - if not hasattr(session, 'custom_prompts'): - session.custom_prompts = {} - - agent_key = _prompt_name_to_agent(prompt_name) - session.custom_prompts[agent_key] = prompt_text - - # Apply to session app instance if available - if hasattr(session, 'app_instance') and hasattr(session.app_instance, 'set_prompt_overrides'): - session.app_instance.set_prompt_overrides(session.custom_prompts) - - return status_html, session - - except Exception as e: - # Fallback to old system - logger.warning(f"Enhanced prompt editor failed, using fallback: {e}") - - if session is None: - session = SimplifiedSessionData() - - if not prompt_text.strip(): - error_html = """
-

āŒ Error

-

Prompt cannot be empty

-
""" - return error_html, session - - # Store custom prompt in session (session-scoped) - if not hasattr(session, 'custom_prompts'): - session.custom_prompts = {} - - agent_key = _prompt_name_to_agent(prompt_name) - session.custom_prompts[agent_key] = prompt_text - - status = f"""
-

āš ļø Fallback Mode - Changes Applied

-

Prompt: {prompt_name}

-

Length: {len(prompt_text)} characters

-

Enhanced features unavailable, using basic session storage.

-
""" - - return status, session - - def reset_prompt(prompt_name: str, session: SimplifiedSessionData): - """Reset prompt to default using enhanced prompt editor.""" - try: - from src.interface.enhanced_prompt_editor import EnhancedPromptEditor - - if session is None: - session = SimplifiedSessionData() - - # Initialize enhanced editor - editor = EnhancedPromptEditor() - - # Get session ID - session_id = getattr(session, 'session_id', 'default_session') - - # Use enhanced editor to reset prompt - prompt_content, info_html, status_html = editor.reset_prompt_to_default(prompt_name, session_id) - - # Also remove from session for backward compatibility - agent_key = _prompt_name_to_agent(prompt_name) - if hasattr(session, 'custom_prompts') and agent_key in session.custom_prompts: - del session.custom_prompts[agent_key] - - # Apply to session app instance if available - if hasattr(session, 'app_instance') and hasattr(session.app_instance, 'set_prompt_overrides'): - session.app_instance.set_prompt_overrides(getattr(session, 'custom_prompts', {})) - - return prompt_content, info_html, status_html, session - - except Exception as e: - # Fallback to old system - logger.warning(f"Enhanced prompt editor failed, using fallback: {e}") - - if session is None: - session = SimplifiedSessionData() - - # Remove from custom prompts - agent_key = _prompt_name_to_agent(prompt_name) - if hasattr(session, 'custom_prompts') and agent_key in session.custom_prompts: - del session.custom_prompts[agent_key] - - # Reload default - prompt_text, info, status = load_prompt(prompt_name, session) - - reset_status = """
-

šŸ”„ Fallback Mode - Reset Complete

-

Prompt restored using basic system. Enhanced features unavailable.

-
""" - - return prompt_text, info, reset_status, session - - def promote_prompt_to_file(prompt_name: str, session: SimplifiedSessionData): - """Promote session prompt override to permanent file.""" - try: - from src.interface.enhanced_prompt_editor import EnhancedPromptEditor - - if session is None: - return """
-

āŒ Error

-

No session data available

-
""", session - - # Initialize enhanced editor - editor = EnhancedPromptEditor() - - # Get session ID - session_id = getattr(session, 'session_id', 'default_session') - - # Use enhanced editor to promote prompt - status_html, success = editor.promote_session_to_file(prompt_name, session_id) - - return status_html, session - - except Exception as e: - logger.warning(f"Enhanced prompt editor failed: {e}") - return f"""
-

āŒ Error

-

Failed to promote prompt: {str(e)}

-
""", session - - def validate_prompt_syntax(prompt_text: str): - """Validate prompt syntax and structure.""" - try: - from src.interface.enhanced_prompt_editor import EnhancedPromptEditor - - # Initialize enhanced editor - editor = EnhancedPromptEditor() - - # Use enhanced editor to validate prompt - validation_html, is_valid = editor.validate_prompt_syntax(prompt_text) - - return validation_html - - except Exception as e: - logger.warning(f"Enhanced prompt editor failed: {e}") - return f"""
-

āŒ Validation Error

-

Failed to validate prompt: {str(e)}

-
""" - - # Verification mode handlers - def load_verification_dataset(dataset_name: str, store: JSONVerificationStore): - """Load a verification dataset.""" - try: - # Find dataset ID from name - datasets = TestDatasetManager.get_dataset_list() - dataset_id = None - for d in datasets: - if d['name'] in dataset_name: - dataset_id = d['dataset_id'] - break - - if not dataset_id: - return ( - None, # verification_session - "āŒ Dataset not found", # dataset_info - "", "", "", "", # message_text, decision_badge, confidence, indicators - "", # progress_display - "āŒ Dataset not found", # error_message - 0, # current_message_index - None, # current_dataset_id - [], # message_queue - [], # verification_records - ) - - # Load dataset - dataset = TestDatasetManager.load_dataset(dataset_id) - - # Create new verification session - new_session = VerificationSession( - session_id=str(uuid.uuid4()), - verifier_name="Medical Professional", - dataset_id=dataset_id, - dataset_name=dataset.name, - total_messages=dataset.message_count, - message_queue=[m.message_id for m in dataset.messages], - ) - - # Save session - store.save_session(new_session) - - # Get first message - if dataset.messages: - first_message = dataset.messages[0] - message_text, decision_badge, confidence, indicators = VerificationUIComponents.render_message_review( - first_message, - first_message.pre_classified_label, - 0.85, # Default confidence - ["Distress indicator 1", "Distress indicator 2"] - ) - - progress = VerificationUIComponents.update_progress_display(0, dataset.message_count) - - dataset_info_text = f"**{dataset.name}**\n\n{dataset.description}\n\nšŸ“Š {dataset.message_count} messages to review" - - return ( - new_session, # verification_session - dataset_info_text, # dataset_info - message_text, # message_text - decision_badge, # decision_badge - confidence, # confidence - indicators, # indicators - progress, # progress_display - "", # error_message (empty = no error) - 0, # current_message_index - dataset_id, # current_dataset_id - [m.message_id for m in dataset.messages], # message_queue - [], # verification_records - ) - else: - return ( - None, # verification_session - "āŒ Dataset is empty", # dataset_info - "", "", "", "", # message_text, decision_badge, confidence, indicators - "", # progress_display - "āŒ Dataset is empty", # error_message - 0, # current_message_index - dataset_id, # current_dataset_id - [], # message_queue - [], # verification_records - ) - - except Exception as e: - return ( - None, # verification_session - f"āŒ Error loading dataset: {str(e)}", # dataset_info - "", "", "", "", # message_text, decision_badge, confidence, indicators - "", # progress_display - f"āŒ Error: {str(e)}", # error_message - 0, # current_message_index - None, # current_dataset_id - [], # message_queue - [], # verification_records - ) - - def handle_correct_feedback(session: VerificationSession, current_idx: int, dataset_id: str, message_queue: List[str], records: List[dict], store: JSONVerificationStore): - """Handle correct feedback.""" - try: - if not session or current_idx >= len(message_queue): - return ( - session, - "āŒ Error: Invalid session state", - "", "", "", "", - "", - "āœ“ Correct: 0", - "āœ— Incorrect: 0", - "šŸ“Š Accuracy: 0%", - current_idx, - records, - ) - - # Get current message - dataset = TestDatasetManager.load_dataset(dataset_id) - current_message_id = message_queue[current_idx] - current_message = next((m for m in dataset.messages if m.message_id == current_message_id), None) - - if not current_message: - return ( - session, - "āŒ Error: Message not found", - "", "", "", "", - "", - "āœ“ Correct: 0", - "āœ— Incorrect: 0", - "šŸ“Š Accuracy: 0%", - current_idx, - records, - ) - - # Create verification record - record = VerificationRecord( - message_id=current_message.message_id, - original_message=current_message.text, - classifier_decision=current_message.pre_classified_label, - classifier_confidence=0.85, - classifier_indicators=["Distress indicator 1", "Distress indicator 2"], - ground_truth_label=current_message.pre_classified_label, - verifier_notes="", - is_correct=True, - ) - - # Add to session - session.verifications.append(record) - session.verified_count += 1 - session.correct_count += 1 - - # Save session - store.save_session(session) - - # Move to next message - next_idx = current_idx + 1 - - if next_idx >= len(message_queue): - # Session complete - session.is_complete = True - session.completed_at = datetime.now() - store.save_session(session) - - correct_str, incorrect_str, accuracy_str = VerificationUIComponents.update_statistics_display( - session.correct_count, - session.incorrect_count - ) - - return ( - session, - "āœ… Verification complete!", - "", "", "", "", - "", - correct_str, - incorrect_str, - accuracy_str, - next_idx, - [r.to_dict() for r in session.verifications], - ) - else: - # Load next message - next_message = next((m for m in dataset.messages if m.message_id == message_queue[next_idx]), None) - if next_message: - message_text, decision_badge, confidence, indicators = VerificationUIComponents.render_message_review( - next_message, - next_message.pre_classified_label, - 0.85, - ["Distress indicator 1", "Distress indicator 2"] - ) - - progress = VerificationUIComponents.update_progress_display(next_idx, len(message_queue)) - correct_str, incorrect_str, accuracy_str = VerificationUIComponents.update_statistics_display( - session.correct_count, - session.incorrect_count - ) - - return ( - session, - "", - message_text, - decision_badge, - confidence, - indicators, - progress, - correct_str, - incorrect_str, - accuracy_str, - next_idx, - [r.to_dict() for r in session.verifications], - ) - - return ( - session, - "āŒ Error processing feedback", - "", "", "", "", - "", - "āœ“ Correct: 0", - "āœ— Incorrect: 0", - "šŸ“Š Accuracy: 0%", - current_idx, - records, - ) - - except Exception as e: - return ( - session, - f"āŒ Error: {str(e)}", - "", "", "", "", - "", - "āœ“ Correct: 0", - "āœ— Incorrect: 0", - "šŸ“Š Accuracy: 0%", - current_idx, - records, - ) - - def handle_incorrect_feedback(session: VerificationSession, current_idx: int, dataset_id: str, message_queue: List[str], records: List[dict]): - """Show correction selector.""" - return "āŒ Please select the correct classification below" - - def handle_submit_correction(session: VerificationSession, current_idx: int, dataset_id: str, message_queue: List[str], records: List[dict], correction: str, notes: str, store: JSONVerificationStore): - """Handle correction submission.""" - try: - if not correction: - return ( - "āŒ Please select a correction before submitting", - session, - current_idx, - dataset_id, - message_queue, - records, - "", "", "", "", - "", - "āœ“ Correct: 0", - "āœ— Incorrect: 0", - "šŸ“Š Accuracy: 0%", - "", - "", - ) - - # Get current message - dataset = TestDatasetManager.load_dataset(dataset_id) - current_message_id = message_queue[current_idx] - current_message = next((m for m in dataset.messages if m.message_id == current_message_id), None) - - if not current_message: - return ( - "āŒ Error: Message not found", - session, - current_idx, - dataset_id, - message_queue, - records, - "", "", "", "", - "", - "āœ“ Correct: 0", - "āœ— Incorrect: 0", - "šŸ“Š Accuracy: 0%", - "", - "", - ) - - # Create verification record - record = VerificationRecord( - message_id=current_message.message_id, - original_message=current_message.text, - classifier_decision=current_message.pre_classified_label, - classifier_confidence=0.85, - classifier_indicators=["Distress indicator 1", "Distress indicator 2"], - ground_truth_label=correction, - verifier_notes=notes, - is_correct=current_message.pre_classified_label == correction, - ) - - # Add to session - session.verifications.append(record) - session.verified_count += 1 - if record.is_correct: - session.correct_count += 1 - else: - session.incorrect_count += 1 - - # Save session - store.save_session(session) - - # Move to next message - next_idx = current_idx + 1 - - if next_idx >= len(message_queue): - # Session complete - session.is_complete = True - session.completed_at = datetime.now() - store.save_session(session) - - correct_str, incorrect_str, accuracy_str = VerificationUIComponents.update_statistics_display( - session.correct_count, - session.incorrect_count - ) - - summary = VerificationUIComponents.render_summary_card(session, session.verifications) - - return ( - "āœ… Verification complete!", - session, - next_idx, - dataset_id, - message_queue, - [r.to_dict() for r in session.verifications], - "", "", "", "", - "", - correct_str, - incorrect_str, - accuracy_str, - "", - summary, - ) - else: - # Load next message - next_message = next((m for m in dataset.messages if m.message_id == message_queue[next_idx]), None) - if next_message: - message_text, decision_badge, confidence, indicators = VerificationUIComponents.render_message_review( - next_message, - next_message.pre_classified_label, - 0.85, - ["Distress indicator 1", "Distress indicator 2"] - ) - - progress = VerificationUIComponents.update_progress_display(next_idx, len(message_queue)) - correct_str, incorrect_str, accuracy_str = VerificationUIComponents.update_statistics_display( - session.correct_count, - session.incorrect_count - ) - - return ( - "", - session, - next_idx, - dataset_id, - message_queue, - [r.to_dict() for r in session.verifications], - message_text, - decision_badge, - confidence, - indicators, - progress, - correct_str, - incorrect_str, - accuracy_str, - "", - "", - ) - - return ( - "āŒ Error processing correction", - session, - current_idx, - dataset_id, - message_queue, - records, - "", "", "", "", - "", - "āœ“ Correct: 0", - "āœ— Incorrect: 0", - "šŸ“Š Accuracy: 0%", - "", - "", - ) - - except Exception as e: - return ( - f"āŒ Error: {str(e)}", - session, - current_idx, - dataset_id, - message_queue, - records, - "", "", "", "", - "", - "āœ“ Correct: 0", - "āœ— Incorrect: 0", - "šŸ“Š Accuracy: 0%", - "", - "", - ) - - def handle_download_csv(session: VerificationSession, store: JSONVerificationStore): - """Handle CSV download - returns file path for DownloadButton.""" - try: - if not session or session.verified_count == 0: - return None - - csv_content = VerificationCSVExporter.generate_csv_content(session) - filename = VerificationCSVExporter.generate_csv_filename() - - import os - import tempfile - - # Use temp directory for Hugging Face compatibility - temp_dir = tempfile.gettempdir() - file_path = os.path.join(temp_dir, filename) - - with open(file_path, 'w', encoding='utf-8') as f: - f.write(csv_content) - - return file_path - - except Exception as e: - import traceback - print(f"CSV Export Error: {traceback.format_exc()}") - return None - - # Bind verification events (only if Standard Verification UI is enabled). - # The Standard Verification tab is currently hidden, so keep bindings disabled - # to avoid referencing undefined components. - if False and is_feature_enabled("standard_verification_enabled"): - load_dataset_btn.click( - load_verification_dataset, - inputs=[dataset_selector, verification_store], - outputs=[ - verification_session, - dataset_info, - message_text, - decision_badge, - confidence, - indicators, - progress_display, - error_message, - current_message_index, - current_dataset_id, - message_queue, - verification_records, - ] - ).then( - lambda: gr.Row(visible=True), # Show message_review_section - outputs=[message_review_section] - ) - - correct_btn.click( - handle_correct_feedback, - inputs=[verification_session, current_message_index, current_dataset_id, message_queue, verification_records, verification_store], - outputs=[ - verification_session, - error_message, - message_text, - decision_badge, - confidence, - indicators, - progress_display, - correct_count_display, - incorrect_count_display, - accuracy_display, - current_message_index, - verification_records, - ] - ) - - incorrect_btn.click( - handle_incorrect_feedback, - inputs=[verification_session, current_message_index, current_dataset_id, message_queue, verification_records], - outputs=[error_message] - ).then( - lambda: (gr.Row(visible=True), gr.Row(visible=True)), - outputs=[correction_section, submit_correction_row] - ) - - submit_correction_btn.click( - handle_submit_correction, - inputs=[verification_session, current_message_index, current_dataset_id, message_queue, verification_records, correction_selector, notes_field, verification_store], - outputs=[ - error_message, - verification_session, - current_message_index, - current_dataset_id, - message_queue, - verification_records, - message_text, - decision_badge, - confidence, - indicators, - progress_display, - correct_count_display, - incorrect_count_display, - accuracy_display, - breakdown_display, - results_summary, - ] - ).then( - lambda: (gr.Row(visible=False), gr.Row(visible=False)), - outputs=[correction_section, submit_correction_row] - ) - - cancel_correction_btn.click( - lambda: "", - outputs=[error_message] - ) - - download_csv_btn.click( - handle_download_csv, - inputs=[verification_session, verification_store], - outputs=[csv_download, error_message] - ) - - # Navigation buttons handlers - def handle_next_message(session: VerificationSession, current_idx: int, dataset_id: str, message_queue: List[str], records: List[dict]): - """Move to next message.""" - if not session or current_idx >= len(message_queue) - 1: - return ( - session, - "āŒ No more messages", - "", "", "", "", - "", - "āœ“ Correct: 0", - "āœ— Incorrect: 0", - "šŸ“Š Accuracy: 0%", - current_idx, - records, - ) - - next_idx = current_idx + 1 - dataset = TestDatasetManager.load_dataset(dataset_id) - next_message = next((m for m in dataset.messages if m.message_id == message_queue[next_idx]), None) - - if next_message: - message_text, decision_badge, confidence, indicators = VerificationUIComponents.render_message_review( - next_message, - next_message.pre_classified_label, - 0.85, - ["Distress indicator 1", "Distress indicator 2"] - ) - - progress = VerificationUIComponents.update_progress_display(next_idx, len(message_queue)) - correct_str, incorrect_str, accuracy_str = VerificationUIComponents.update_statistics_display( - session.correct_count, - session.incorrect_count - ) - - return ( - session, - "", - message_text, - decision_badge, - confidence, - indicators, - progress, - correct_str, - incorrect_str, - accuracy_str, - next_idx, - records, - ) - - return ( - session, - "āŒ Error loading next message", - "", "", "", "", - "", - "āœ“ Correct: 0", - "āœ— Incorrect: 0", - "šŸ“Š Accuracy: 0%", - current_idx, - records, - ) - - def handle_previous_message(session: VerificationSession, current_idx: int, dataset_id: str, message_queue: List[str], records: List[dict]): - """Move to previous message.""" - if not session or current_idx <= 0: - return ( - session, - "āŒ No previous messages", - "", "", "", "", - "", - "āœ“ Correct: 0", - "āœ— Incorrect: 0", - "šŸ“Š Accuracy: 0%", - current_idx, - records, - ) - - prev_idx = current_idx - 1 - dataset = TestDatasetManager.load_dataset(dataset_id) - prev_message = next((m for m in dataset.messages if m.message_id == message_queue[prev_idx]), None) - - if prev_message: - message_text, decision_badge, confidence, indicators = VerificationUIComponents.render_message_review( - prev_message, - prev_message.pre_classified_label, - 0.85, - ["Distress indicator 1", "Distress indicator 2"] - ) - - progress = VerificationUIComponents.update_progress_display(prev_idx, len(message_queue)) - correct_str, incorrect_str, accuracy_str = VerificationUIComponents.update_statistics_display( - session.correct_count, - session.incorrect_count - ) - - return ( - session, - "", - message_text, - decision_badge, - confidence, - indicators, - progress, - correct_str, - incorrect_str, - accuracy_str, - prev_idx, - records, - ) - - return ( - session, - "āŒ Error loading previous message", - "", "", "", "", - "", - "āœ“ Correct: 0", - "āœ— Incorrect: 0", - "šŸ“Š Accuracy: 0%", - current_idx, - records, - ) - - def handle_skip_message(session: VerificationSession, current_idx: int, dataset_id: str, message_queue: List[str], records: List[dict]): - """Skip current message and move to next.""" - return handle_next_message(session, current_idx, dataset_id, message_queue, records) - - # Bind navigation buttons - next_btn.click( - handle_next_message, - inputs=[verification_session, current_message_index, current_dataset_id, message_queue, verification_records], - outputs=[ - verification_session, - error_message, - message_text, - decision_badge, - confidence, - indicators, - progress_display, - correct_count_display, - incorrect_count_display, - accuracy_display, - current_message_index, - verification_records, - ] - ) - - prev_btn.click( - handle_previous_message, - inputs=[verification_session, current_message_index, current_dataset_id, message_queue, verification_records], - outputs=[ - verification_session, - error_message, - message_text, - decision_badge, - confidence, - indicators, - progress_display, - correct_count_display, - incorrect_count_display, - accuracy_display, - current_message_index, - verification_records, - ] - ) - - skip_btn.click( - handle_skip_message, - inputs=[verification_session, current_message_index, current_dataset_id, message_queue, verification_records], - outputs=[ - verification_session, - error_message, - message_text, - decision_badge, - confidence, - indicators, - progress_display, - correct_count_display, - incorrect_count_display, - accuracy_display, - current_message_index, - verification_records, - ] - ) - - # Save results button - DownloadButton triggers download directly - save_results_btn.click( - handle_download_csv, - inputs=[verification_session, verification_store], - outputs=[save_results_btn] - ) - - # Clear session button - def handle_clear_session(): - """Clear current verification session.""" - return ( - None, # verification_session - "āœ… Session cleared", # error_message - "", "", "", "", # message components - "", # progress - "āœ“ Correct: 0", # correct count - "āœ— Incorrect: 0", # incorrect count - "šŸ“Š Accuracy: 0%", # accuracy - 0, # current index - [], # records - ) - - clear_session_btn.click( - handle_clear_session, - outputs=[ - verification_session, - error_message, - message_text, - decision_badge, - confidence, - indicators, - progress_display, - correct_count_display, - incorrect_count_display, - accuracy_display, - current_message_index, - verification_records, - ] - ) - - # Chaplain Feedback Event Handlers - def show_chaplain_feedback_section(): - """Show chaplain feedback section after message review.""" - return gr.Row(visible=True) - - def handle_submit_feedback( - classification_correct: bool, - classification_subcategory: Optional[str], - correct_classification: Optional[str], - question_issues: List[str], - question_comments: str, - referral_issues: List[str], - referral_comments: str, - indicator_issues: str, - indicator_comments: str, - general_notes: str, - session: VerificationSession, - current_idx: int, - message_queue: List[str], - ): - """Handle chaplain feedback submission.""" - try: - if not session or current_idx >= len(message_queue): - return "āŒ Error: Invalid session state", session, current_idx - - # Create tagging record - from src.core.chaplain_models import TaggingRecord - import uuid - - current_message_id = message_queue[current_idx] - - tagging_record = TaggingRecord( - record_id=str(uuid.uuid4()), - message_id=current_message_id, - is_classification_correct=classification_correct, - classification_subcategory=classification_subcategory, - correct_classification=correct_classification, - question_issues=question_issues or [], - question_comments=question_comments, - referral_issues=referral_issues or [], - referral_comments=referral_comments, - indicator_issues=[i.strip() for i in indicator_issues.split(",") if i.strip()], - indicator_comments=indicator_comments, - general_notes=general_notes, - ) - - # Store tagging record in session (would need to extend VerificationSession) - # For now, just confirm submission - success_msg = f"āœ… Feedback submitted for message {current_idx + 1}" - - return success_msg, session, current_idx - - except Exception as e: - return f"āŒ Error: {str(e)}", session, current_idx - - def display_classification_flow(flow_result: Optional[ClassificationFlowResult]): - """Display classification flow result.""" - if not flow_result: - return "", "", "", "" - - badge, explanation, content, indicators = ChaplainFeedbackUIComponents.render_classification_flow(flow_result) - return badge, explanation, content, indicators - - # Bind chaplain feedback events - submit_feedback_btn.click( - handle_submit_feedback, - inputs=[ - is_correct, # is_correct radio - subcategory, # subcategory dropdown - correct_classification, # correct_classification radio - question_issues, # question_issues checkbox - question_comments, # question_comments textbox - referral_issues, # referral_issues checkbox - referral_comments, # referral_comments textbox - indicator_issues, # indicator_issues textbox - indicator_comments, # indicator_comments textbox - general_notes, - verification_session, - current_message_index, - message_queue, - ], - outputs=[error_message, verification_session, current_message_index] - ).then( - lambda: gr.Row(visible=False), - outputs=[chaplain_feedback_section] - ) # Bind events demo.load( @@ -2088,549 +587,110 @@ Use the **Download Summary** button below to access the complete provider summar # Send message send_btn.click( - handle_message, + chat_handlers.handle_message, inputs=[msg, chatbot, session_data], outputs=[chatbot, status_box, session_data, msg, conversation_stats, provider_summary_content, provider_summary_status, provider_summary_display, spiritual_care_message] ) msg.submit( - handle_message, + chat_handlers.handle_message, inputs=[msg, chatbot, session_data], outputs=[chatbot, status_box, session_data, msg, conversation_stats, provider_summary_content, provider_summary_status, provider_summary_display, spiritual_care_message] ) # Clear chat clear_btn.click( - handle_clear, + chat_handlers.handle_clear, inputs=[session_data], outputs=[chatbot, status_box, session_data, provider_summary_content, provider_summary_status, provider_summary_display, spiritual_care_message] ) # Refresh status refresh_btn.click( - get_status, + stats_handlers.get_status, inputs=[session_data], outputs=[status_box, conversation_stats, provider_summary_content, provider_summary_status, provider_summary_display, spiritual_care_message] ) # Example buttons - def send_example_with_stats(example_text: str, history, session: SimplifiedSessionData): - """Send example message and return stats.""" - return handle_message(example_text, history, session) - example_medical.click( - lambda h, s: send_example_with_stats("I am fine", h, s), + lambda h, s: chat_handlers.send_example_with_stats("I am fine", h, s), inputs=[chatbot, session_data], outputs=[chatbot, status_box, session_data, msg, conversation_stats, provider_summary_content, provider_summary_status, provider_summary_display, spiritual_care_message] ) example_wellness.click( - lambda h, s: send_example_with_stats("I'm feeling stressed and overwhelmed lately", h, s), + lambda h, s: chat_handlers.send_example_with_stats("I'm feeling stressed and overwhelmed lately", h, s), inputs=[chatbot, session_data], outputs=[chatbot, status_box, session_data, msg, conversation_stats, provider_summary_content, provider_summary_status, provider_summary_display, spiritual_care_message] ) example_help.click( - lambda h, s: send_example_with_stats("I am currently experiencing an emotional crisis", h, s), + lambda h, s: chat_handlers.send_example_with_stats("I am currently experiencing an emotional crisis", h, s), inputs=[chatbot, session_data], outputs=[chatbot, status_box, session_data, msg, conversation_stats, provider_summary_content, provider_summary_status, provider_summary_display, spiritual_care_message] ) # Conversation logging buttons download_json_btn.click( - download_conversation_json, + stats_handlers.download_conversation_json, inputs=[session_data], outputs=[download_json_btn] ) download_csv_btn.click( - download_conversation_csv, + stats_handlers.download_conversation_csv, inputs=[session_data], outputs=[download_csv_btn] ) # Provider Summary panel handlers - def download_provider_summary(session: SimplifiedSessionData): - """Download provider summary as text file.""" - if session is None: - return None - - last_summary = session.app_instance.get_last_provider_summary() - if not last_summary: - return None - - try: - import tempfile - import os - from datetime import datetime - - # Create temp file with summary - timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") - filename = f"provider_summary_{timestamp}.txt" - filepath = os.path.join(tempfile.gettempdir(), filename) - - summary_text = session.app_instance.provider_summary_generator.format_for_display(last_summary) - with open(filepath, 'w', encoding='utf-8') as f: - f.write(summary_text) - - return filepath - except Exception as e: - print(f"Error downloading provider summary: {e}") - return None - - def clear_provider_summary(): - """Clear provider summary panel.""" - return gr.update(visible=False), "No provider summary available", "", "" - - def regenerate_spiritual_care_message( - session: SimplifiedSessionData, - include_conversation: bool = True, - include_situation: bool = False, - include_indicators: bool = False, - include_profile: bool = False - ): - """Regenerate LLM-based spiritual care message.""" - if session is None: - return "" - - try: - message = session.app_instance.generate_spiritual_care_message( - language="English", - session_id=session.session_id, - include_conversation_context=include_conversation, - include_situation_analysis=include_situation, - include_distress_indicators=include_indicators, - include_patient_profile=include_profile - ) - return message if message else "No provider summary available to generate message from." - except Exception as e: - print(f"Error regenerating spiritual care message: {e}") - return f"Error generating message: {str(e)}" - - def download_spiritual_care_message(session: SimplifiedSessionData): - """Download spiritual care message as text file.""" - if session is None: - return None - - try: - import tempfile - import os - from datetime import datetime - - message = session.app_instance.generate_spiritual_care_message( - language="English", - session_id=session.session_id - ) - if not message: - return None - - # Create temp file with message - timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") - filename = f"spiritual_care_message_{timestamp}.txt" - filepath = os.path.join(tempfile.gettempdir(), filename) - - # Add header - patient_name = session.app_instance.patient_info.get('name', 'Patient') - header = f"SPIRITUAL CARE TEAM MESSAGE\n" - header += f"Generated: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}\n" - header += f"Patient: {patient_name}\n" - header += f"Classification: RED FLAG - Spiritual Distress\n" - header += "=" * 60 + "\n\n" - - with open(filepath, 'w', encoding='utf-8') as f: - f.write(header) - f.write(message) - - return filepath - except Exception as e: - print(f"Error downloading spiritual care message: {e}") - return None - download_summary_btn.click( - download_provider_summary, + stats_handlers.download_provider_summary, inputs=[session_data], outputs=[download_summary_btn] ) clear_summary_btn.click( - clear_provider_summary, + stats_handlers.clear_provider_summary, inputs=[], outputs=[provider_summary_content, provider_summary_status, provider_summary_display, spiritual_care_message] ) # Spiritual care message handlers generate_message_btn.click( - regenerate_spiritual_care_message, + stats_handlers.regenerate_spiritual_care_message, inputs=[session_data, include_conversation_context, include_situation_analysis, include_distress_indicators, include_patient_profile], outputs=[spiritual_care_message] ) download_message_btn.click( - download_spiritual_care_message, + stats_handlers.download_spiritual_care_message, inputs=[session_data], outputs=[download_message_btn] ) - # Download helper (used by embedded Conversation Verification tab) - def _download_latest_verification_json(session: SimplifiedSessionData): - """Return the most recently exported verification session JSON path (if present).""" - # open_verification_window exports into ./verification_sessions - import glob - import os - - export_dir = os.path.join(os.getcwd(), "verification_sessions") - if not os.path.isdir(export_dir): - return None - - candidates = sorted( - glob.glob(os.path.join(export_dir, "verification_session_*.json")), - key=lambda p: os.path.getmtime(p), - reverse=True, - ) - return candidates[0] if candidates else None - - - - # Embedded conversation verification (tab) - def _render_conv_exchange(records: list, index: int): - if not records: - return "", "", "" - index = max(0, min(index, len(records) - 1)) - r = records[index] - # Reuse renderer from conversation_verification_ui to keep style consistent - from src.core.conversation_verification import VerificationRecord - from src.interface.conversation_verification_ui import VerificationInterface - from src.core.conversation_verification import ConversationVerificationManager - - vi = VerificationInterface(ConversationVerificationManager()) - # If we already have dicts, build a lightweight VerificationRecord - if isinstance(r, dict): - rec = VerificationRecord( - exchange_id=r.get("exchange_id") or r.get("record_id", ""), - exchange_number=r.get("exchange_number", 0), - user_message=r.get("user_message", ""), - assistant_response=r.get("assistant_response", ""), - original_classification=r.get("original_classification", ""), - original_confidence=r.get("original_confidence", 0.0), - original_indicators=r.get("original_indicators", []), - original_reasoning=r.get("original_reasoning", ""), - timestamp=r.get("timestamp"), - is_correct=r.get("is_correct"), - correct_classification=r.get("correct_classification"), - correction_reason=r.get("correction_reason"), - verifier_notes=r.get("verifier_notes"), - ) - else: - rec = r - html = vi._render_exchange_review(rec) - # status badge - cur_is_correct = (r.get("is_correct") if isinstance(r, dict) else getattr(r, "is_correct", None)) - if cur_is_correct is True: - badge = "āœ…" - elif cur_is_correct is False: - badge = "āŒ" - else: - badge = "ā³" - pos = f"### {badge} Exchange {index + 1} of {len(records)}" - - # richer stats - reviewed = 0 - correct = 0 - incorrect = 0 - incorrect_with_comment = 0 - corrections = {} # Track classification corrections - - for x in records: - v = (x.get("is_correct") if isinstance(x, dict) else getattr(x, "is_correct", None)) - if v is None: - continue - reviewed += 1 - if v is True: - correct += 1 - else: - incorrect += 1 - note = (x.get("verifier_notes") if isinstance(x, dict) else getattr(x, "verifier_notes", None)) - if note and str(note).strip(): - incorrect_with_comment += 1 - - # Track classification corrections - original_class = (x.get("original_classification") if isinstance(x, dict) else getattr(x, "original_classification", "")) - correct_class = (x.get("correct_classification") if isinstance(x, dict) else getattr(x, "correct_classification", None)) - if original_class and correct_class: - correction_key = f"{original_class}ā†’{correct_class}" - corrections[correction_key] = corrections.get(correction_key, 0) + 1 - - stats_parts = [ - f"
Reviewed: {reviewed}/{len(records)}
", - f"
āœ… Correct: {correct}
", - f"
āŒ Incorrect: {incorrect}
", - f"
šŸ“ Incorrect w/ comment: {incorrect_with_comment}
" - ] - - # Add correction breakdown if any corrections exist - if corrections: - correction_text = ", ".join([f"{k}: {v}" for k, v in corrections.items()]) - stats_parts.append(f"
šŸ”„ Corrections: {correction_text}
") - - stats = ( - "
" - + "".join(stats_parts) + - "
" - ) - return html, pos, stats - - def _comment_ui_state(records: list, idx: int): - """Return (row_update, note_value) based on current record state.""" - if not records: - return gr.update(visible=False), "" - idx = max(0, min(idx, len(records) - 1)) - r = records[idx] - is_incorrect = (r.get("is_correct") is False) if isinstance(r, dict) else (getattr(r, "is_correct", None) is False) - if not is_incorrect: - return gr.update(visible=False), "" - note = (r.get("verifier_notes") or "") if isinstance(r, dict) else (getattr(r, "verifier_notes", "") or "") - return gr.update(visible=True), str(note) - - def _export_conv_records_to_json(meta: dict, records: list): - """Write reviewed conversation verification results to a JSON file and return its path.""" - import json - import os - from datetime import datetime - - export_dir = os.path.join(os.getcwd(), "verification_sessions") - os.makedirs(export_dir, exist_ok=True) - - session_id = (meta or {}).get("session_id") or "conversation_verification" - export_filename = f"conversation_verification_reviewed_{datetime.utcnow().strftime('%Y%m%d_%H%M%S')}_{session_id}.json" - export_path = os.path.join(export_dir, export_filename) - - payload = { - **(meta or {}), - "verification_records": records or [], - } - - with open(export_path, "w", encoding="utf-8") as f: - json.dump(payload, f, ensure_ascii=False, indent=2, default=str) - return export_path - - def _export_conv_records_to_csv(meta: dict, records: list): - """Write reviewed conversation verification results to a CSV file and return its path.""" - import csv - import os - from datetime import datetime - - export_dir = os.path.join(os.getcwd(), "verification_exports") - os.makedirs(export_dir, exist_ok=True) - - session_id = (meta or {}).get("session_id") or "conversation_verification" - export_filename = f"conversation_verification_reviewed_{datetime.utcnow().strftime('%Y%m%d_%H%M%S')}_{session_id}.csv" - export_path = os.path.join(export_dir, export_filename) - - fieldnames = [ - "session_id", - "patient_name", - "patient_phone", - "verifier_name", - "start_time", - "exchange_number", - "exchange_id", - "original_classification", - "original_confidence", - "is_correct", - "correct_classification", - "verifier_notes", - "user_message", - "assistant_response", - "provider_summary", - ] - - with open(export_path, "w", encoding="utf-8", newline="") as f: - w = csv.DictWriter(f, fieldnames=fieldnames) - w.writeheader() - for r in records or []: - # Include provider_summary only for RED cases - provider_summary = "" - if r.get("original_classification", "").upper() == "RED": - provider_summary = r.get("provider_summary") or "" - - row = { - "session_id": (meta or {}).get("session_id"), - "patient_name": (meta or {}).get("patient_name"), - "patient_phone": (meta or {}).get("patient_phone") or "", - "verifier_name": (meta or {}).get("verifier_name"), - "start_time": (meta or {}).get("start_time"), - "exchange_number": r.get("exchange_number"), - "exchange_id": r.get("exchange_id") or r.get("record_id"), - "original_classification": r.get("original_classification"), - "original_confidence": r.get("original_confidence"), - "is_correct": r.get("is_correct"), - "correct_classification": r.get("correct_classification") or "", - "verifier_notes": r.get("verifier_notes") or "", - "user_message": r.get("user_message"), - "assistant_response": r.get("assistant_response"), - "provider_summary": provider_summary, - } - w.writerow(row) - return export_path - - def _generate_conv_verification(session: SimplifiedSessionData): - if session is None or not hasattr(session.app_instance, "conversation_logger"): - return None, [], 0, "āŒ No session/conversation found", "", "" - if not session.app_instance.conversation_logger.entries: - return None, [], 0, "āš ļø No exchanges to verify yet", "", "" - - from src.core.conversation_verification import ConversationVerificationManager - manager = ConversationVerificationManager() - vs = manager.create_verification_session(session.app_instance.conversation_logger, "Medical Professional") - - # Get patient phone from app if available - patient_phone = "" - if hasattr(session.app_instance, 'patient_info'): - patient_phone = session.app_instance.patient_info.get("phone") or "" - - meta = { - "session_id": vs.session_id, - "patient_name": vs.patient_name, - "patient_phone": patient_phone, - "verifier_name": vs.verifier_name, - "start_time": vs.start_time.isoformat() if hasattr(vs, "start_time") else None, - } - - # Get provider summary if available (for RED cases) - provider_summary_text = "" - if hasattr(session.app_instance, 'get_last_provider_summary'): - summary = session.app_instance.get_last_provider_summary() - if summary and hasattr(session.app_instance, 'provider_summary_generator'): - provider_summary_text = session.app_instance.provider_summary_generator.format_for_export(summary) - - records_as_dicts = [ - { - "exchange_id": r.exchange_id, - "exchange_number": r.exchange_number, - "record_id": r.exchange_id, - "timestamp": r.timestamp, - "user_message": r.user_message, - "assistant_response": r.assistant_response, - "original_classification": r.original_classification, - "original_confidence": r.original_confidence, - "original_indicators": r.original_indicators, - "original_reasoning": r.original_reasoning, - "is_correct": r.is_correct, - "correct_classification": r.correct_classification, - "correction_reason": r.correction_reason, - "verifier_notes": r.verifier_notes, - "provider_summary": provider_summary_text if r.original_classification.upper() == "RED" else "", - } - for r in vs.verification_records - ] - html, pos, stats = _render_conv_exchange(records_as_dicts, 0) - return meta, records_as_dicts, 0, f"āœ… Generated session `{vs.session_id}`", html, pos, stats - - def _mark_conv_correct(records: list, idx: int): - if not records: - return records, idx, "", "", "", gr.update(visible=False), "", "" - idx = max(0, min(idx, len(records) - 1)) - if isinstance(records[idx], dict): - records[idx]["is_correct"] = True - # clear comment and correct_classification when marked correct (avoid stale data) - records[idx]["verifier_notes"] = "" - records[idx]["correct_classification"] = None - html, pos, stats = _render_conv_exchange(records, idx) - row_upd, note_val = _comment_ui_state(records, idx) - return records, idx, "āœ… Marked correct", html, pos, stats, row_upd, note_val, "" - - def _mark_conv_incorrect(records: list, idx: int): - if not records: - return records, idx, "", "", "", gr.update(visible=False), "", "" - idx = max(0, min(idx, len(records) - 1)) - if isinstance(records[idx], dict): - records[idx]["is_correct"] = False - html, pos, stats = _render_conv_exchange(records, idx) - row_upd, note_val = _comment_ui_state(records, idx) - # Get existing correct_classification if any - existing_classification = "" - if isinstance(records[idx], dict): - correct_class = records[idx].get("correct_classification") - if correct_class: - # Map back to display text - reverse_map = { - "GREEN": "šŸŸ¢ Should be GREEN - No distress", - "YELLOW": "šŸŸ” Should be YELLOW - Needs clarification", - "RED": "šŸ”“ Should be RED - Spiritual distress" - } - existing_classification = reverse_map.get(correct_class, "") - return records, idx, "āŒ Marked incorrect", html, pos, stats, row_upd, note_val, existing_classification - - def _show_incorrect_comment_ui(records: list, idx: int): - """Mark incorrect and open the comment row, pre-filling any existing note.""" - records, idx, status, html, pos, stats, _row, note, existing_classification = _mark_conv_incorrect(records, idx) - return records, idx, status, html, pos, stats, gr.update(visible=True), note, existing_classification - - def _save_incorrect_comment(records: list, idx: int, note: str, correct_classification: str): - if not records: - return records, idx, "", "", "", "", gr.update(visible=False), "", "" - idx = max(0, min(idx, len(records) - 1)) - if isinstance(records[idx], dict): - records[idx]["verifier_notes"] = (note or "").strip() - # Map display text to classification code - classification_map = { - "šŸŸ¢ Should be GREEN - No distress": "GREEN", - "šŸŸ” Should be YELLOW - Needs clarification": "YELLOW", - "šŸ”“ Should be RED - Spiritual distress": "RED" - } - if correct_classification and correct_classification in classification_map: - records[idx]["correct_classification"] = classification_map[correct_classification] - html, pos, stats = _render_conv_exchange(records, idx) - row_upd, note_val = _comment_ui_state(records, idx) - # keep row visible after save (since still incorrect) - return records, idx, "šŸ’¾ Comment saved", html, pos, stats, row_upd, note_val, "" - - def _download_reviewed_json(meta: dict, records: list): - return _export_conv_records_to_json(meta, records) - - def _download_reviewed_csv(meta: dict, records: list): - return _export_conv_records_to_csv(meta, records) - - def _nav_conv(records: list, idx: int, delta: int): - if not records: - return idx, "", "", "", gr.update(visible=False), "", "" - idx = max(0, min(idx + delta, len(records) - 1)) - html, pos, stats = _render_conv_exchange(records, idx) - row_upd, note_val = _comment_ui_state(records, idx) - # Get existing correct_classification if any - existing_classification = "" - if isinstance(records[idx], dict): - correct_class = records[idx].get("correct_classification") - if correct_class: - # Map back to display text - reverse_map = { - "GREEN": "šŸŸ¢ Should be GREEN - No distress", - "YELLOW": "šŸŸ” Should be YELLOW - Needs clarification", - "RED": "šŸ”“ Should be RED - Spiritual distress" - } - existing_classification = reverse_map.get(correct_class, "") - return idx, html, pos, stats, row_upd, note_val, existing_classification - + # Conversation Verification events generate_conv_verification_btn.click( - _generate_conv_verification, + verification_handlers._generate_conv_verification, inputs=[session_data], outputs=[conv_verify_state, conv_verify_records, conv_verify_index, conv_verify_status, conv_verify_exchange, conv_position, conv_stats] ) conv_verify_download_btn.click( - _download_reviewed_json, + verification_handlers._download_reviewed_json, inputs=[conv_verify_state, conv_verify_records], outputs=[conv_verify_download_btn] ) conv_verify_download_csv_btn.click( - _download_reviewed_csv, + verification_handlers._download_reviewed_csv, inputs=[conv_verify_state, conv_verify_records], outputs=[conv_verify_download_csv_btn] ) conv_correct_btn.click( - _mark_conv_correct, + verification_handlers._mark_conv_correct, inputs=[conv_verify_records, conv_verify_index], outputs=[ conv_verify_records, @@ -2646,7 +706,7 @@ Use the **Download Summary** button below to access the complete provider summar ) conv_incorrect_btn.click( - _show_incorrect_comment_ui, + verification_handlers._show_incorrect_comment_ui, inputs=[conv_verify_records, conv_verify_index], outputs=[ conv_verify_records, @@ -2662,7 +722,7 @@ Use the **Download Summary** button below to access the complete provider summar ) conv_save_comment_btn.click( - _save_incorrect_comment, + verification_handlers._save_incorrect_comment, inputs=[conv_verify_records, conv_verify_index, conv_incorrect_comment, conv_correct_classification], outputs=[ conv_verify_records, @@ -2678,413 +738,84 @@ Use the **Download Summary** button below to access the complete provider summar ) conv_prev_btn.click( - lambda records, idx: _nav_conv(records, idx, -1), + lambda records, idx: verification_handlers._nav_conv(records, idx, -1), inputs=[conv_verify_records, conv_verify_index], outputs=[conv_verify_index, conv_verify_exchange, conv_position, conv_stats, conv_incorrect_comment_row, conv_incorrect_comment, conv_correct_classification] ) conv_next_btn.click( - lambda records, idx: _nav_conv(records, idx, 1), + lambda records, idx: verification_handlers._nav_conv(records, idx, 1), inputs=[conv_verify_records, conv_verify_index], outputs=[conv_verify_index, conv_verify_exchange, conv_position, conv_stats, conv_incorrect_comment_row, conv_incorrect_comment, conv_correct_classification] ) + # Prompt editing events load_prompt_btn.click( - load_prompt, + prompt_handlers.load_prompt, inputs=[prompt_selector, session_data], outputs=[prompt_editor, prompt_info_display, prompt_status] ) apply_prompt_btn.click( - apply_prompt_changes, + prompt_handlers.apply_prompt_changes, inputs=[prompt_selector, prompt_editor, session_data], outputs=[prompt_status, session_data] ) reset_prompt_btn.click( - reset_prompt, + prompt_handlers.reset_prompt, inputs=[prompt_selector, session_data], outputs=[prompt_editor, prompt_info_display, prompt_status, session_data] ) promote_prompt_btn.click( - promote_prompt_to_file, + prompt_handlers.promote_prompt_to_file, inputs=[prompt_selector, session_data], outputs=[prompt_status, session_data] ) validate_prompt_btn.click( - validate_prompt_syntax, + prompt_handlers.validate_prompt_syntax, inputs=[prompt_editor], outputs=[prompt_status] ) # Auto-load prompt when selector changes prompt_selector.change( - load_prompt, + prompt_handlers.load_prompt, inputs=[prompt_selector, session_data], outputs=[prompt_editor, prompt_info_display, prompt_status] ) - # Model selection handlers - def apply_model_settings(spiritual_model: str, soft_spiritual_triage_model: str, triage_evaluate_model: str, medical_model: str, soft_triage_model: str, spiritual_care_message_model: str, session: SimplifiedSessionData): - """Apply custom model settings.""" - if session is None: - session = SimplifiedSessionData() - - # Store model settings in session - if not hasattr(session, 'custom_models'): - session.custom_models = {} - - session.custom_models = { - 'SpiritualDistressAnalyzer': spiritual_model, - 'SoftSpiritualTriage': soft_spiritual_triage_model, - 'TriageResponseEvaluator': triage_evaluate_model, - 'MedicalAssistant': medical_model, - 'SoftMedicalTriage': soft_triage_model, - 'SpiritualCareMessage': spiritual_care_message_model - } - - status = f"""
-

āœ… Model Settings Applied

- -

šŸ” Spiritual Monitor: {spiritual_model}

-

šŸŸ” Soft Spiritual Triage: {soft_spiritual_triage_model}

-

šŸ“Š Triage Response Evaluator: {triage_evaluate_model}

-

šŸ„ Medical Assistant: {medical_model}

-

šŸ©ŗ Soft Medical Triage: {soft_triage_model}

-

šŸ’¬ Spiritual Care Message: {spiritual_care_message_model}

- -

-āš ļø Note: Model changes apply to this session only. -

-
""" - - return status, session - - def reset_model_settings(session: SimplifiedSessionData): - """Reset models to defaults.""" - if session is None: - session = SimplifiedSessionData() - - # Clear custom models - if hasattr(session, 'custom_models'): - session.custom_models = {} - - status = """
-

šŸ”„ Models Reset to Defaults

- -

šŸ” Spiritual Monitor: gemini-2.5-flash

-

šŸŸ” Soft Spiritual Triage: claude-sonnet-4-5-20250929

-

šŸ“Š Triage Response Evaluator: gemini-2.5-flash

-

šŸ„ Medical Assistant: claude-sonnet-4-5-20250929

-

šŸ©ŗ Soft Medical Triage: claude-sonnet-4-5-20250929

-

šŸ’¬ Spiritual Care Message: claude-sonnet-4-5-20250929

- -

Default models are now active.

-
""" - - return status, session - # Bind model selection events apply_models_btn.click( - apply_model_settings, + model_handlers.apply_model_settings, inputs=[spiritual_model, soft_spiritual_triage_model, triage_evaluate_model, medical_model, soft_triage_model, spiritual_care_message_model, session_data], outputs=[model_status, session_data] ) reset_models_btn.click( - reset_model_settings, + model_handlers.reset_model_settings, inputs=[session_data], outputs=[model_status, session_data] ) - # Patient profile handlers - def load_profile(profile_name: str, session: SimplifiedSessionData): - """Load predefined patient profile and apply it to the session.""" - profiles = { - "šŸ‘¤ Default Profile (Serhii)": { - "name": "Serhii", - "phone": "(555) 123-4567", - "age": 52, - "conditions": "Atrial fibrillation, Deep vein thrombosis, Obesity, Hypertension", - "goal": "Weight reduction and cardiovascular fitness improvement", - "exercise": "Swimming, Walking, Light cardio", - "limitations": "Anticoagulation therapy, Post-thrombotic recovery" - }, - "šŸŸ¢ GREEN - Healthy Coping": { - "name": "James", - "phone": "(555) 234-5678", - "age": 40, - "conditions": "No chronic conditions, Excellent health", - "goal": "Maintain fitness and wellness", - "exercise": "Running, Gym, Sports", - "limitations": "None" - }, - "šŸŸ” YELLOW - Mild Distress": { - "name": "Lisa", - "phone": "(555) 345-6789", - "age": 45, - "conditions": "Hypertension, Mild anxiety, Sleep issues", - "goal": "Manage stress and improve sleep quality", - "exercise": "Yoga, Walking, Meditation", - "limitations": "Stress-related fatigue, Occasional insomnia" - }, - "šŸŸ” YELLOW - Grief & Loss": { - "name": "Michael", - "phone": "(555) 456-7890", - "age": 58, - "conditions": "Recent loss of spouse, Mild depression", - "goal": "Process grief and rebuild routine", - "exercise": "Gentle walking, Support groups", - "limitations": "Low motivation, Emotional exhaustion" - }, - "šŸŸ” YELLOW - Existential Questions": { - "name": "Patricia", - "phone": "(555) 567-8901", - "age": 62, - "conditions": "Chronic pain, Questioning life purpose", - "goal": "Find meaning and manage chronic pain", - "exercise": "Tai Chi, Meditation, Gentle stretching", - "limitations": "Chronic pain, Existential concerns" - }, - "šŸŸ” YELLOW - Spiritual Disconnection": { - "name": "David", - "phone": "(555) 678-9012", - "age": 55, - "conditions": "Loss of faith, Isolation from community", - "goal": "Reconnect with spiritual community", - "exercise": "Walking, Community activities", - "limitations": "Spiritual disconnection, Social isolation" - }, - "šŸ”“ RED - Crisis (Suicidal Risk)": { - "name": "Thomas", - "phone": "(555) 789-0123", - "age": 35, - "conditions": "Severe depression, Suicidal ideation", - "goal": "Immediate crisis intervention and support", - "exercise": "None - Crisis support priority", - "limitations": "CRISIS - Suicidal thoughts, Immediate referral needed" - }, - "šŸ”“ RED - Severe Hopelessness": { - "name": "Jennifer", - "phone": "(555) 890-1234", - "age": 48, - "conditions": "Major depression, Complete hopelessness", - "goal": "Crisis stabilization and professional support", - "exercise": "None - Medical intervention priority", - "limitations": "CRISIS - Severe hopelessness, Unable to function" - }, - "šŸ”“ RED - Spiritual Crisis": { - "name": "Christopher", - "phone": "(555) 901-2345", - "age": 52, - "conditions": "Moral injury, Spiritual crisis, Anger at God", - "goal": "Spiritual crisis intervention and healing", - "exercise": "None - Spiritual support priority", - "limitations": "CRISIS - Spiritual crisis, Rage, Existential despair" - }, - "šŸ«€ Cardiac Patient (Rehabilitation)": { - "name": "John", - "phone": "(555) 012-3456", - "age": 65, - "conditions": "Coronary artery disease, Hypertension, Hyperlipidemia", - "goal": "Cardiac rehabilitation and risk factor management", - "exercise": "Supervised walking, Cardiac rehab program", - "limitations": "Recent MI, Limited exertion tolerance" - }, - "šŸ©ø Diabetic Patient (Management)": { - "name": "Maria", - "phone": "(555) 111-2222", - "age": 58, - "conditions": "Type 2 Diabetes, Obesity, Hypertension", - "goal": "Blood sugar control and weight management", - "exercise": "Moderate walking, Resistance training", - "limitations": "Neuropathy, Retinopathy risk" - }, - "šŸ„ Post-Surgery (Recovery)": { - "name": "Alex", - "phone": "(555) 222-3333", - "age": 45, - "conditions": "Post-surgical recovery, Pain management", - "goal": "Safe return to normal activities", - "exercise": "Gentle mobility, Gradual progression", - "limitations": "Surgical site healing, Limited ROM" - }, - "šŸ§  Mental Health (Anxiety/Depression)": { - "name": "Emma", - "phone": "(555) 333-4444", - "age": 35, - "conditions": "Depression, Anxiety, Sedentary lifestyle", - "goal": "Mood improvement through activity", - "exercise": "Yoga, Walking, Group activities", - "limitations": "Low motivation, Energy fluctuations" - }, - "šŸ‘“ Elderly Patient (Chronic Care)": { - "name": "Robert", - "phone": "(555) 444-5555", - "age": 78, - "conditions": "Arthritis, Osteoporosis, Hypertension", - "goal": "Maintain independence and mobility", - "exercise": "Tai Chi, Water aerobics, Balance training", - "limitations": "Fall risk, Joint pain, Medication interactions" - }, - "šŸƒ Athletic Patient (Injury/Training)": { - "name": "Sarah", - "phone": "(555) 555-6666", - "age": 32, - "conditions": "Mild hypertension, Overtraining syndrome", - "goal": "Optimize performance and prevent injury", - "exercise": "Running, Strength training, Cross-training", - "limitations": "Overuse injuries, Recovery needs" - } - } - - profile = profiles.get(profile_name, profiles["šŸ‘¤ Default Profile (Serhii)"]) - - # Automatically apply the profile to the session - if session and hasattr(session.app_instance, 'set_patient_info'): - session.app_instance.set_patient_info( - name=profile['name'], - phone=profile.get('phone', '') - ) - - # Update clinical_background for medical context - from src.core.core_classes import ClinicalBackground - session.app_instance.clinical_background = ClinicalBackground( - patient_name=profile['name'], - age=profile['age'], - conditions=profile['conditions'].split(',') if isinstance(profile['conditions'], str) else profile['conditions'], - primary_goal=profile['goal'], - exercise_preferences=profile['exercise'].split(',') if isinstance(profile['exercise'], str) else profile['exercise'], - exercise_limitations=profile['limitations'].split(',') if isinstance(profile['limitations'], str) else profile['limitations'] - ) - - # Update conversation logger patient name - if hasattr(session.app_instance, 'conversation_logger'): - session.app_instance.conversation_logger.patient_name = profile['name'] - - print(f"DEBUG: Auto-applied profile - Name: {profile['name']}, Phone: {profile.get('phone', '')}") - print(f"DEBUG: Clinical background updated - Age: {profile['age']}, Conditions: {profile['conditions']}") - - status = f"""
-

āœ… Profile Loaded & Applied

-

Patient: {profile['name']}, {profile['age']} years old

-

Phone: {profile.get('phone', 'Not provided')}

-

Profile: {profile_name}

-

Status: Profile has been automatically applied to this session

-

āœ“ Ready to use in conversations and reports

-
""" - - return ( - profile['name'], - profile.get('phone', ''), - profile['age'], - profile['conditions'], - profile['goal'], - profile['exercise'], - profile['limitations'], - status - ) - - def save_profile(name: str, phone: str, age: float, conditions: str, goal: str, exercise: str, limitations: str, session: SimplifiedSessionData): - """Save current profile settings and update app patient info.""" - if not name.strip(): - return """
-

āŒ Error

-

Patient name cannot be empty

-
""" - - # Update session's app instance patient info for provider summaries - if session and hasattr(session.app_instance, 'set_patient_info'): - session.app_instance.set_patient_info(name=name.strip(), phone=phone.strip() if phone else None) - - # Also update clinical_background for medical context - from src.core.core_classes import ClinicalBackground - session.app_instance.clinical_background = ClinicalBackground( - patient_name=name.strip(), - age=int(age) if age else None, - conditions=conditions.strip().split(',') if conditions.strip() else [], - primary_goal=goal.strip(), - exercise_preferences=exercise.strip().split(',') if exercise.strip() else [], - exercise_limitations=limitations.strip().split(',') if limitations.strip() else [] - ) - - # Update conversation logger patient name - if hasattr(session.app_instance, 'conversation_logger'): - session.app_instance.conversation_logger.patient_name = name.strip() - - print(f"DEBUG: Updated patient info - Name: {name.strip()}, Phone: {phone.strip() if phone else None}") - print(f"DEBUG: Updated clinical_background - Age: {int(age) if age else None}, Conditions: {conditions}") - - status = f"""
-

šŸ’¾ Profile Saved

-

Patient: {name}, {int(age)} years old

-

Phone: {phone if phone else 'Not provided'}

-

Conditions: {conditions}

-

Primary Goal: {goal}

-

Profile settings have been updated for this session.

-
""" - - return status - - def reset_profile(session: SimplifiedSessionData): - """Reset profile to default.""" - # Reset session's app instance patient info - if session and hasattr(session.app_instance, 'set_patient_info'): - session.app_instance.set_patient_info(name="Serhii", phone="(555) 123-4567") - - # Also reset clinical_background - from src.core.core_classes import ClinicalBackground - session.app_instance.clinical_background = ClinicalBackground( - patient_name="Serhii", - age=52, - conditions=["Atrial fibrillation", "Deep vein thrombosis", "Obesity", "Hypertension"], - primary_goal="Weight reduction and cardiovascular fitness improvement", - exercise_preferences=["Swimming", "Walking", "Light cardio"], - exercise_limitations=["Anticoagulation therapy", "Post-thrombotic recovery"] - ) - - # Update conversation logger patient name - if hasattr(session.app_instance, 'conversation_logger'): - session.app_instance.conversation_logger.patient_name = "Serhii" - - print(f"DEBUG: Reset patient info to default") - - status = """
-

šŸ”„ Profile Reset

-

Patient: Serhii, 52 years old

-

Phone: (555) 123-4567

-

Default profile has been restored.

-
""" - - return ( - "Serhii", - "(555) 123-4567", - 52, - "Atrial fibrillation, Deep vein thrombosis, Obesity, Hypertension", - "Weight reduction and cardiovascular fitness improvement", - "Swimming, Walking, Light cardio", - "Anticoagulation therapy, Post-thrombotic recovery", - status - ) - # Bind profile events load_profile_btn.click( - load_profile, + profile_handlers.load_profile, inputs=[profile_selector, session_data], outputs=[patient_name, patient_phone, patient_age, conditions, primary_goal, exercise_prefs, exercise_limits, profile_status] ) save_profile_btn.click( - save_profile, + profile_handlers.save_profile, inputs=[patient_name, patient_phone, patient_age, conditions, primary_goal, exercise_prefs, exercise_limits, session_data], outputs=[profile_save_status] ) reset_profile_btn.click( - reset_profile, + profile_handlers.reset_profile, inputs=[session_data], outputs=[patient_name, patient_phone, patient_age, conditions, primary_goal, exercise_prefs, exercise_limits, profile_save_status] )