gradio_app.py

#!/usr/bin/env python3
# gradio_app.py - Gradio interface for CareLoop

import os
import json
import gradio as gr
from datetime import datetime, timedelta
from dataclasses import asdict
from typing import Dict, List, Any, Optional
import random
import pandas as pd
import numpy as np
import matplotlib.pyplot as plt

from openai import OpenAI
from langchain_core.messages import HumanMessage, AIMessage

# Import CareLoop components
from careloop_main import (
    MockDataGenerator, 
    CareState, 
    HealthMonitorAgent,
    MedicationAgent,
    FamilyCommunicationAgent,
    ActionPlannerAgent,
    AlertLevel
)

# Initialize the OpenAI client for Nebius API
client = OpenAI(
    base_url="https://api.studio.nebius.com/v1/",
    api_key=os.environ.get("NEBIUS_API_KEY", "demo-key-for-hackathon")
)

class NebiusLLM:
    """LLM wrapper for Nebius API"""
    
    def __init__(self, model_name="meta-llama/Meta-Llama-3.1-70B-Instruct"):
        self.model_name = model_name
        self.client = client
        
    def generate(self, prompt: str) -> str:
        """Generate text using Nebius API"""
        try:
            response = self.client.chat.completions.create(
                model=self.model_name,
                max_tokens=512,
                temperature=0.6,
                top_p=0.9,
                extra_body={
                    "top_k": 50
                },
                messages=[{"role": "user", "content": prompt}]
            )
            return response.choices[0].message.content
        except Exception as e:
            print(f"Error calling Nebius API: {e}")
            # Fall back to mock responses in case of API error
            return self._generate_mock_response(prompt)
    
    def _generate_mock_response(self, prompt: str) -> str:
        """Generate a mock response for demo purposes"""
        if "health" in prompt.lower():
            return "The patient's health metrics show stable vital signs with glucose levels within acceptable range."
        elif "medication" in prompt.lower():
            return "Medication compliance has been good this week with only one missed dose of evening medication."
        elif "summary" in prompt.lower():
            return "Overall, the patient is doing well with stable health metrics and good medication compliance."
        else:
            return "The care system is monitoring the patient's condition and no significant issues have been detected."

class CareLoopSystem:
    """CareLoop system with Nebius LLM integration"""
    
    def __init__(self):
        self.mock_data = MockDataGenerator()
        self.llm = NebiusLLM()
        
        # Initialize specialized agents
        self.health_monitor = HealthMonitorAgent(self.llm)
        self.medication_agent = MedicationAgent(self.llm)
        self.family_communicator = FamilyCommunicationAgent(self.llm)
        self.action_planner = ActionPlannerAgent(self.llm)
        
        # Store chat history for each parent
        self.chat_history = {}
        
        # Generate historical data for timelines
        self.historical_data = self._generate_historical_data()
        
        # Generate medication compliance data
        self.medication_compliance = self._generate_medication_compliance()
    
    def _generate_historical_data(self) -> Dict[str, Dict[str, List]]:
        """Generate 7 days of historical health data for each parent"""
        data = {}
        today = datetime.now()
        
        for parent_id, parent in self.mock_data.parents.items():
            parent_data = {
                "dates": [],
                "blood_pressure_systolic": [],
                "blood_pressure_diastolic": [],
                "heart_rate": [],
                "blood_glucose": [],
                "weight": [],
                "temperature": [],
                "sleep_hours": []
            }
            
            # Set default weight based on parent
            if parent.name == "Margaret Chen":
                base_weight = 145.0  # lbs
            elif parent.name == "Robert Johnson":
                base_weight = 175.0  # lbs
            elif parent.name == "Elena Gonzalez":
                base_weight = 130.0  # lbs
            else:
                base_weight = 150.0  # default weight
            
            # Generate data for the last 7 days
            for i in range(7, 0, -1):
                date = today - timedelta(days=i)
                parent_data["dates"].append(date.strftime("%Y-%m-%d"))
                
                # Generate values based on parent's conditions
                if "Diabetes" in parent.conditions:
                    glucose_base = 130
                    glucose_var = 30
                else:
                    glucose_base = 95
                    glucose_var = 15
                
                if "Hypertension" in parent.conditions:
                    bp_sys_base = 145
                    bp_sys_var = 15
                    bp_dia_base = 90
                    bp_dia_var = 10
                else:
                    bp_sys_base = 125
                    bp_sys_var = 10
                    bp_dia_base = 80
                    bp_dia_var = 5
                
                # Add some random variation to simulate real-world data
                parent_data["blood_pressure_systolic"].append(bp_sys_base + random.randint(-bp_sys_var, bp_sys_var))
                parent_data["blood_pressure_diastolic"].append(bp_dia_base + random.randint(-bp_dia_var, bp_dia_var))
                parent_data["heart_rate"].append(75 + random.randint(-10, 10))
                parent_data["blood_glucose"].append(glucose_base + random.randint(-glucose_var, glucose_var))
                parent_data["weight"].append(base_weight + random.uniform(-0.5, 0.5))
                parent_data["temperature"].append(round(98.2 + random.uniform(-0.5, 0.8), 1))
                parent_data["sleep_hours"].append(round(6.5 + random.uniform(-1.5, 1.5), 1))
            
            data[parent_id] = parent_data
        
        return data
    
    def _generate_medication_compliance(self) -> Dict[str, Dict[str, List]]:
        """Generate medication compliance data for each parent"""
        compliance = {}
        
        for parent_id, parent in self.mock_data.parents.items():
            parent_compliance = {}
            
            for med in parent.medications:
                med_name = med["name"]
                med_compliance = []
                
                # Generate 7 days of compliance data
                for _ in range(7):
                    # 85% chance of taking medication
                    took_med = random.random() < 0.85
                    med_compliance.append(took_med)
                
                parent_compliance[med_name] = med_compliance
            
            compliance[parent_id] = parent_compliance
        
        return compliance
    
    def run_care_check(self, parent_id: str) -> Dict[str, Any]:
        """Run a full care check for a specific parent"""
        # Validate parent_id exists
        if parent_id not in self.mock_data.parents:
            raise ValueError(f"Parent ID {parent_id} not found")
        
        # Initialize state
        state = CareState(
            parent_id=parent_id,
            date=datetime.now().strftime("%Y-%m-%d"),
            health_metrics=[],
            medication_status=[],
            concerns=[],
            alerts=[],
            daily_summary="",
            action_items=[],
            family_notifications=[],
            emergency_level="normal"
        )
        
        # Run the analysis pipeline manually since we're not using langgraph here
        state = self.health_monitor.analyze_health_patterns(state)
        state = self.medication_agent.check_medication_compliance(state)
        
        # Check for emergency
        urgent_alerts = [a for a in state["alerts"] if a["severity"] == AlertLevel.URGENT.value]
        state["emergency_level"] = "urgent" if urgent_alerts else "normal"
        
        # Generate daily summary
        state = self.family_communicator.create_daily_summary(state)
        
        # Generate family notifications
        state = self.family_communicator.generate_family_updates(state)
        
        # Generate action items
        state = self.action_planner.generate_action_items(state)
        
        # Add timestamp
        result = dict(state)
        result["processed_at"] = datetime.now().isoformat()
        result["next_check"] = (datetime.now() + timedelta(hours=24)).isoformat()
        
        return result

    def get_parent_info(self, parent_id: str) -> Dict[str, Any]:
        """Get information about a specific parent"""
        parent = self.mock_data.get_parent_by_id(parent_id)
        family = self.mock_data.get_family_by_parent_id(parent_id)
        
        if not parent:
            return {"error": f"Parent ID {parent_id} not found"}
        
        return {
            "parent": asdict(parent),
            "family": [asdict(fm) for fm in family]
        }
    
    def get_health_timeline(self, parent_id: str) -> Dict[str, Any]:
        """Get historical health data for a specific parent"""
        if parent_id not in self.historical_data:
            return {"error": f"No historical data for parent ID {parent_id}"}
        
        return self.historical_data[parent_id]
    
    def get_medication_compliance(self, parent_id: str) -> Dict[str, Any]:
        """Get medication compliance data for a specific parent"""
        if parent_id not in self.medication_compliance:
            return {"error": f"No medication data for parent ID {parent_id}"}
        
        return self.medication_compliance[parent_id]
    
    def chat_with_system(self, parent_id: str, message: str) -> str:
        """Chat with the CareLoop system about a specific parent"""
        if parent_id not in self.chat_history:
            self.chat_history[parent_id] = []
        
        self.chat_history[parent_id].append({"role": "user", "content": message})
        
        # Generate context about the parent
        parent = self.mock_data.get_parent_by_id(parent_id)
        if not parent:
            response = "I couldn't find information about this parent."
            self.chat_history[parent_id].append({"role": "assistant", "content": response})
            return response
        
        # Create a prompt for the LLM
        prompt = f"""
        You are CareLoop, an AI assistant for family caregivers. 
        
        Parent information:
        - Name: {parent.name}
        - Age: {parent.age}
        - Health conditions: {', '.join(parent.conditions)}
        - Medications: {', '.join(med['name'] for med in parent.medications)}
        
        The caregiver has asked: {message}
        
        Provide a helpful, compassionate response focused on elderly care.
        """
        
        # Get response from LLM
        response = self.llm.generate(prompt)
        self.chat_history[parent_id].append({"role": "assistant", "content": response})
        
        return response
    
    def get_chat_history(self, parent_id: str) -> List[Dict[str, str]]:
        """Get chat history for a specific parent"""
        return self.chat_history.get(parent_id, [])

# Initialize the CareLoop system
care_system = CareLoopSystem()

def format_markdown_report(report: Dict[str, Any]) -> str:
    """Format the care report as markdown for Gradio display"""
    if not report:
        return ""
    
    md = []
    md.append(f"# {report['daily_summary']}")
    
    md.append("\n## Action Items")
    for item in report["action_items"]:
        md.append(f"- {item}")
    
    md.append("\n## Family Notifications")
    for notif in report["family_notifications"]:
        md.append(f"**To: {notif['recipient']}** ({notif['urgency']} priority)")
        md.append(f"```\n{notif['message']}\n```")
    
    if report["alerts"]:
        md.append("\n## Alerts")
        for alert in report["alerts"]:
            md.append(f"- **{alert['severity'].upper()}**: {alert['message']}")
            md.append(f"  - *Recommended Action:* {alert['recommended_action']}")
    
    return "\n".join(md)

def format_parent_info(parent_info: Dict[str, Any]) -> str:
    """Format parent info as markdown for Gradio display"""
    if not parent_info or "error" in parent_info:
        return "No parent information available"
    
    parent = parent_info["parent"]
    family = parent_info["family"]
    
    md = []
    md.append(f"# {parent['name']} ({parent['age']})")
    
    md.append("\n## Health Conditions")
    for condition in parent["conditions"]:
        md.append(f"- {condition}")
    
    md.append("\n## Medications")
    for med in parent["medications"]:
        times = ", ".join(med["times"])
        md.append(f"- {med['name']} ({med['dosage']}) - {med['frequency']} at {times}")
    
    md.append("\n## Family Caregivers")
    for member in family:
        md.append(f"- {member['name']} ({member['relationship']}) - {member['role']}")
        md.append(f"  - Contact: {member['phone']} | {member['email']}")
    
    return "\n".join(md)

def run_care_check(parent_id: str) -> tuple:
    """Run care check and return formatted results"""
    try:
        # Get parent info
        parent_info = care_system.get_parent_info(parent_id)
        parent_md = format_parent_info(parent_info)
        
        # Run care check
        report = care_system.run_care_check(parent_id)
        report_md = format_markdown_report(report)
        
        # Create metrics display
        health_metrics = len(report["health_metrics"])
        medication_events = len(report["medication_status"])
        alerts = len(report["alerts"])
        actions = len(report["action_items"])
        notifications = len(report["family_notifications"])
        
        # Get health timeline data
        timeline_data = care_system.get_health_timeline(parent_id)
        timeline_plot = create_health_timeline_plot(timeline_data)
        
        # Get medication compliance data
        compliance_data = care_system.get_medication_compliance(parent_id)
        compliance_plot = create_medication_compliance_plot(compliance_data)
        
        return parent_md, report_md, health_metrics, medication_events, alerts, actions, notifications, timeline_plot, compliance_plot
    except Exception as e:
        return f"Error: {str(e)}", "", 0, 0, 0, 0, 0, None, None

def create_health_timeline_plot(timeline_data: Dict[str, List]) -> gr.Plot:
    """Create a plot of health metrics over time"""
    if "error" in timeline_data:
        fig, ax = plt.subplots(figsize=(10, 6))
        ax.text(0.5, 0.5, "No timeline data available", ha='center', va='center')
        return fig
    
    # Create a figure with multiple subplots
    fig, axs = plt.subplots(3, 1, figsize=(10, 10), sharex=True)
    fig.suptitle("Health Metrics Over Time", fontsize=16)
    
    # Plot blood pressure and heart rate
    ax1 = axs[0]
    dates = timeline_data["dates"]
    ax1.plot(dates, timeline_data["blood_pressure_systolic"], 'r-', label='Systolic BP')
    ax1.plot(dates, timeline_data["blood_pressure_diastolic"], 'b-', label='Diastolic BP')
    ax1.set_ylabel('Blood Pressure (mmHg)')
    ax1.grid(True)
    ax1.legend(loc='upper left')
    
    # Add heart rate on secondary y-axis
    ax1_hr = ax1.twinx()
    ax1_hr.plot(dates, timeline_data["heart_rate"], 'g-', label='Heart Rate')
    ax1_hr.set_ylabel('Heart Rate (bpm)')
    ax1_hr.legend(loc='upper right')
    
    # Plot blood glucose
    ax2 = axs[1]
    ax2.plot(dates, timeline_data["blood_glucose"], 'm-', label='Blood Glucose')
    ax2.set_ylabel('Blood Glucose (mg/dL)')
    ax2.grid(True)
    ax2.legend()
    
    # Plot weight and temperature
    ax3 = axs[2]
    ax3.plot(dates, timeline_data["weight"], 'k-', label='Weight')
    ax3.set_xlabel('Date')
    ax3.set_ylabel('Weight (lbs)')
    ax3.grid(True)
    ax3.legend(loc='upper left')
    
    # Add temperature on secondary y-axis
    ax3_temp = ax3.twinx()
    ax3_temp.plot(dates, timeline_data["temperature"], 'c-', label='Temperature')
    ax3_temp.set_ylabel('Temperature (°F)')
    ax3_temp.legend(loc='upper right')
    
    plt.tight_layout()
    return fig

def create_medication_compliance_plot(compliance_data: Dict[str, List]) -> gr.Plot:
    """Create a plot of medication compliance"""
    if "error" in compliance_data:
        fig, ax = plt.subplots(figsize=(10, 6))
        ax.text(0.5, 0.5, "No medication compliance data available", ha='center', va='center')
        return fig
    
    # Create a figure
    fig, ax = plt.subplots(figsize=(10, 6))
    fig.suptitle("7-Day Medication Compliance", fontsize=16)
    
    # Set up data
    medications = list(compliance_data.keys())
    dates = [f"Day {i+1}" for i in range(7)]
    
    # Create a matrix of compliance data
    compliance_matrix = np.zeros((len(medications), 7))
    for i, med in enumerate(medications):
        for j in range(7):
            compliance_matrix[i, j] = 1 if compliance_data[med][j] else 0
    
    # Create heatmap
    im = ax.imshow(compliance_matrix, cmap='RdYlGn', aspect='auto', vmin=0, vmax=1)
    
    # Configure axes
    ax.set_xticks(np.arange(len(dates)))
    ax.set_yticks(np.arange(len(medications)))
    ax.set_xticklabels(dates)
    ax.set_yticklabels(medications)
    
    # Add text annotations
    for i in range(len(medications)):
        for j in range(len(dates)):
            text = "✓" if compliance_matrix[i, j] == 1 else "✗"
            color = "black" if compliance_matrix[i, j] == 1 else "white"
            ax.text(j, i, text, ha="center", va="center", color=color, fontweight="bold")
    
    ax.set_xlabel("Day")
    ax.set_title("✓ = Taken, ✗ = Missed")
    plt.tight_layout()
    
    return fig

# Create Gradio interface
with gr.Blocks(title="CareLoop AI - Family Caregiving Platform", theme=gr.themes.Soft()) as demo:
    gr.Markdown("# 🏠 CareLoop - AI-Powered Family Caregiving")
    gr.Markdown("Select a parent and run a care check to see AI analysis of their health and care needs.")
    
    # Store parent_id as a state variable
    current_parent_id = gr.State("parent_001")
    
    with gr.Row():
        with gr.Column(scale=1):
            parent_dropdown = gr.Dropdown(
                choices=[
                    "Margaret Chen (78) - Diabetes/Hypertension",
                    "Robert Johnson (72) - Stroke Recovery",
                    "Elena Gonzalez (81) - Early Alzheimer's"
                ],
                value="Margaret Chen (78) - Diabetes/Hypertension",
                label="Select Parent"
            )
            
            run_button = gr.Button("🔄 Run Care Check", variant="primary")
            
            with gr.Accordion("About CareLoop", open=False):
                gr.Markdown("""
                ## About CareLoop
                
                CareLoop is an AI-powered platform that helps families care for elderly relatives by:
                
                - Monitoring health metrics and medication compliance
                - Generating personalized care insights and recommendations
                - Coordinating communication between family caregivers
                - Detecting potential health issues early
                
                This demo showcases how AI can transform elderly care by analyzing complex health data
                and generating actionable insights for family caregivers.
                """)
        
        with gr.Column(scale=2):
            with gr.Tabs():
                with gr.Tab("Care Analysis"):
                    with gr.Row():
                        with gr.Column(scale=1):
                            parent_info = gr.Markdown("Select a parent to view their information")
                        
                        with gr.Column(scale=2):
                            care_report = gr.Markdown("Run a care check to see the AI analysis")
                
                with gr.Tab("Health Timeline"):
                    timeline_plot = gr.Plot(label="Health Metrics Over Time")
                
                with gr.Tab("Medication Tracker"):
                    compliance_plot = gr.Plot(label="Medication Compliance")
                
                with gr.Tab("Care Metrics"):
                    with gr.Row():
                        metric1 = gr.Number(label="Health Data Points", value=0)
                        metric2 = gr.Number(label="Medication Events", value=0)
                        metric3 = gr.Number(label="Alerts Generated", value=0)
                        metric4 = gr.Number(label="Action Items", value=0)
                        metric5 = gr.Number(label="Family Notifications", value=0)
                
                with gr.Tab("Care Assistant"):
                    chatbot = gr.Chatbot(label="Chat with CareLoop")
                    msg = gr.Textbox(
                        placeholder="Ask a question about the patient's care...",
                        show_label=False
                    )
                    clear = gr.Button("Clear")
    
    # Set up the event handlers
    parent_map = {
        "Margaret Chen (78) - Diabetes/Hypertension": "parent_001",
        "Robert Johnson (72) - Stroke Recovery": "parent_002",
        "Elena Gonzalez (81) - Early Alzheimer's": "parent_003"
    }
    
    def get_parent_id(display_name):
        return parent_map.get(display_name, "parent_001")
    
    def update_parent_id(display_name):
        parent_id = get_parent_id(display_name)
        return parent_id
    
    parent_dropdown.change(
        fn=update_parent_id,
        inputs=[parent_dropdown],
        outputs=[current_parent_id]
    )
    
    run_button.click(
        fn=lambda p_id: run_care_check(p_id),
        inputs=[current_parent_id],
        outputs=[parent_info, care_report, metric1, metric2, metric3, metric4, metric5, timeline_plot, compliance_plot]
    )
    
    # Also update parent info when dropdown changes
    parent_dropdown.change(
        fn=lambda p_id: (format_parent_info(care_system.get_parent_info(p_id)), "", 0, 0, 0, 0, 0, 
                         create_health_timeline_plot(care_system.get_health_timeline(p_id)),
                         create_medication_compliance_plot(care_system.get_medication_compliance(p_id))),
        inputs=[current_parent_id],
        outputs=[parent_info, care_report, metric1, metric2, metric3, metric4, metric5, timeline_plot, compliance_plot]
    )
    
    # Chat functionality - modified to avoid Gradio version compatibility issues
    def submit_message(p_id, message, history):
        if not message or message.strip() == "":
            return "", history
        response = care_system.chat_with_system(p_id, message)
        new_history = history + [[message, response]]
        return "", new_history
    
    def clear_chat():
        return []
    
    msg.submit(
        fn=submit_message,
        inputs=[current_parent_id, msg, chatbot],
        outputs=[msg, chatbot]
    )
    
    clear.click(fn=clear_chat, inputs=[], outputs=[chatbot])

if __name__ == "__main__":
    print("🚀 Starting CareLoop Gradio Interface")
    print("=" * 50)
    
    # Display available parents
    mock_data = MockDataGenerator()
    print("🏠 Available Parent Profiles:")
    for parent_id, parent in mock_data.parents.items():
        family_count = len(mock_data.families.get(parent_id, []))
        conditions = ", ".join(parent.conditions[:2]) + ("..." if len(parent.conditions) > 2 else "")
        print(f"  • {parent.name} ({parent.age}) - ID: {parent_id}")
        print(f"    Health: {conditions}")
        print(f"    Family caregivers: {family_count}")
    
    print("\n" + "=" * 50)
    print("💡 Note: Using Nebius API for LLM. Set NEBIUS_API_KEY environment variable for production use.")
    print("=" * 50)
    
    # Launch the Gradio app
    try:
        # Launch with standard options for Gradio 4.16.0
        demo.launch(share=False, show_error=True) 
    except Exception as e:
        print(f"Error launching Gradio app: {e}")
        print("\nTrying alternative launch method...")
        try:
            # Alternative launch approach with minimal options
            demo.queue(False).launch(share=False)
        except Exception as e2:
            print(f"Alternative launch also failed: {e2}")
            print("\nPlease try updating Gradio with: pip install --upgrade gradio")