GlycoAI
AI-Powered Glucose Chatbot
Connect your Dexcom CGM data OR try demo users and chat with AI for personalized glucose insights
"""
GlycoAI - AI-Powered Glucose Insights
Main Gradio application with both demo users AND real Dexcom OAuth
"""
import gradio as gr
import plotly.graph_objects as go
import plotly.express as px
from datetime import datetime, timedelta
import pandas as pd
from typing import Optional, Tuple, List
import logging
import os
import webbrowser
import urllib.parse
# Load environment variables from .env file
from dotenv import load_dotenv
load_dotenv()
# Import the Mistral chat class and unified data manager
from mistral_chat import GlucoBuddyMistralChat, validate_environment
from unified_data_manager import UnifiedDataManager
# Setup logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
# Import our custom functions
from apifunctions import (
DexcomAPI,
GlucoseAnalyzer,
DEMO_USERS,
format_glucose_data_for_display
)
# Import real Dexcom OAuth (now working!)
try:
from dexcom_real_auth_system import DexcomRealAPI
REAL_OAUTH_AVAILABLE = True
logger.info("✅ Real Dexcom OAuth available")
except ImportError as e:
REAL_OAUTH_AVAILABLE = False
logger.warning(f"⚠️ Real Dexcom OAuth not available: {e}")
class GlucoBuddyApp:
"""Main application class for GlucoBuddy with demo users AND real OAuth"""
def __init__(self):
# Validate environment before initializing
if not validate_environment():
raise ValueError("Environment validation failed - check your .env file or environment variables")
# Single data manager for consistency
self.data_manager = UnifiedDataManager()
# Chat interface (will use data manager's context)
self.mistral_chat = GlucoBuddyMistralChat()
# Real OAuth API (if available)
self.real_api = DexcomRealAPI(environment="sandbox") if REAL_OAUTH_AVAILABLE else None
# UI state
self.chat_history = []
self.current_user_type = None # "demo" or "real"
def select_demo_user(self, user_key: str) -> Tuple[str, str]:
"""Handle demo user selection and load data consistently"""
if user_key not in DEMO_USERS:
return "❌ Invalid user selection", gr.update(visible=False)
try:
# Load data through unified manager
load_result = self.data_manager.load_user_data(user_key)
if not load_result['success']:
return f"❌ {load_result['message']}", gr.update(visible=False)
user = self.data_manager.current_user
self.current_user_type = "demo"
# Update Mistral chat with the same context
self._sync_chat_with_data_manager()
# Clear chat history when switching users
self.chat_history = []
self.mistral_chat.clear_conversation()
return (
f"Connected: {user.name} ({user.device_type}) - DEMO DATA - Click 'Load Data' to begin",
gr.update(visible=True)
)
except Exception as e:
logger.error(f"Demo user selection failed: {str(e)}")
return f"❌ Connection failed: {str(e)}", gr.update(visible=False)
def start_real_oauth(self) -> str:
"""Start real Dexcom OAuth process"""
if not REAL_OAUTH_AVAILABLE:
return """
❌ **Real Dexcom OAuth Not Available**
The real authentication module is not properly configured.
Please ensure:
1. dexcom_real_auth_system.py exists and imports correctly
2. You have valid Dexcom developer credentials
3. All dependencies are installed
For now, please use the demo users above for instant access to realistic glucose data.
"""
try:
# Start OAuth flow
success = self.real_api.start_oauth_flow()
if success:
return f"""
🚀 **Real Dexcom OAuth Started**
**SIMPLIFIED PROCESS FOR PORT 7860:**
1. ✅ Browser should have opened to Dexcom login page
2. 📝 Log in with your **real Dexcom account credentials**
- For sandbox testing: `sandboxuser1@dexcom.com` / `Dexcom123!`
3. ✅ Authorize GlycoAI to access your data
4. ❌ **You will get a 404 error - THIS IS EXPECTED!**
5. 📋 **Copy ONLY the authorization code** from the URL
**Example callback URL:**
`http://localhost:7860/callback?code=ABC123XYZ&state=...`
**Copy just this part:** `ABC123XYZ`
**Why simpler?** Your token generator script works perfectly with just the code, so we're using the same approach!
"""
else:
return "❌ Failed to start OAuth process. Check console for details."
except Exception as e:
logger.error(f"OAuth start error: {e}")
return f"❌ OAuth error: {str(e)}"
def complete_real_oauth(self, auth_code_input: str) -> Tuple[str, str]:
"""Complete real OAuth with authorization code (like the working script)"""
if not REAL_OAUTH_AVAILABLE:
return "❌ Real OAuth not available", gr.update(visible=False)
if not auth_code_input or not auth_code_input.strip():
return "❌ Please paste the authorization code", gr.update(visible=False)
try:
# Clean up the input - handle both full URLs and just codes
auth_code = self._extract_auth_code(auth_code_input.strip())
if not auth_code:
return "❌ No authorization code found in input", gr.update(visible=False)
logger.info(f"Processing authorization code: {auth_code[:20]}...")
# Use the same method as the working script - direct token exchange
success = self.real_api.exchange_code_for_tokens(auth_code)
if success:
logger.info("✅ Token exchange successful")
# Load real data into data manager
real_data_result = self._load_real_dexcom_data()
if real_data_result['success']:
self.current_user_type = "real"
# Update chat context
self._sync_chat_with_data_manager()
# Clear chat history for new user
self.chat_history = []
self.mistral_chat.clear_conversation()
return (
f"✅ Connected: Real Dexcom User - LIVE DATA - Click 'Load Data' to begin",
gr.update(visible=True)
)
else:
return f"❌ Data loading failed: {real_data_result['message']}", gr.update(visible=False)
else:
return "❌ Token exchange failed - check the authorization code", gr.update(visible=False)
except Exception as e:
logger.error(f"OAuth completion error: {e}")
return f"❌ OAuth completion failed: {str(e)}", gr.update(visible=False)
def _extract_auth_code(self, input_text: str) -> str:
"""Extract authorization code from various input formats"""
try:
# If it's a full URL, parse it
if input_text.startswith('http'):
parsed_url = urllib.parse.urlparse(input_text)
query_params = urllib.parse.parse_qs(parsed_url.query)
if 'code' in query_params:
return query_params['code'][0]
else:
logger.warning(f"No 'code' parameter found in URL: {input_text}")
return ""
else:
# Assume it's just the authorization code
# Remove any "code=" prefix if present
if input_text.startswith('code='):
return input_text[5:]
else:
return input_text
except Exception as e:
logger.error(f"Error extracting auth code: {e}")
return ""
def _load_real_dexcom_data(self) -> dict:
"""Load real Dexcom data through the unified data manager"""
try:
# Get data range
data_range = self.real_api.get_data_range()
# Get glucose data (last 14 days)
end_time = datetime.now()
start_time = end_time - timedelta(days=14)
egv_data = self.real_api.get_egv_data(
start_date=start_time.isoformat(),
end_date=end_time.isoformat()
)
# Get events data
events_data = self.real_api.get_events_data(
start_date=start_time.isoformat(),
end_date=end_time.isoformat()
)
# Create a real user profile
from dataclasses import dataclass
@dataclass
class RealDexcomUser:
name: str = "Real Dexcom User"
age: int = 0
device_type: str = "Real Dexcom Device"
username: str = "authenticated_user"
password: str = "oauth_token"
description: str = "Authenticated real Dexcom user with live data"
diabetes_type: str = "Real Patient"
years_with_diabetes: int = 0
typical_glucose_pattern: str = "real_data"
real_user = RealDexcomUser()
# Process the real data through unified data manager
# Convert to format expected by data manager
formatted_data = {
"data_range": data_range,
"egv_data": egv_data,
"events_data": events_data,
"source": "real_dexcom_api"
}
# Load into data manager
self.data_manager.current_user = real_user
self.data_manager.processed_glucose_data = pd.DataFrame(egv_data) if egv_data else pd.DataFrame()
if not self.data_manager.processed_glucose_data.empty:
# Process timestamps
self.data_manager.processed_glucose_data['systemTime'] = pd.to_datetime(
self.data_manager.processed_glucose_data['systemTime']
)
self.data_manager.processed_glucose_data['displayTime'] = pd.to_datetime(
self.data_manager.processed_glucose_data['displayTime']
)
self.data_manager.processed_glucose_data['value'] = pd.to_numeric(
self.data_manager.processed_glucose_data['value'], errors='coerce'
)
# Calculate stats
from apifunctions import GlucoseAnalyzer
self.data_manager.calculated_stats = GlucoseAnalyzer.calculate_basic_stats(
self.data_manager.processed_glucose_data
)
self.data_manager.identified_patterns = GlucoseAnalyzer.identify_patterns(
self.data_manager.processed_glucose_data
)
logger.info(f"Loaded real Dexcom data: {len(egv_data)} glucose readings")
return {
'success': True,
'message': f'Loaded {len(egv_data)} real glucose readings'
}
except Exception as e:
logger.error(f"Failed to load real Dexcom data: {e}")
return {
'success': False,
'message': f'Failed to load real data: {str(e)}'
}
def load_glucose_data(self) -> Tuple[str, go.Figure]:
"""Load and display glucose data using unified manager"""
if not self.data_manager.current_user:
return "Please select a user first (demo or real Dexcom)", None
try:
# For real users, we already loaded the data during OAuth
if self.current_user_type == "real":
if self.data_manager.processed_glucose_data.empty:
return "No real glucose data available", None
else:
# For demo users, force reload data to ensure freshness
load_result = self.data_manager.load_user_data(
self._get_current_user_key(),
force_reload=True
)
if not load_result['success']:
return load_result['message'], None
# Get unified stats
stats = self.data_manager.get_stats_for_ui()
chart_data = self.data_manager.get_chart_data()
# Sync chat with fresh data
self._sync_chat_with_data_manager()
if chart_data is None or chart_data.empty:
return "No glucose data available", None
# Build data summary with CONSISTENT metrics
user = self.data_manager.current_user
data_points = stats.get('total_readings', 0)
avg_glucose = stats.get('average_glucose', 0)
std_glucose = stats.get('std_glucose', 0)
min_glucose = stats.get('min_glucose', 0)
max_glucose = stats.get('max_glucose', 0)
time_in_range = stats.get('time_in_range_70_180', 0)
time_below_range = stats.get('time_below_70', 0)
time_above_range = stats.get('time_above_180', 0)
gmi = stats.get('gmi', 0)
cv = stats.get('cv', 0)
# Calculate date range
end_date = datetime.now()
start_date = end_date - timedelta(days=14)
# Determine data source
data_source = "REAL DEXCOM DATA" if self.current_user_type == "real" else "DEMO DATA"
data_summary = f"""
## 📊 Data Summary for {user.name}
### Basic Information
• **Data Type:** {data_source}
• **Analysis Period:** {start_date.strftime('%B %d, %Y')} to {end_date.strftime('%B %d, %Y')} (14 days)
• **Total Readings:** {data_points:,} glucose measurements
• **Device:** {user.device_type}
• **Data Source:** {stats.get('data_source', 'unknown').upper()}
### Glucose Statistics
• **Average Glucose:** {avg_glucose:.1f} mg/dL
• **Standard Deviation:** {std_glucose:.1f} mg/dL
• **Coefficient of Variation:** {cv:.1f}%
• **Glucose Range:** {min_glucose:.0f} - {max_glucose:.0f} mg/dL
• **GMI (Glucose Management Indicator):** {gmi:.1f}%
### Time in Range Analysis
• **Time in Range (70-180 mg/dL):** {time_in_range:.1f}%
• **Time Below Range (<70 mg/dL):** {time_below_range:.1f}%
• **Time Above Range (>180 mg/dL):** {time_above_range:.1f}%
### Clinical Targets
• **Target Time in Range:** >70% (Current: {time_in_range:.1f}%)
• **Target Time Below Range:** <4% (Current: {time_below_range:.1f}%)
• **Target CV:** <36% (Current: {cv:.1f}%)
### Data Validation
• **In Range Count:** {stats.get('in_range_count', 0)} readings
• **Below Range Count:** {stats.get('below_range_count', 0)} readings
• **Above Range Count:** {stats.get('above_range_count', 0)} readings
• **Total Verified:** {stats.get('in_range_count', 0) + stats.get('below_range_count', 0) + stats.get('above_range_count', 0)} readings
### 14-Day Analysis Benefits
• **Enhanced Pattern Recognition:** Captures full weekly cycles and variations
• **Improved Trend Analysis:** Identifies consistent patterns vs. one-time events
• **Better Clinical Insights:** More reliable data for healthcare decisions
• **AI Consistency:** Same data used for chat analysis and UI display
### Authentication Status
• **User Type:** {self.current_user_type.upper() if self.current_user_type else 'Unknown'}
• **OAuth Status:** {'✅ Authenticated with real Dexcom account' if self.current_user_type == 'real' else '🎭 Using demo data for testing'}
"""
chart = self.create_glucose_chart()
return data_summary, chart
except Exception as e:
logger.error(f"Failed to load glucose data: {str(e)}")
return f"Failed to load glucose data: {str(e)}", None
def _sync_chat_with_data_manager(self):
"""Ensure chat uses the same data as the UI"""
try:
# Get context from unified data manager
context = self.data_manager.get_context_for_agent()
# Update chat's internal data to match
if not context.get("error"):
self.mistral_chat.current_user = self.data_manager.current_user
self.mistral_chat.current_glucose_data = self.data_manager.processed_glucose_data
self.mistral_chat.current_stats = self.data_manager.calculated_stats
self.mistral_chat.current_patterns = self.data_manager.identified_patterns
logger.info(f"Synced chat with data manager - TIR: {self.data_manager.calculated_stats.get('time_in_range_70_180', 0):.1f}%")
except Exception as e:
logger.error(f"Failed to sync chat with data manager: {e}")
def _get_current_user_key(self) -> str:
"""Get the current user key"""
if not self.data_manager.current_user:
return ""
# Find the key for current user
for key, user in DEMO_USERS.items():
if user == self.data_manager.current_user:
return key
return ""
def get_template_prompts(self) -> List[str]:
"""Get template prompts based on current user data"""
if not self.data_manager.current_user or not self.data_manager.calculated_stats:
return [
"What should I know about managing my diabetes?",
"How can I improve my glucose control?"
]
stats = self.data_manager.calculated_stats
time_in_range = stats.get('time_in_range_70_180', 0)
time_below_70 = stats.get('time_below_70', 0)
templates = []
if time_in_range < 70:
templates.append(f"My time in range is {time_in_range:.1f}% which is below the 70% target. What specific strategies can help me improve it?")
else:
templates.append(f"My time in range is {time_in_range:.1f}% which meets the target. How can I maintain this level of control?")
if time_below_70 > 4:
templates.append(f"I'm experiencing {time_below_70:.1f}% time below 70 mg/dL. What can I do to prevent these low episodes?")
else:
templates.append("What are the best practices for preventing hypoglycemia in my situation?")
# Add data source specific template
if self.current_user_type == "real":
templates.append("This is my real Dexcom data. What insights can you provide about my actual glucose patterns?")
else:
templates.append("Based on this demo data, what would you recommend for someone with similar patterns?")
return templates
def chat_with_mistral(self, message: str, history: List) -> Tuple[str, List]:
"""Handle chat interaction with Mistral using unified data"""
if not message.strip():
return "", history
if not self.data_manager.current_user:
response = "Please select a user first (demo or real Dexcom) to get personalized insights about glucose data."
history.append([message, response])
return "", history
try:
# Ensure chat is synced with latest data
self._sync_chat_with_data_manager()
# Send message to Mistral chat
result = self.mistral_chat.chat_with_mistral(message)
if result['success']:
response = result['response']
# Add data consistency note
validation = self.data_manager.validate_data_consistency()
if validation.get('valid'):
data_age = validation.get('data_age_minutes', 0)
if data_age > 10: # Warn if data is old
response += f"\n\n📊 *Note: Analysis based on data from {data_age} minutes ago. Reload data for most current insights.*"
# Add data source context
data_type = "real Dexcom data" if self.current_user_type == "real" else "demo data"
if self.current_user_type == "real":
response += f"\n\n🔐 *This analysis is based on your real Dexcom data from your authenticated account.*"
else:
response += f"\n\n🎭 *This analysis is based on demo data for testing purposes.*"
# Add context note if no user data was included
if not result.get('context_included', True):
response += f"\n\n💡 *For more personalized advice, make sure your glucose data is loaded.*"
else:
response = f"I apologize, but I encountered an error: {result.get('error', 'Unknown error')}. Please try again or rephrase your question."
history.append([message, response])
return "", history
except Exception as e:
logger.error(f"Chat error: {str(e)}")
error_response = f"I apologize, but I encountered an error while processing your question: {str(e)}. Please try rephrasing your question."
history.append([message, error_response])
return "", history
def use_template_prompt(self, template_text: str) -> str:
"""Use a template prompt in the chat"""
return template_text
def clear_chat_history(self) -> List:
"""Clear chat history"""
self.chat_history = []
self.mistral_chat.clear_conversation()
return []
def create_glucose_chart(self) -> Optional[go.Figure]:
"""Create an interactive glucose chart using unified data"""
chart_data = self.data_manager.get_chart_data()
if chart_data is None or chart_data.empty:
return None
fig = go.Figure()
# Color code based on glucose ranges
colors = []
for value in chart_data['value']:
if value < 70:
colors.append('#E74C3C') # Red for low
elif value > 180:
colors.append('#F39C12') # Orange for high
else:
colors.append('#27AE60') # Green for in range
fig.add_trace(go.Scatter(
x=chart_data['systemTime'],
y=chart_data['value'],
mode='lines+markers',
name='Glucose',
line=dict(color='#2E86AB', width=2),
marker=dict(size=4, color=colors),
hovertemplate='%{y} mg/dL
%{x}
AI-Powered Glucose Chatbot
Connect your Dexcom CGM data OR try demo users and chat with AI for personalized glucose insights
⚠️ Important Medical Disclaimer
GlycoAI is for informational and educational purposes only. Always consult your healthcare provider before making any changes to your diabetes management plan. This tool does not replace professional medical advice.
🔒 Your data is processed securely and not stored permanently.
💡 Powered by Dexcom API integration and Mistral AI.
{"🎭 Demo data available instantly • 🔐 Real OAuth: " + ("Available" if REAL_OAUTH_AVAILABLE else "Not configured")}