Upload gradio_app.py

gradio_app.py

@@ -0,0 +1,542 @@
+# app.py - Gradio version for Hugging Face Spaces
+import gradio as gr
+import requests
+import json
+import time
+from typing import Dict, Any, Tuple, List
+
+# Configuration - Can switch between local and AWS
+LOCAL_API = "http://localhost:8080"
+AWS_API = "https://rj92jktas7.us-east-2.awsapprunner.com"
+
+# Global variables to maintain state
+current_session = {
+    'session_id': None,
+    'validation_started': False,
+    'all_questions': [],
+    'current_question_id': None,
+    'validation_complete': False,
+    'api_base_url': AWS_API
+}
+
+# Sample records for testing
+SAMPLE_RECORDS = {
+    "Medical Record 1 - Multiple Issues": {
+        "UUID": "sample-001",
+        "patient_info": {
+            "name": "John Doe",
+            "age": 28
+        },
+        "clinical_risk": {
+            "thyroid_disease": "how are you?",
+            "anemia": None,
+            "preeclampsia_current": "not sure",
+            "geriatric_mother": "maybe",
+            "malaria_chronic_history": ".",
+            "preeclampsia_prior": "[insert explanation here]"
+        },
+        "rmc1": {
+            "rmc1_respect": True,
+            "rmc1_explanation": "[insert explanation here]",
+            "rmc1_facility": "Test Hospital"
+        },
+        "immunization_birth": {
+            "birth_bcg": False,
+            "birth_oral_polio": True,
+            "birth_no_vax_reason": "not sure"
+        }
+    },
+    "Medical Record 2 - Few Issues": {
+        "UUID": "sample-002",
+        "patient_info": {
+            "name": "Jane Smith",
+            "age": 35
+        },
+        "clinical_risk": {
+            "thyroid_disease": True,
+            "anemia": "maybe",
+            "preeclampsia_current": False,
+            "malaria_chronic_history": "[insert explanation here]"
+        },
+        "rmc2": {
+            "rmc2_respect": False,
+            "rmc2_explanation": ".",
+            "rmc2_facility": "Another Hospital"
+        }
+    }
+}
+
+def check_api_health():
+    """Check if the API is running and get feature info"""
+    try:
+        response = requests.get(f"{current_session['api_base_url']}/health", timeout=10)
+        if response.status_code == 200:
+            health_data = response.json()
+            return True, health_data
+        else:
+            return False, f"Status code: {response.status_code}"
+    except Exception as e:
+        return False, f"Error: {str(e)}"
+
+def start_validation_session(record: Dict[str, Any]):
+    """Start a validation session with the API"""
+    try:
+        response = requests.post(f"{current_session['api_base_url']}/start-validation", json=record, timeout=30)
+        if response.status_code == 200:
+            return response.json()
+        else:
+            return None, f"Failed to start validation: {response.text}"
+    except Exception as e:
+        return None, f"Error connecting to API: {str(e)}"
+
+def get_question_by_id(question_id: str):
+    """Get a specific question by its unique ID"""
+    try:
+        response = requests.get(f"{current_session['api_base_url']}/get-question/{question_id}", timeout=15)
+        if response.status_code == 200:
+            return response.json()
+        else:
+            return None
+    except Exception as e:
+        return None
+
+def list_all_questions(session_id: str):
+    """Get list of all questions for a session"""
+    try:
+        response = requests.get(f"{current_session['api_base_url']}/list-questions/{session_id}", timeout=15)
+        if response.status_code == 200:
+            return response.json()
+        else:
+            return None
+    except Exception as e:
+        return None
+
+def submit_response_to_question(question_id: str, user_response: str):
+    """Submit a response to a specific question"""
+    try:
+        response = requests.post(f"{current_session['api_base_url']}/submit-response/{question_id}", 
+                               json={"response": user_response}, timeout=30)
+        if response.status_code == 200:
+            return response.json()
+        else:
+            return None
+    except Exception as e:
+        return None
+
+def get_final_results(session_id: str):
+    """Get the final validation results"""
+    try:
+        response = requests.get(f"{current_session['api_base_url']}/get-results/{session_id}", timeout=15)
+        if response.status_code == 200:
+            return response.json()
+        else:
+            return None
+    except Exception as e:
+        return None
+
+# Gradio Interface Functions
+
+def update_api_endpoint(api_choice, custom_url=""):
+    """Update the API endpoint based on user selection"""
+    if api_choice == "AWS Deployment":
+        current_session['api_base_url'] = AWS_API
+        status = f"✅ Using AWS App Runner: {AWS_API}"
+    elif api_choice == "Local Development":
+        current_session['api_base_url'] = LOCAL_API
+        status = f"ℹ️ Using Local API: {LOCAL_API}"
+    else:  # Custom URL
+        current_session['api_base_url'] = custom_url if custom_url else AWS_API
+        status = f"🔧 Using Custom URL: {current_session['api_base_url']}"
+    
+    # Check API health
+    is_healthy, health_info = check_api_health()
+    if is_healthy:
+        if isinstance(health_info, dict):
+            status += f"\n\n📊 API Status:\n"
+            status += f"• Version: {health_info.get('version', 'N/A')}\n"
+            status += f"• Active Sessions: {health_info.get('active_sessions', 0)}\n"
+            status += f"• Total Questions: {health_info.get('total_questions', 0)}\n"
+            
+            features = health_info.get('features', [])
+            if features:
+                status += f"• Features: {', '.join(features)}"
+    else:
+        status += f"\n\n❌ API Error: {health_info}"
+    
+    return status
+
+def load_sample_record(sample_choice):
+    """Load a sample record based on user selection"""
+    if sample_choice and sample_choice in SAMPLE_RECORDS:
+        record_json = json.dumps(SAMPLE_RECORDS[sample_choice], indent=2)
+        return record_json, f"✅ Loaded: {sample_choice}"
+    return "", "Please select a sample record"
+
+def parse_json_record(json_text):
+    """Parse and validate JSON record"""
+    if not json_text.strip():
+        return "❌ Please provide a JSON record", ""
+    
+    try:
+        record = json.loads(json_text)
+        preview = json.dumps(record, indent=2)
+        return f"✅ Valid JSON record parsed successfully!", preview
+    except json.JSONDecodeError as e:
+        return f"❌ Invalid JSON: {str(e)}", ""
+
+def start_validation(json_text):
+    """Start the validation process"""
+    if not json_text.strip():
+        return "❌ Please provide a JSON record first", "", gr.update(visible=False)
+    
+    try:
+        record = json.loads(json_text)
+    except json.JSONDecodeError as e:
+        return f"❌ Invalid JSON: {str(e)}", "", gr.update(visible=False)
+    
+    # Start validation session
+    result = start_validation_session(record)
+    if isinstance(result, tuple):  # Error case
+        return f"❌ {result[1]}", "", gr.update(visible=False)
+    
+    if result:
+        current_session['session_id'] = result["session_id"]
+        current_session['all_questions'] = result.get("questions", [])
+        current_session['validation_started'] = True
+        
+        if result["total_questions"] == 0:
+            current_session['validation_complete'] = True
+            return "✅ No missing fields found! Record is already complete.", "", gr.update(visible=False)
+        else:
+            status = f"✅ Created {result['total_questions']} individual questions with unique tracking IDs\n"
+            
+            # Show tracking info
+            tracking_info = result.get("tracking", {})
+            if tracking_info.get("question_tracking_enabled"):
+                status += "🔍 Each question now has a unique ID for precise tracking"
+            
+            # Get first question
+            question_status = list_all_questions(current_session['session_id'])
+            if question_status:
+                questions_list = question_status["questions"]
+                if questions_list:
+                    first_question = questions_list[0]
+                    current_session['current_question_id'] = first_question['question_id']
+                    
+                    # Get question details
+                    question_details = get_question_by_id(first_question['question_id'])
+                    if question_details:
+                        question_display = f"**Field:** {question_details['field_path']}\n"
+                        question_display += f"**Question ID:** {question_details['question_id']}\n\n"
+                        question_display += f"### {question_details['question']}"
+                        
+                        progress_info = f"Progress: 0/{result['total_questions']} (0.0%)"
+                        
+                        return status, f"{progress_info}\n\n{question_display}", gr.update(visible=True)
+            
+            return status, "Loading questions...", gr.update(visible=True)
+    else:
+        return "❌ Failed to start validation session", "", gr.update(visible=False)
+
+def submit_answer(answer_text):
+    """Submit an answer to the current question"""
+    if not answer_text.strip():
+        return "❌ Please provide an answer", "", gr.update(visible=True)
+    
+    if not current_session.get('current_question_id'):
+        return "❌ No active question", "", gr.update(visible=True)
+    
+    # Submit response
+    response = submit_response_to_question(current_session['current_question_id'], answer_text)
+    
+    if not response:
+        return "❌ Failed to submit response", "", gr.update(visible=True)
+    
+    # Process response
+    result_text = ""
+    
+    if response.get("status") == "needs_clarification":
+        result_text = f"⚠️ Your answer needs clarification\n\n"
+        result_text += f"**Follow-up question:** {response['clarification_question']}\n\n"
+        result_text += "Please provide a clearer, more specific answer."
+        
+        # Update current question to the clarification
+        current_session['current_question_id'] = response.get('clarification_question_id', current_session['current_question_id'])
+        
+        return result_text, "", gr.update(visible=True)
+    
+    elif response.get("status") in ["completed", "low_confidence"]:
+        validation = response["validation"]
+        confidence = validation["confidence"]
+        
+        if response.get("status") == "completed":
+            result_text = f"✅ Answer accepted! (Confidence: {confidence:.2f})\n\n"
+        else:
+            result_text = f"⚠️ Answer accepted with low confidence: {confidence:.2f}\n\n"
+        
+        # Show tracking details
+        tracking_info = response.get("tracking", {})
+        if tracking_info:
+            result_text += f"🔍 **Tracking Details:**\n"
+            result_text += f"• Question ID: {tracking_info.get('question_unique_id', 'N/A')}\n"
+            result_text += f"• Response Time: {tracking_info.get('response_timestamp', 'N/A')}\n"
+            result_text += f"• Attempt Count: {tracking_info.get('attempt_count', 'N/A')}\n"
+            result_text += f"• Validation Confidence: {tracking_info.get('validation_confidence', 0):.3f}\n\n"
+        
+        # Check if all completed
+        progress = response.get("progress", {})
+        if progress.get("all_completed"):
+            current_session['validation_complete'] = True
+            result_text += "🎉 **All questions completed!**"
+            return result_text, "All questions have been answered. You can now view the final results.", gr.update(visible=False)
+        else:
+            # Get next question
+            question_status = list_all_questions(current_session['session_id'])
+            if question_status:
+                questions_list = question_status["questions"]
+                completed_count = question_status["completed_questions"]
+                total_count = question_status["total_questions"]
+                
+                # Find next pending question
+                next_question = None
+                for q in questions_list:
+                    if q['status'] in ['pending', 'needs_clarification']:
+                        next_question = q
+                        break
+                
+                if next_question:
+                    current_session['current_question_id'] = next_question['question_id']
+                    
+                    # Get question details
+                    question_details = get_question_by_id(next_question['question_id'])
+                    if question_details:
+                        question_display = f"**Field:** {question_details['field_path']}\n"
+                        question_display += f"**Question ID:** {question_details['question_id']}\n\n"
+                        question_display += f"### {question_details['question']}"
+                        
+                        progress_info = f"Progress: {completed_count}/{total_count} ({completed_count/total_count*100:.1f}%)"
+                        
+                        return result_text, f"{progress_info}\n\n{question_display}", gr.update(visible=True)
+            
+            return result_text, "Loading next question...", gr.update(visible=True)
+    
+    return "❓ Unexpected response", "", gr.update(visible=True)
+
+def get_results():
+    """Get the final validation results"""
+    if not current_session.get('session_id'):
+        return "❌ No active session"
+    
+    results = get_final_results(current_session['session_id'])
+    
+    if not results:
+        return "❌ Failed to get results"
+    
+    # Format results
+    result_text = "# 🎉 Validation Complete!\n\n"
+    
+    # Summary metrics
+    result_text += "## 📊 Summary\n"
+    result_text += f"• **Fields Processed:** {results['summary']['total_fields_processed']}\n"
+    result_text += f"• **Success Rate:** {results['summary']['success_rate']:.1f}%\n"
+    result_text += f"• **Average Confidence:** {results['summary']['average_confidence']:.2f}\n"
+    result_text += f"• **Average Attempts per Question:** {results['summary']['average_attempts_per_question']:.1f}\n\n"
+    
+    # Tracking metrics
+    result_text += "## 🔍 Question Tracking\n"
+    result_text += f"• **Total Questions:** {results['summary']['total_questions']}\n"
+    result_text += f"• **Questions with Clarification:** {results['summary']['questions_with_clarification']}\n"
+    result_text += f"• **Completed Questions:** {results['summary']['completed_questions']}\n\n"
+    
+    # Updated record
+    result_text += "## 📋 Updated Record\n"
+    result_text += "```json\n"
+    result_text += json.dumps(results["updated_record"], indent=2)
+    result_text += "\n```\n\n"
+    
+    # Download files
+    updated_record_json = json.dumps(results["updated_record"], indent=2)
+    tracking_data = {
+        "session_info": results.get("session_info", {}),
+        "question_details": results.get("question_details", {}),
+        "summary": results.get("summary", {})
+    }
+    tracking_json = json.dumps(tracking_data, indent=2)
+    full_results_json = json.dumps(results, indent=2)
+    
+    return result_text, updated_record_json, tracking_json, full_results_json
+
+def reset_session():
+    """Reset the current session"""
+    global current_session
+    session_id = current_session.get('session_id', 'new')
+    current_session = {
+        'session_id': None,
+        'validation_started': False,
+        'all_questions': [],
+        'current_question_id': None,
+        'validation_complete': False,
+        'api_base_url': current_session['api_base_url']  # Keep the API setting
+    }
+    return f"🔄 Session reset. Ready for new validation.", "", "", gr.update(visible=False)
+
+# Create Gradio Interface
+def create_interface():
+    with gr.Blocks(title="Medical Record Validation", theme=gr.themes.Soft()) as demo:
+        gr.Markdown("# 🏥 Medical Record Data Validation")
+        gr.Markdown("**Enhanced with Question-Level Tracking**")
+        
+        with gr.Tab("🚀 Setup & Configuration"):
+            gr.Markdown("## API Configuration")
+            
+            api_choice = gr.Radio(
+                choices=["AWS Deployment", "Local Development", "Custom URL"],
+                value="AWS Deployment",
+                label="Choose API endpoint"
+            )
+            
+            custom_url = gr.Textbox(
+                label="Custom API URL",
+                placeholder="https://your-api-url.com",
+                visible=False
+            )
+            
+            api_status = gr.Markdown("🔄 Checking API status...")
+            
+            # Update API status when choice changes
+            def update_custom_url_visibility(choice):
+                return gr.update(visible=(choice == "Custom URL"))
+            
+            api_choice.change(
+                update_custom_url_visibility,
+                inputs=[api_choice],
+                outputs=[custom_url]
+            )
+            
+            api_choice.change(
+                update_api_endpoint,
+                inputs=[api_choice, custom_url],
+                outputs=[api_status]
+            )
+            
+            custom_url.change(
+                update_api_endpoint,
+                inputs=[api_choice, custom_url],
+                outputs=[api_status]
+            )
+        
+        with gr.Tab("📋 Record Input"):
+            gr.Markdown("## Step 1: Provide Medical Record")
+            
+            with gr.Row():
+                with gr.Column():
+                    sample_choice = gr.Dropdown(
+                        choices=list(SAMPLE_RECORDS.keys()),
+                        label="Choose from sample records",
+                        value=None
+                    )
+                    
+                    load_sample_btn = gr.Button("Load Sample Record", variant="secondary")
+                    
+                with gr.Column():
+                    upload_file = gr.File(
+                        label="Upload JSON file",
+                        file_types=[".json"]
+                    )
+            
+            json_input = gr.Textbox(
+                label="Medical Record JSON",
+                placeholder="Paste your JSON record here or use the options above...",
+                lines=10
+            )
+            
+            parse_status = gr.Markdown("")
+            json_preview = gr.Code(label="Record Preview", language="json", visible=False)
+            
+            validate_btn = gr.Button("🚀 Start Validation", variant="primary", size="lg")
+            validation_status = gr.Markdown("")
+            
+            # Event handlers
+            load_sample_btn.click(
+                load_sample_record,
+                inputs=[sample_choice],
+                outputs=[json_input, parse_status]
+            )
+            
+            json_input.change(
+                parse_json_record,
+                inputs=[json_input],
+                outputs=[parse_status, json_preview]
+            )
+        
+        with gr.Tab("❓ Answer Questions"):
+            gr.Markdown("## Step 2: Answer Validation Questions")
+            
+            question_display = gr.Markdown("Start validation to see questions here...")
+            
+            with gr.Row():
+                answer_input = gr.Textbox(
+                    label="Your Answer",
+                    placeholder="Enter your response here...",
+                    scale=3
+                )
+                submit_btn = gr.Button("Submit Answer", variant="primary", scale=1)
+            
+            answer_result = gr.Markdown("")
+            
+            question_section = gr.Group(visible=False)
+            
+            with question_section:
+                submit_btn.click(
+                    submit_answer,
+                    inputs=[answer_input],
+                    outputs=[answer_result, question_display, question_section]
+                )
+        
+        with gr.Tab("📊 Results"):
+            gr.Markdown("## Step 3: Validation Results")
+            
+            get_results_btn = gr.Button("📥 Get Final Results", variant="primary")
+            results_display = gr.Markdown("Complete validation to see results here...")
+            
+            with gr.Row():
+                download_record = gr.File(label="Updated Record", visible=False)
+                download_tracking = gr.File(label="Tracking Details", visible=False)
+                download_full = gr.File(label="Complete Results", visible=False)
+            
+            reset_btn = gr.Button("🔄 Start New Validation", variant="secondary")
+            reset_status = gr.Markdown("")
+        
+        # Main event handlers
+        validate_btn.click(
+            start_validation,
+            inputs=[json_input],
+            outputs=[validation_status, question_display, question_section]
+        )
+        
+        get_results_btn.click(
+            get_results,
+            outputs=[results_display, download_record, download_tracking, download_full]
+        )
+        
+        reset_btn.click(
+            reset_session,
+            outputs=[reset_status, validation_status, question_display, question_section]
+        )
+        
+        # Initialize API status
+        demo.load(
+            lambda: update_api_endpoint("AWS Deployment"),
+            outputs=[api_status]
+        )
+    
+    return demo
+
+# Launch the app
+if __name__ == "__main__":
+    demo = create_interface()
+    demo.launch(
+        server_name="0.0.0.0",
+        server_port=7860,
+        share=False
+    )