Spaces:
Running
Running
#!/usr/bin/env python3 | |
""" | |
GlycoAI - AI-Powered Glucose Insights | |
Complete application with Demo Users + Dexcom Sandbox OAuth | |
IMPROVED UI VERSION - Clean, readable design with blue theme | |
""" | |
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 | |
# 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 Dexcom Sandbox OAuth | |
try: | |
from dexcom_sandbox_oauth import DexcomSandboxIntegration, DexcomSandboxUser | |
DEXCOM_SANDBOX_AVAILABLE = True | |
logger.info("β Dexcom Sandbox OAuth available") | |
except ImportError as e: | |
DEXCOM_SANDBOX_AVAILABLE = False | |
logger.warning(f"β οΈ Dexcom Sandbox OAuth not available: {e}") | |
class GlycoAIApp: | |
"""Main application class for GlycoAI with demo users AND Dexcom Sandbox 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() | |
# Dexcom Sandbox OAuth API | |
self.dexcom_sandbox = DexcomSandboxIntegration() if DEXCOM_SANDBOX_AVAILABLE else None | |
# UI state | |
self.chat_history = [] | |
self.current_user_type = None # "demo" or "dexcom_sandbox" | |
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", | |
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 initialize_chat_with_prompts(self) -> List: | |
"""Initialize chat with demo prompts as conversation bubbles""" | |
if not self.data_manager.current_user: | |
return [ | |
[None, "π Welcome to GlycoAI! Please select a demo user or connect Dexcom Sandbox to get started."], | |
[None, "π‘ Once you load your glucose data, I'll provide personalized insights about your patterns and trends."] | |
] | |
templates = self.get_template_prompts() | |
# Create initial conversation with demo prompts | |
initial_chat = [ | |
[None, f"π Hi! I'm ready to analyze {self.data_manager.current_user.name}'s glucose data. Here are some quick ways to get started:"], | |
[None, f"π― **{templates[0] if templates else 'Analyze my recent glucose patterns and trends'}**"], | |
[None, f"β‘ **{templates[1] if len(templates) > 1 else 'What can I do to improve my glucose control?'}**"], | |
[None, f"π½οΈ **What are some meal management strategies for better glucose control?**"], | |
[None, "π¬ You can click on any of these questions above, or ask me anything about glucose management!"] | |
] | |
return initial_chat | |
def handle_demo_prompt_click(self, prompt_text: str, history: List) -> Tuple[str, List]: | |
"""Handle clicking on demo prompts in chat""" | |
# Remove the emoji and formatting from the prompt | |
clean_prompt = prompt_text.replace("π― **", "").replace("β‘ **", "").replace("π½οΈ **", "").replace("**", "") | |
# Process the prompt as if user typed it | |
return self.chat_with_mistral(clean_prompt, history) | |
def start_dexcom_sandbox_oauth(self) -> str: | |
"""Start Dexcom Sandbox OAuth process""" | |
if not DEXCOM_SANDBOX_AVAILABLE: | |
return """ | |
β **Dexcom Sandbox OAuth Not Available** | |
The Dexcom Sandbox authentication module is not properly configured. | |
Please ensure: | |
1. dexcom_sandbox_oauth.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 for Dexcom Sandbox | |
auth_url = self.dexcom_sandbox.oauth.generate_auth_url() | |
# Try to open browser automatically | |
try: | |
import webbrowser | |
webbrowser.open(auth_url) | |
browser_status = "β Browser opened automatically" | |
except: | |
browser_status = "β οΈ Please open the URL manually" | |
return f""" | |
π **Dexcom Sandbox OAuth Started** | |
{browser_status} | |
**π OAuth URL:** {auth_url} | |
**Step-by-Step Instructions:** | |
1. Browser should open automatically (or open URL above) | |
2. Select a sandbox user from the dropdown (SandboxUser6 recommended) | |
3. Click "Authorize" to grant access | |
4. **You will get a 404 error - THIS IS EXPECTED!** | |
5. Copy the COMPLETE callback URL from address bar | |
**Example callback URL:** | |
`http://localhost:7860/callback?code=ABC123XYZ&state=sandbox_test` | |
**Important:** Copy the entire URL (not just the code part)! | |
""" | |
except Exception as e: | |
logger.error(f"Dexcom Sandbox OAuth start error: {e}") | |
return f"β OAuth error: {str(e)}" | |
def complete_dexcom_sandbox_oauth(self, callback_url_input: str) -> Tuple[str, str]: | |
"""Complete Dexcom Sandbox OAuth with full callback URL""" | |
if not DEXCOM_SANDBOX_AVAILABLE: | |
return "β Dexcom Sandbox OAuth not available", gr.update(visible=False) | |
if not callback_url_input or not callback_url_input.strip(): | |
return "β Please paste the complete callback URL", gr.update(visible=False) | |
try: | |
callback_url = callback_url_input.strip() | |
logger.info(f"Processing Dexcom Sandbox callback: {callback_url[:50]}...") | |
# Use Dexcom Sandbox OAuth completion | |
status_message, show_interface = self.dexcom_sandbox.complete_oauth(callback_url) | |
if show_interface: | |
logger.info("β Dexcom Sandbox OAuth successful") | |
# Load Dexcom Sandbox data into data manager | |
sandbox_data_result = self._load_dexcom_sandbox_data() | |
if sandbox_data_result['success']: | |
self.current_user_type = "dexcom_sandbox" | |
# 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: Dexcom Sandbox User - OAuth Authenticated", | |
gr.update(visible=True) | |
) | |
else: | |
return f"β Dexcom Sandbox data loading failed: {sandbox_data_result['message']}", gr.update(visible=False) | |
else: | |
logger.error(f"Dexcom Sandbox OAuth failed: {status_message}") | |
return f"β {status_message}", gr.update(visible=False) | |
except Exception as e: | |
logger.error(f"Dexcom Sandbox OAuth completion error: {e}") | |
return f"β OAuth completion failed: {str(e)}", gr.update(visible=False) | |
def _load_dexcom_sandbox_data(self) -> dict: | |
"""Load Dexcom Sandbox data through the unified data manager""" | |
try: | |
# Get Dexcom Sandbox user profile | |
sandbox_profile = self.dexcom_sandbox.get_user_profile() | |
if not sandbox_profile: | |
return { | |
'success': False, | |
'message': 'No Dexcom Sandbox user profile available' | |
} | |
# Set in data manager (compatible with existing structure) | |
self.data_manager.current_user = sandbox_profile | |
self.data_manager.data_source = "dexcom_sandbox_oauth" | |
self.data_manager.data_loaded_at = datetime.now() | |
logger.info("β Dexcom Sandbox data integrated with data manager") | |
return { | |
'success': True, | |
'message': 'Dexcom Sandbox user profile loaded successfully' | |
} | |
except Exception as e: | |
logger.error(f"Failed to load Dexcom Sandbox data: {e}") | |
return { | |
'success': False, | |
'message': f'Failed to load OAuth data: {str(e)}' | |
} | |
def load_glucose_data(self) -> Tuple[str, go.Figure, str]: | |
"""Load and display glucose data using unified manager with notifications""" | |
if not self.data_manager.current_user: | |
return "Please select a user first (demo or Dexcom Sandbox)", None, "" | |
try: | |
# For Dexcom Sandbox users, load real data via OAuth | |
if self.current_user_type == "dexcom_sandbox": | |
overview, chart = self._load_dexcom_sandbox_glucose_data() | |
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 and build display | |
overview, chart = self._build_glucose_display() | |
# Create notification message based on user and data quality | |
notification = self._create_data_loaded_notification() | |
return overview, chart, notification | |
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 _create_data_loaded_notification(self) -> str: | |
"""Create appropriate notification based on loaded data""" | |
if not self.data_manager.current_user or not self.data_manager.calculated_stats: | |
return "" | |
user_name = self.data_manager.current_user.name | |
stats = self.data_manager.calculated_stats | |
tir = stats.get('time_in_range_70_180', 0) | |
cv = stats.get('cv', 0) | |
avg_glucose = stats.get('average_glucose', 0) | |
total_readings = stats.get('total_readings', 0) | |
# Special handling for Sarah (unstable patterns) | |
if user_name == "Sarah Thompson": | |
if tir < 50 and cv > 40: | |
notification = f""" | |
π¨ **DATA LOADED - CONCERNING PATTERNS DETECTED** | |
**Patient:** {user_name} ({total_readings:,} readings analyzed) | |
**β οΈ Critical Findings:** | |
β’ Time in Range: {tir:.1f}% (Target: >70%) | |
β’ High Variability: CV {cv:.1f}% (Target: <36%) | |
β’ Average Glucose: {avg_glucose:.1f} mg/dL | |
**π₯ Immediate Action Required** | |
β’ Frequent hypoglycemia detected | |
β’ Severe glucose instability | |
β’ Healthcare provider consultation recommended | |
*AI analysis ready - Click Chat tab for urgent insights* | |
""" | |
else: | |
notification = f""" | |
β **DATA LOADED SUCCESSFULLY** | |
**Patient:** {user_name} ({total_readings:,} readings analyzed) | |
**Time in Range:** {tir:.1f}% | **Average:** {avg_glucose:.1f} mg/dL | |
*14-day analysis complete - Ready for AI insights* | |
""" | |
else: | |
# For other users with better control | |
if tir >= 70: | |
notification = f""" | |
β **DATA LOADED - EXCELLENT CONTROL** | |
**Patient:** {user_name} ({total_readings:,} readings analyzed) | |
**Time in Range:** {tir:.1f}% β | **CV:** {cv:.1f}% | |
*Great glucose management - AI ready to help maintain control* | |
""" | |
else: | |
notification = f""" | |
π **DATA LOADED SUCCESSFULLY** | |
**Patient:** {user_name} ({total_readings:,} readings analyzed) | |
**Time in Range:** {tir:.1f}% | **Average:** {avg_glucose:.1f} mg/dL | |
*Analysis complete - AI ready to provide insights* | |
""" | |
return notification | |
def _load_dexcom_sandbox_glucose_data(self) -> Tuple[str, go.Figure]: | |
"""Load Dexcom Sandbox glucose data via OAuth""" | |
if not self.dexcom_sandbox.authenticated: | |
return "β Dexcom Sandbox not authenticated. Please complete OAuth first.", None | |
try: | |
# Load 14 days of data from Dexcom Sandbox | |
data_result = self.dexcom_sandbox.load_glucose_data(days=14) | |
if not data_result['success']: | |
return f"β {data_result['error']}", None | |
# Convert Dexcom Sandbox data to data manager format | |
self._convert_dexcom_sandbox_to_dataframe() | |
return self._build_glucose_display() | |
except Exception as e: | |
logger.error(f"Failed to load Dexcom Sandbox data: {e}") | |
return f"β Failed to load Dexcom Sandbox data: {str(e)}", None | |
def _convert_dexcom_sandbox_to_dataframe(self): | |
"""Convert Dexcom Sandbox glucose data to DataFrame format""" | |
try: | |
glucose_data = self.dexcom_sandbox.get_glucose_data_for_ui() | |
if not glucose_data: | |
raise Exception("No glucose data available from Dexcom Sandbox") | |
# Convert to DataFrame | |
df = pd.DataFrame(glucose_data) | |
# Ensure proper datetime conversion | |
df['systemTime'] = pd.to_datetime(df['systemTime']) | |
df['displayTime'] = pd.to_datetime(df['displayTime']) | |
df['value'] = pd.to_numeric(df['value'], errors='coerce') | |
# Sort by time | |
df = df.sort_values('systemTime') | |
# Set in data manager | |
self.data_manager.processed_glucose_data = df | |
# Calculate statistics using existing analyzer | |
self.data_manager.calculated_stats = self.data_manager._calculate_unified_stats() | |
self.data_manager.identified_patterns = GlucoseAnalyzer.identify_patterns(df) | |
logger.info(f"β Converted {len(df)} Dexcom Sandbox readings to DataFrame") | |
except Exception as e: | |
logger.error(f"Failed to convert Dexcom Sandbox data: {e}") | |
raise | |
def _build_glucose_display(self) -> Tuple[str, go.Figure]: | |
"""Build glucose data display (common for demo and Dexcom Sandbox)""" | |
# 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 | |
if self.current_user_type == "dexcom_sandbox": | |
data_source = "Dexcom Sandbox OAuth" | |
oauth_status = "β Authenticated Dexcom Sandbox with working OAuth" | |
else: | |
data_source = "Demo Data" | |
oauth_status = "π Using demo data for testing" | |
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} | |
### 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}%) | |
### Authentication Status | |
β’ **User Type:** {self.current_user_type.upper() if self.current_user_type else 'Unknown'} | |
β’ **OAuth Status:** {oauth_status} | |
""" | |
chart = self.create_glucose_chart() | |
return data_summary, chart | |
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 == "dexcom_sandbox": | |
templates.append("This is my Dexcom Sandbox OAuth-authenticated data. What insights can you provide about these 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 Dexcom Sandbox) 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 | |
if self.current_user_type == "dexcom_sandbox": | |
response += f"\n\nπ *This analysis is based on your OAuth-authenticated Dexcom Sandbox data.*" | |
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 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('#3498DB') # Blue for in range | |
fig.add_trace(go.Scatter( | |
x=chart_data['systemTime'], | |
y=chart_data['value'], | |
mode='lines+markers', | |
name='Glucose', | |
line=dict(color='#2980B9', width=2), | |
marker=dict(size=4, color=colors), | |
hovertemplate='<b>%{y} mg/dL</b><br>%{x}<extra></extra>' | |
)) | |
# Add target range shading | |
fig.add_hrect( | |
y0=70, y1=180, | |
fillcolor="rgba(52, 152, 219, 0.1)", | |
layer="below", | |
line_width=0, | |
annotation_text="Target Range", | |
annotation_position="top left" | |
) | |
# Add reference lines | |
fig.add_hline(y=70, line_dash="dash", line_color="#E67E22", | |
annotation_text="Low (70 mg/dL)", annotation_position="right") | |
fig.add_hline(y=180, line_dash="dash", line_color="#E67E22", | |
annotation_text="High (180 mg/dL)", annotation_position="right") | |
fig.add_hline(y=54, line_dash="dot", line_color="#E74C3C", | |
annotation_text="Severe Low (54 mg/dL)", annotation_position="right") | |
fig.add_hline(y=250, line_dash="dot", line_color="#E74C3C", | |
annotation_text="Severe High (250 mg/dL)", annotation_position="right") | |
# Get current stats for title | |
stats = self.data_manager.get_stats_for_ui() | |
tir = stats.get('time_in_range_70_180', 0) | |
if self.current_user_type == "dexcom_sandbox": | |
data_type = "Dexcom Sandbox" | |
else: | |
data_type = "Demo Data" | |
fig.update_layout( | |
title={ | |
'text': f"14-Day Glucose Trends - {self.data_manager.current_user.name} ({data_type} - TIR: {tir:.1f}%)", | |
'x': 0.5, | |
'xanchor': 'center' | |
}, | |
xaxis_title="Time", | |
yaxis_title="Glucose (mg/dL)", | |
hovermode='x unified', | |
height=500, | |
showlegend=False, | |
plot_bgcolor='rgba(0,0,0,0)', | |
paper_bgcolor='rgba(0,0,0,0)', | |
font=dict(size=12), | |
margin=dict(l=60, r=60, t=80, b=60) | |
) | |
fig.update_xaxes(showgrid=True, gridwidth=1, gridcolor='rgba(128,128,128,0.2)') | |
fig.update_yaxes(showgrid=True, gridwidth=1, gridcolor='rgba(128,128,128,0.2)') | |
return fig | |
def create_interface(): | |
"""Create the Gradio interface with improved, cleaner design""" | |
app = GlycoAIApp() | |
# Clean blue-themed CSS | |
custom_css = """ | |
/* Main header styling */ | |
.main-header { | |
text-align: center; | |
background: linear-gradient(135deg, #3498db 0%, #2980b9 100%); | |
color: white; | |
padding: 2rem; | |
border-radius: 12px; | |
margin-bottom: 2rem; | |
box-shadow: 0 4px 20px rgba(52, 152, 219, 0.3); | |
} | |
/* Demo user buttons - consistent size and light blue */ | |
.demo-user-btn { | |
background: linear-gradient(135deg, #85c1e9 0%, #5dade2 100%) !important; | |
border: none !important; | |
border-radius: 8px !important; | |
padding: 1rem !important; | |
font-size: 0.95rem !important; | |
font-weight: 600 !important; | |
color: white !important; | |
box-shadow: 0 3px 12px rgba(93, 173, 226, 0.3) !important; | |
transition: all 0.3s ease !important; | |
min-height: 80px !important; | |
text-align: center !important; | |
width: 100% !important; | |
} | |
.demo-user-btn:hover { | |
transform: translateY(-2px) !important; | |
box-shadow: 0 6px 20px rgba(93, 173, 226, 0.4) !important; | |
background: linear-gradient(135deg, #7fb3d3 0%, #5499c7 100%) !important; | |
} | |
/* Dexcom OAuth button - smaller and distinct */ | |
.dexcom-oauth-btn { | |
background: linear-gradient(135deg, #2980b9 0%, #1f618d 100%) !important; | |
border: none !important; | |
border-radius: 8px !important; | |
padding: 0.8rem 1.5rem !important; | |
font-size: 0.9rem !important; | |
font-weight: 600 !important; | |
color: white !important; | |
box-shadow: 0 3px 12px rgba(41, 128, 185, 0.3) !important; | |
transition: all 0.3s ease !important; | |
text-align: center !important; | |
} | |
.dexcom-oauth-btn:hover { | |
transform: translateY(-1px) !important; | |
box-shadow: 0 5px 16px rgba(41, 128, 185, 0.4) !important; | |
} | |
/* Prominent load data button */ | |
.load-data-btn { | |
background: linear-gradient(135deg, #3498db 0%, #2980b9 100%) !important; | |
border: none !important; | |
border-radius: 12px !important; | |
padding: 1.5rem 2rem !important; | |
font-size: 1.1rem !important; | |
font-weight: bold !important; | |
color: white !important; | |
box-shadow: 0 6px 24px rgba(52, 152, 219, 0.4) !important; | |
transition: all 0.3s ease !important; | |
min-height: 80px !important; | |
text-align: center !important; | |
} | |
.load-data-btn:hover { | |
transform: translateY(-2px) !important; | |
box-shadow: 0 8px 32px rgba(52, 152, 219, 0.5) !important; | |
} | |
/* Tab styling - more visible */ | |
.gradio-tabs .tab-nav { | |
background: linear-gradient(135deg, #e3f2fd 0%, #bbdefb 100%) !important; | |
border-radius: 8px !important; | |
padding: 0.5rem !important; | |
margin-bottom: 1rem !important; | |
} | |
.gradio-tabs .tab-nav button { | |
background: white !important; | |
border: 1px solid #90caf9 !important; | |
border-radius: 6px !important; | |
margin: 0 0.25rem !important; | |
padding: 0.75rem 1.5rem !important; | |
font-weight: 600 !important; | |
color: #1565c0 !important; | |
transition: all 0.3s ease !important; | |
} | |
.gradio-tabs .tab-nav button:hover { | |
background: linear-gradient(135deg, #e3f2fd 0%, #bbdefb 100%) !important; | |
transform: translateY(-1px) !important; | |
} | |
.gradio-tabs .tab-nav button.selected { | |
background: linear-gradient(135deg, #3498db 0%, #2980b9 100%) !important; | |
color: white !important; | |
border-color: #2980b9 !important; | |
box-shadow: 0 3px 12px rgba(52, 152, 219, 0.3) !important; | |
} | |
/* Chat bubble styling for demo prompts */ | |
.demo-prompt-bubble { | |
background: linear-gradient(135deg, #e3f2fd 0%, #bbdefb 100%); | |
border: 1px solid #90caf9; | |
border-radius: 15px; | |
padding: 0.75rem 1rem; | |
margin: 0.5rem 0; | |
color: #1565c0; | |
font-size: 0.9rem; | |
cursor: pointer; | |
transition: all 0.2s ease; | |
display: inline-block; | |
max-width: 80%; | |
} | |
.demo-prompt-bubble:hover { | |
background: linear-gradient(135deg, #bbdefb 0%, #90caf9 100%); | |
transform: translateY(-1px); | |
box-shadow: 0 3px 8px rgba(52, 152, 219, 0.2); | |
} | |
/* Toggle styling */ | |
.oauth-toggle { | |
background: #f8f9fa; | |
border: 1px solid #e3f2fd; | |
border-radius: 6px; | |
padding: 0.5rem; | |
} | |
/* Notification styling */ | |
.notification-success { | |
background: white !important; | |
border: 2px solid #27ae60 !important; | |
border-radius: 8px !important; | |
padding: 1rem !important; | |
margin: 1rem 0 !important; | |
box-shadow: 0 4px 12px rgba(39, 174, 96, 0.2) !important; | |
animation: slideIn 0.5s ease-out !important; | |
} | |
.notification-warning { | |
background: white !important; | |
border: 2px solid #f39c12 !important; | |
border-radius: 8px !important; | |
padding: 1rem !important; | |
margin: 1rem 0 !important; | |
box-shadow: 0 4px 12px rgba(243, 156, 18, 0.2) !important; | |
animation: slideIn 0.5s ease-out !important; | |
} | |
.notification-critical { | |
background: white !important; | |
border: 2px solid #e74c3c !important; | |
border-radius: 8px !important; | |
padding: 1rem !important; | |
margin: 1rem 0 !important; | |
box-shadow: 0 4px 12px rgba(231, 76, 60, 0.2) !important; | |
animation: slideIn 0.5s ease-out !important; | |
} | |
@keyframes slideIn { | |
from { | |
opacity: 0; | |
transform: translateY(-20px); | |
} | |
to { | |
opacity: 1; | |
transform: translateY(0); | |
} | |
} | |
/* Group styling */ | |
.user-selection-group { | |
background: #f8f9fa; | |
border: 1px solid #e3f2fd; | |
border-radius: 8px; | |
padding: 1.5rem; | |
margin-bottom: 1rem; | |
} | |
/* Connection status */ | |
.connection-status { | |
background: #e3f2fd; | |
border: 1px solid #bbdefb; | |
border-radius: 6px; | |
padding: 1rem; | |
color: #1565c0; | |
font-weight: 500; | |
} | |
""" | |
with gr.Blocks( | |
title="GlycoAI - AI Glucose Insights", | |
theme=gr.themes.Soft( | |
primary_hue="blue", | |
secondary_hue="blue", | |
neutral_hue="slate" | |
), | |
css=custom_css | |
) as interface: | |
# Clean Header | |
with gr.Row(): | |
with gr.Column(): | |
gr.HTML(""" | |
<div class="main-header"> | |
<div style="display: flex; align-items: center; justify-content: center; gap: 1rem;"> | |
<div style="width: 50px; height: 50px; background: white; border-radius: 50%; display: flex; align-items: center; justify-content: center;"> | |
<span style="color: #3498db; font-size: 20px; font-weight: bold;">π©Ί</span> | |
</div> | |
<div> | |
<h1 style="margin: 0; font-size: 2rem; color: white;">GlycoAI</h1> | |
<p style="margin: 0; font-size: 1rem; color: white; opacity: 0.9;">AI-Powered Glucose Insights</p> | |
</div> | |
</div> | |
<p style="margin-top: 1rem; font-size: 0.9rem; color: white; opacity: 0.8;"> | |
Demo Users + Dexcom Sandbox OAuth β’ Chat with AI for personalized glucose insights | |
</p> | |
</div> | |
""") | |
# User Selection Section - Cleaner Layout | |
with gr.Row(): | |
with gr.Column(): | |
gr.Markdown("### π₯ Choose Your Data Source") | |
# Demo Users Section | |
with gr.Group(): | |
gr.Markdown("#### π Demo Users") | |
gr.Markdown("*Instant access to realistic glucose data for testing*") | |
with gr.Row(): | |
with gr.Column(scale=1): | |
sarah_btn = gr.Button( | |
"Sarah Thompson\nG7 Mobile - β οΈ Unstable Control", | |
elem_classes=["demo-user-btn"] | |
) | |
with gr.Column(scale=1): | |
marcus_btn = gr.Button( | |
"Marcus Rodriguez\nONE+ Mobile - Type 2", | |
elem_classes=["demo-user-btn"] | |
) | |
with gr.Column(scale=1): | |
jennifer_btn = gr.Button( | |
"Jennifer Chen\nG6 Mobile - Athletic", | |
elem_classes=["demo-user-btn"] | |
) | |
with gr.Column(scale=1): | |
robert_btn = gr.Button( | |
"Robert Williams\nG6 Receiver - Experienced", | |
elem_classes=["demo-user-btn"] | |
) | |
# Show/Hide OAuth Toggle | |
with gr.Row(): | |
with gr.Column(scale=4): | |
pass | |
with gr.Column(scale=2): | |
show_oauth_toggle = gr.Checkbox( | |
label="Show Dexcom OAuth Options", | |
value=False, | |
container=False, | |
elem_classes=["oauth-toggle"] | |
) | |
# Dexcom Sandbox OAuth Section (Collapsible) | |
with gr.Group(visible=False) as oauth_section: | |
if DEXCOM_SANDBOX_AVAILABLE: | |
gr.Markdown("#### π Dexcom Sandbox OAuth") | |
gr.Markdown("*Connect with OAuth-authenticated sandbox data*") | |
with gr.Row(): | |
with gr.Column(scale=2): | |
dexcom_sandbox_btn = gr.Button( | |
"π Connect Dexcom Sandbox", | |
elem_classes=["dexcom-oauth-btn"] | |
) | |
with gr.Column(scale=3): | |
oauth_instructions = gr.Markdown( | |
"Click to start Dexcom Sandbox authentication", | |
visible=True | |
) | |
with gr.Row(visible=False) as oauth_completion_row: | |
with gr.Column(): | |
callback_url_input = gr.Textbox( | |
label="Paste Complete Callback URL", | |
placeholder="http://localhost:7860/callback?code=ABC123XYZ&state=sandbox_test", | |
lines=2 | |
) | |
complete_oauth_btn = gr.Button( | |
"β Complete OAuth", | |
elem_classes=["dexcom-oauth-btn"] | |
) | |
else: | |
gr.Markdown("#### π Dexcom Sandbox OAuth") | |
gr.Markdown("*Not configured - demo users available*") | |
gr.Button( | |
"π Dexcom Sandbox Not Available", | |
interactive=False, | |
elem_classes=["dexcom-oauth-btn"] | |
) | |
# Create dummy variables for consistency | |
oauth_instructions = gr.Markdown("", visible=False) | |
callback_url_input = gr.Textbox(visible=False) | |
complete_oauth_btn = gr.Button(visible=False) | |
oauth_completion_row = gr.Row(visible=False) | |
# Connection Status | |
with gr.Row(): | |
with gr.Column(): | |
connection_status = gr.Textbox( | |
label="Connection Status", | |
value="No user selected - Choose a demo user or connect Dexcom Sandbox", | |
interactive=False, | |
elem_classes=["connection-status"] | |
) | |
# Section Divider | |
gr.HTML('<div class="section-divider"></div>') | |
# Update button description for Sarah's unstable patterns | |
with gr.Group(visible=False) as main_interface: | |
# Prominent Load Data Button | |
with gr.Row(): | |
with gr.Column(scale=1): | |
pass # Left spacer | |
with gr.Column(scale=2): | |
load_data_btn = gr.Button( | |
"π Load 14-Day Glucose Data\nπ Start Analysis & Enable AI Chat", | |
elem_classes=["load-data-btn"] | |
) | |
with gr.Column(scale=1): | |
pass # Right spacer | |
# Notification area for data loading feedback | |
with gr.Row(): | |
notification_area = gr.Markdown( | |
"", | |
visible=False, | |
elem_classes=["notification-success"] | |
) | |
# Section Divider | |
gr.HTML('<div class="section-divider"></div>') | |
# Main Content Tabs - Reordered with Chat first | |
with gr.Tabs(): | |
# Chat Tab - FIRST for priority | |
with gr.TabItem("π¬ Chat with AI"): | |
with gr.Column(): | |
gr.Markdown("### π€ Chat with GlycoAI") | |
# Chat Interface with integrated demo prompts | |
chatbot = gr.Chatbot( | |
label="π¬ Chat with GlycoAI", | |
height=450, | |
show_label=False, | |
container=True, | |
bubble_full_width=False, | |
avatar_images=(None, "π©Ί") | |
) | |
# Chat Input | |
with gr.Row(): | |
chat_input = gr.Textbox( | |
placeholder="Ask about your glucose patterns, trends, or management strategies...", | |
label="Your Question", | |
lines=2, | |
scale=4 | |
) | |
send_btn = gr.Button( | |
"Send", | |
variant="primary", | |
scale=1 | |
) | |
# Chat Controls | |
with gr.Row(): | |
clear_chat_btn = gr.Button( | |
"ποΈ Clear Chat", | |
size="sm" | |
) | |
gr.Markdown("*AI responses are for informational purposes only. Always consult your healthcare provider.*") | |
# Data Overview Tab - SECOND | |
with gr.TabItem("π Data Overview"): | |
with gr.Column(): | |
gr.Markdown("### π Comprehensive Data Analysis") | |
data_display = gr.Markdown( | |
"Load your glucose data to see detailed statistics and insights", | |
container=True | |
) | |
# Glucose Chart Tab - THIRD | |
with gr.TabItem("π Glucose Chart"): | |
with gr.Column(): | |
gr.Markdown("### π Interactive 14-Day Glucose Analysis") | |
glucose_chart = gr.Plot( | |
label="Interactive Glucose Trends", | |
container=True | |
) | |
# Event Handlers | |
def handle_demo_user_selection(user_key): | |
status, interface_visibility = app.select_demo_user(user_key) | |
initial_chat = app.initialize_chat_with_prompts() | |
return status, interface_visibility, initial_chat | |
def handle_load_data(): | |
overview, chart, notification = app.load_glucose_data() | |
# Determine notification class based on content | |
if "CONCERNING PATTERNS" in notification or "CRITICAL" in notification: | |
notification_class = "notification-critical" | |
elif "EXCELLENT CONTROL" in notification: | |
notification_class = "notification-success" | |
elif notification: | |
notification_class = "notification-warning" | |
else: | |
notification_class = "notification-success" | |
# Show notification with appropriate styling | |
notification_update = gr.update( | |
value=notification, | |
visible=bool(notification), | |
elem_classes=[notification_class] | |
) | |
return overview, chart, notification_update | |
def handle_chat_submit(message, history): | |
return app.chat_with_mistral(message, history) | |
def handle_enter_key(message, history): | |
if message.strip(): | |
return app.chat_with_mistral(message, history) | |
return "", history | |
def handle_chatbot_click(history, evt: gr.SelectData): | |
"""Handle clicking on chat bubbles (demo prompts)""" | |
if evt.index is not None and len(history) > evt.index[0]: | |
clicked_message = history[evt.index[0]][1] # Get AI message | |
# Check if it's a demo prompt (contains ** formatting) | |
if "**" in clicked_message and ("π―" in clicked_message or "β‘" in clicked_message or "π½οΈ" in clicked_message): | |
return app.handle_demo_prompt_click(clicked_message, history) | |
return "", history | |
# Toggle OAuth section visibility | |
show_oauth_toggle.change( | |
lambda show: gr.update(visible=show), | |
inputs=[show_oauth_toggle], | |
outputs=[oauth_section] | |
) | |
# Connect Event Handlers for Demo Users | |
sarah_btn.click( | |
lambda: handle_demo_user_selection("sarah_g7"), | |
outputs=[connection_status, main_interface, chatbot] | |
) | |
marcus_btn.click( | |
lambda: handle_demo_user_selection("marcus_one"), | |
outputs=[connection_status, main_interface, chatbot] | |
) | |
jennifer_btn.click( | |
lambda: handle_demo_user_selection("jennifer_g6"), | |
outputs=[connection_status, main_interface, chatbot] | |
) | |
robert_btn.click( | |
lambda: handle_demo_user_selection("robert_receiver"), | |
outputs=[connection_status, main_interface, chatbot] | |
) | |
# Connect Event Handlers for Dexcom Sandbox OAuth | |
if DEXCOM_SANDBOX_AVAILABLE: | |
dexcom_sandbox_btn.click( | |
app.start_dexcom_sandbox_oauth, | |
outputs=[oauth_instructions] | |
).then( | |
lambda: gr.update(visible=True), | |
outputs=[oauth_completion_row] | |
) | |
complete_oauth_btn.click( | |
app.complete_dexcom_sandbox_oauth, | |
inputs=[callback_url_input], | |
outputs=[connection_status, main_interface] | |
).then( | |
app.initialize_chat_with_prompts, # Initialize chat with prompts after OAuth | |
outputs=[chatbot] | |
) | |
# Data Loading | |
load_data_btn.click( | |
handle_load_data, | |
outputs=[data_display, glucose_chart, notification_area] | |
) | |
# Chat Handlers | |
send_btn.click( | |
handle_chat_submit, | |
inputs=[chat_input, chatbot], | |
outputs=[chat_input, chatbot] | |
) | |
chat_input.submit( | |
handle_enter_key, | |
inputs=[chat_input, chatbot], | |
outputs=[chat_input, chatbot] | |
) | |
# Handle clicking on chat bubbles (demo prompts) | |
chatbot.select( | |
handle_chatbot_click, | |
inputs=[chatbot], | |
outputs=[chat_input, chatbot] | |
) | |
# Clear Chat | |
clear_chat_btn.click( | |
app.clear_chat_history, | |
outputs=[chatbot] | |
) | |
# Clean Footer | |
with gr.Row(): | |
gr.HTML(f""" | |
<div style="text-align: center; padding: 1.5rem; margin-top: 2rem; border-top: 1px solid #e3f2fd; color: #546e7a;"> | |
<p><strong>β οΈ Medical Disclaimer</strong></p> | |
<p style="font-size: 0.9rem;">GlycoAI is for informational and educational purposes only. Always consult your healthcare provider | |
before making any changes to your diabetes management plan.</p> | |
<p style="margin-top: 1rem; font-size: 0.85rem; color: #78909c;"> | |
π Data processed securely β’ π‘ Powered by Dexcom API & Mistral AI<br> | |
π Demo: Available β’ π Dexcom Sandbox: {"Available" if DEXCOM_SANDBOX_AVAILABLE else "Not configured"} | |
</p> | |
</div> | |
""") | |
return interface | |
def main(): | |
"""Main function to launch the application""" | |
print("π Starting GlycoAI - AI-Powered Glucose Insights...") | |
# Check OAuth availability | |
oauth_status = "β Available" if DEXCOM_SANDBOX_AVAILABLE else "β Not configured" | |
print(f"π― Dexcom Sandbox OAuth: {oauth_status}") | |
# Validate environment before starting | |
print("π Validating environment configuration...") | |
if not validate_environment(): | |
print("β Environment validation failed!") | |
print("Please check your .env file or environment variables.") | |
return | |
print("β Environment validation passed!") | |
try: | |
# Create and launch the interface | |
demo = create_interface() | |
print("π― GlycoAI Features:") | |
print("π Clean UI with blue theme, consistent button sizes, improved readability") | |
print("π Demo users: 4 realistic profiles for instant testing") | |
if DEXCOM_SANDBOX_AVAILABLE: | |
print("β Dexcom Sandbox: Available - OAuth authentication ready") | |
else: | |
print("π Dexcom Sandbox: Not configured - demo users only") | |
# Launch with custom settings | |
demo.launch( | |
server_name="0.0.0.0", # Allow external access | |
server_port=7860, # Your port | |
share=True, # Set to True for public sharing (tunneling) | |
debug=os.getenv("DEBUG", "false").lower() == "true", | |
show_error=True, # Show errors in the interface | |
auth=None, # No authentication required | |
favicon_path=None, # Use default favicon | |
ssl_verify=False # Disable SSL verification for development | |
) | |
except Exception as e: | |
logger.error(f"Failed to launch GlycoAI application: {e}") | |
print(f"β Error launching application: {e}") | |
# Provide helpful error information | |
if "environment" in str(e).lower(): | |
print("\nπ‘ Environment troubleshooting:") | |
print("1. Check if .env file exists with MISTRAL_API_KEY") | |
print("2. Verify your API key is valid") | |
print("3. For Hugging Face Spaces, check Repository secrets") | |
else: | |
print("\nπ‘ Try checking:") | |
print("1. All dependencies are installed: pip install -r requirements.txt") | |
print("2. Port 7860 is available") | |
print("3. Check the logs above for specific error details") | |
raise | |
if __name__ == "__main__": | |
# Setup logging configuration | |
log_level = os.getenv("LOG_LEVEL", "INFO") | |
logging.basicConfig( | |
level=getattr(logging, log_level.upper()), | |
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s', | |
handlers=[ | |
logging.FileHandler('glycoai.log'), | |
logging.StreamHandler() | |
] | |
) | |
# Run the main application | |
main() |