#!/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")