Delete unified_data_manager.py

unified_data_manager.py

@@ -1,413 +0,0 @@
-"""
-Unified Data Manager for GlycoAI - FIXED VERSION
-Restores the original working API calls that were working before
-"""
-
-import logging
-from typing import Dict, Any, Optional, Tuple
-import pandas as pd
-from datetime import datetime, timedelta
-from dataclasses import asdict
-
-from apifunctions import (
-    DexcomAPI,
-    GlucoseAnalyzer,
-    DEMO_USERS,
-    DemoUser
-)
-
-logger = logging.getLogger(__name__)
-
-class UnifiedDataManager:
-    """
-    FIXED: Unified data manager that calls the API exactly as it was working before
-    """
-    
-    def __init__(self):
-        self.dexcom_api = DexcomAPI()
-        self.analyzer = GlucoseAnalyzer()
-        
-        logger.info(f"UnifiedDataManager initialized - RESTORED to working version")
-        
-        # Single source of truth for all data
-        self.current_user: Optional[DemoUser] = None
-        self.raw_glucose_data: Optional[list] = None
-        self.processed_glucose_data: Optional[pd.DataFrame] = None
-        self.calculated_stats: Optional[Dict] = None
-        self.identified_patterns: Optional[Dict] = None
-        
-        # Metadata
-        self.data_loaded_at: Optional[datetime] = None
-        self.data_source: str = "none"  # "dexcom_api", "mock", or "none"
-        
-    def load_user_data(self, user_key: str, force_reload: bool = False) -> Dict[str, Any]:
-        """
-        FIXED: Load glucose data using the ORIGINAL WORKING method
-        """
-        
-        # Check if we already have data for this user and it's recent
-        if (not force_reload and 
-            self.current_user and 
-            self.current_user == DEMO_USERS.get(user_key) and
-            self.data_loaded_at and 
-            (datetime.now() - self.data_loaded_at).seconds < 300):  # 5 minutes cache
-            
-            logger.info(f"Using cached data for {user_key}")
-            return self._build_success_response()
-        
-        try:
-            if user_key not in DEMO_USERS:
-                return {
-                    "success": False,
-                    "message": f"❌ Invalid user key '{user_key}'. Available: {', '.join(DEMO_USERS.keys())}"
-                }
-            
-            logger.info(f"Loading data for user: {user_key}")
-            
-            # Set current user
-            self.current_user = DEMO_USERS[user_key]
-            
-            # Call API EXACTLY as it was working before
-            try:
-                logger.info(f"Attempting Dexcom API authentication for {user_key}")
-                
-                # ORIGINAL WORKING METHOD: Use the simulate_demo_login exactly as before
-                access_token = self.dexcom_api.simulate_demo_login(user_key)
-                logger.info(f"Dexcom authentication result: {bool(access_token)}")
-                
-                if access_token:
-                    # ORIGINAL WORKING METHOD: Get data with 14-day range
-                    end_date = datetime.now()
-                    start_date = end_date - timedelta(days=14)
-                    
-                    # Call get_egv_data EXACTLY as it was working before
-                    self.raw_glucose_data = self.dexcom_api.get_egv_data(
-                        start_date.isoformat(),
-                        end_date.isoformat()
-                    )
-                    
-                    if self.raw_glucose_data and len(self.raw_glucose_data) > 0:
-                        self.data_source = "dexcom_api"
-                        logger.info(f"✅ Successfully loaded {len(self.raw_glucose_data)} readings from Dexcom API")
-                    else:
-                        logger.warning("Dexcom API returned empty data - falling back to mock data")
-                        raise Exception("Empty data from Dexcom API")
-                else:
-                    logger.warning("Failed to get access token - falling back to mock data")
-                    raise Exception("Authentication failed")
-                    
-            except Exception as api_error:
-                logger.warning(f"Dexcom API failed ({str(api_error)}) - using mock data fallback")
-                self.raw_glucose_data = self._generate_realistic_mock_data(user_key)
-                self.data_source = "mock"
-            
-            # Process the raw data (same processing for everyone)
-            self.processed_glucose_data = self.analyzer.process_egv_data(self.raw_glucose_data)
-            
-            if self.processed_glucose_data is None or self.processed_glucose_data.empty:
-                return {
-                    "success": False,
-                    "message": "❌ Failed to process glucose data"
-                }
-            
-            # Calculate statistics (single source of truth)
-            self.calculated_stats = self._calculate_unified_stats()
-            
-            # Identify patterns
-            self.identified_patterns = self.analyzer.identify_patterns(self.processed_glucose_data)
-            
-            # Mark when data was loaded
-            self.data_loaded_at = datetime.now()
-            
-            logger.info(f"Successfully loaded and processed data for {self.current_user.name}")
-            logger.info(f"Data source: {self.data_source}, Readings: {len(self.processed_glucose_data)}")
-            logger.info(f"TIR: {self.calculated_stats.get('time_in_range_70_180', 0):.1f}%")
-            
-            return self._build_success_response()
-            
-        except Exception as e:
-            logger.error(f"Failed to load user data: {e}")
-            return {
-                "success": False,
-                "message": f"❌ Failed to load user data: {str(e)}"
-            }
-    
-    def get_stats_for_ui(self) -> Dict[str, Any]:
-        """Get statistics formatted for the UI display"""
-        if not self.calculated_stats:
-            return {}
-        
-        return {
-            **self.calculated_stats,
-            "data_source": self.data_source,
-            "loaded_at": self.data_loaded_at.isoformat() if self.data_loaded_at else None,
-            "user_name": self.current_user.name if self.current_user else None
-        }
-    
-    def get_context_for_agent(self) -> Dict[str, Any]:
-        """Get context formatted for the AI agent"""
-        if not self.current_user or not self.calculated_stats:
-            return {"error": "No user data loaded"}
-        
-        # Build agent context with the SAME data as UI
-        context = {
-            "user": {
-                "name": self.current_user.name,
-                "age": self.current_user.age,
-                "diabetes_type": self.current_user.diabetes_type,
-                "device_type": self.current_user.device_type,
-                "years_with_diabetes": self.current_user.years_with_diabetes,
-                "typical_pattern": getattr(self.current_user, 'typical_glucose_pattern', 'normal')
-            },
-            "statistics": self._safe_convert_for_json(self.calculated_stats),
-            "patterns": self._safe_convert_for_json(self.identified_patterns),
-            "data_points": len(self.processed_glucose_data) if self.processed_glucose_data is not None else 0,
-            "recent_readings": self._get_recent_readings_for_agent(),
-            "data_metadata": {
-                "source": self.data_source,
-                "loaded_at": self.data_loaded_at.isoformat() if self.data_loaded_at else None,
-                "data_age_minutes": int((datetime.now() - self.data_loaded_at).total_seconds() / 60) if self.data_loaded_at else None
-            }
-        }
-        
-        return context
-    
-    def get_chart_data(self) -> Optional[pd.DataFrame]:
-        """Get processed data for chart display"""
-        return self.processed_glucose_data
-    
-    def _calculate_unified_stats(self) -> Dict[str, Any]:
-        """Calculate statistics using a single, consistent method"""
-        if self.processed_glucose_data is None or self.processed_glucose_data.empty:
-            return {"error": "No data available"}
-        
-        try:
-            # Get glucose values
-            glucose_values = self.processed_glucose_data['value'].dropna()
-            
-            if len(glucose_values) == 0:
-                return {"error": "No valid glucose values"}
-            
-            # Convert to numpy array for consistent calculations
-            import numpy as np
-            values = np.array(glucose_values.tolist(), dtype=float)
-            
-            # Calculate basic statistics
-            avg_glucose = float(np.mean(values))
-            min_glucose = float(np.min(values))
-            max_glucose = float(np.max(values))
-            std_glucose = float(np.std(values))
-            total_readings = int(len(values))
-            
-            # Calculate time in ranges - CONSISTENT METHOD
-            in_range_mask = (values >= 70) & (values <= 180)
-            below_range_mask = values < 70
-            above_range_mask = values > 180
-            
-            in_range_count = int(np.sum(in_range_mask))
-            below_range_count = int(np.sum(below_range_mask))
-            above_range_count = int(np.sum(above_range_mask))
-            
-            # Calculate percentages
-            time_in_range = (in_range_count / total_readings) * 100 if total_readings > 0 else 0
-            time_below_70 = (below_range_count / total_readings) * 100 if total_readings > 0 else 0
-            time_above_180 = (above_range_count / total_readings) * 100 if total_readings > 0 else 0
-            
-            # Calculate additional metrics
-            gmi = 3.31 + (0.02392 * avg_glucose)  # Glucose Management Indicator
-            cv = (std_glucose / avg_glucose) * 100 if avg_glucose > 0 else 0  # Coefficient of Variation
-            
-            stats = {
-                "average_glucose": avg_glucose,
-                "min_glucose": min_glucose,
-                "max_glucose": max_glucose,
-                "std_glucose": std_glucose,
-                "time_in_range_70_180": time_in_range,
-                "time_below_70": time_below_70,
-                "time_above_180": time_above_180,
-                "total_readings": total_readings,
-                "gmi": gmi,
-                "cv": cv,
-                "in_range_count": in_range_count,
-                "below_range_count": below_range_count,
-                "above_range_count": above_range_count
-            }
-            
-            # Log for debugging
-            logger.info(f"Calculated stats - TIR: {time_in_range:.1f}%, Total: {total_readings}, In range: {in_range_count}")
-            
-            return stats
-            
-        except Exception as e:
-            logger.error(f"Error calculating unified stats: {e}")
-            return {"error": f"Statistics calculation failed: {str(e)}"}
-    
-    def _generate_realistic_mock_data(self, user_key: str) -> list:
-        """Generate consistent mock data for demo users"""
-        from mistral_chat import GlucoseDataGenerator
-        
-        # Map users to patterns
-        pattern_map = {
-            "sarah_g7": "normal",
-            "marcus_one": "dawn_phenomenon", 
-            "jennifer_g6": "normal",
-            "robert_receiver": "dawn_phenomenon"
-        }
-        
-        user_pattern = pattern_map.get(user_key, "normal")
-        
-        # Generate 14 days of data
-        mock_data = GlucoseDataGenerator.create_realistic_pattern(days=14, user_type=user_pattern)
-        
-        logger.info(f"Generated {len(mock_data)} mock data points for {user_key} with pattern {user_pattern}")
-        
-        return mock_data
-    
-    def _get_recent_readings_for_agent(self, count: int = 5) -> list:
-        """Get recent readings formatted for agent context"""
-        if self.processed_glucose_data is None or self.processed_glucose_data.empty:
-            return []
-        
-        try:
-            recent_df = self.processed_glucose_data.tail(count)
-            readings = []
-            
-            for _, row in recent_df.iterrows():
-                display_time = row.get('displayTime') or row.get('systemTime')
-                glucose_value = row.get('value')
-                trend_value = row.get('trend', 'flat')
-                
-                if pd.notna(display_time):
-                    if isinstance(display_time, str):
-                        time_str = display_time
-                    else:
-                        time_str = pd.to_datetime(display_time).isoformat()
-                else:
-                    time_str = datetime.now().isoformat()
-                
-                if pd.notna(glucose_value):
-                    glucose_clean = self._safe_convert_for_json(glucose_value)
-                else:
-                    glucose_clean = None
-                
-                trend_clean = str(trend_value) if pd.notna(trend_value) else 'flat'
-                
-                readings.append({
-                    "time": time_str,
-                    "glucose": glucose_clean,
-                    "trend": trend_clean
-                })
-            
-            return readings
-            
-        except Exception as e:
-            logger.error(f"Error getting recent readings: {e}")
-            return []
-    
-    def _safe_convert_for_json(self, obj):
-        """Safely convert objects for JSON serialization"""
-        import numpy as np
-        
-        if obj is None:
-            return None
-        elif isinstance(obj, (np.integer, np.int64, np.int32)):
-            return int(obj)
-        elif isinstance(obj, (np.floating, np.float64, np.float32)):
-            if np.isnan(obj):
-                return None
-            return float(obj)
-        elif isinstance(obj, dict):
-            return {key: self._safe_convert_for_json(value) for key, value in obj.items()}
-        elif isinstance(obj, list):
-            return [self._safe_convert_for_json(item) for item in obj]
-        elif isinstance(obj, pd.Timestamp):
-            return obj.isoformat()
-        else:
-            return obj
-    
-    def _build_success_response(self) -> Dict[str, Any]:
-        """Build a consistent success response"""
-        data_points = len(self.processed_glucose_data) if self.processed_glucose_data is not None else 0
-        avg_glucose = self.calculated_stats.get('average_glucose', 0)
-        time_in_range = self.calculated_stats.get('time_in_range_70_180', 0)
-        
-        return {
-            "success": True,
-            "message": f"✅ Successfully loaded data for {self.current_user.name}",
-            "user": asdict(self.current_user),
-            "data_points": data_points,
-            "stats": self.calculated_stats,
-            "data_source": self.data_source,
-            "summary": f"📊 {data_points} readings | Avg: {avg_glucose:.1f} mg/dL | TIR: {time_in_range:.1f}% | Source: {self.data_source}"
-        }
-    
-    def validate_data_consistency(self) -> Dict[str, Any]:
-        """Validate that all components are using consistent data"""
-        if not self.calculated_stats:
-            return {"valid": False, "message": "No data loaded"}
-        
-        validation = {
-            "valid": True,
-            "data_source": self.data_source,
-            "data_age_minutes": int((datetime.now() - self.data_loaded_at).total_seconds() / 60) if self.data_loaded_at else None,
-            "total_readings": self.calculated_stats.get('total_readings', 0),
-            "time_in_range": self.calculated_stats.get('time_in_range_70_180', 0),
-            "average_glucose": self.calculated_stats.get('average_glucose', 0),
-            "user": self.current_user.name if self.current_user else None
-        }
-        
-        logger.info(f"Data consistency check: {validation}")
-        
-        return validation
-
-# ADDITIONAL: Debug function to test the API connection as it was working before
-def test_original_api_method():
-    """Test the API exactly as it was working before unified data manager"""
-    from apifunctions import DexcomAPI, DEMO_USERS
-    
-    print("🔍 Testing API exactly as it was working before...")
-    
-    api = DexcomAPI()
-    
-    # Test with sarah_g7 as it was working before
-    user_key = "sarah_g7"
-    user = DEMO_USERS[user_key]
-    
-    print(f"Testing with {user.name} ({user.username})")
-    
-    try:
-        # Call simulate_demo_login exactly as before
-        access_token = api.simulate_demo_login(user_key)
-        print(f"✅ Authentication: {bool(access_token)}")
-        
-        if access_token:
-            # Call get_egv_data exactly as before
-            end_date = datetime.now()
-            start_date = end_date - timedelta(days=14)
-            
-            egv_data = api.get_egv_data(
-                start_date.isoformat(),
-                end_date.isoformat()
-            )
-            
-            print(f"✅ EGV Data: {len(egv_data)} readings")
-            
-            if egv_data:
-                print(f"✅ SUCCESS! API is working as before")
-                sample = egv_data[0] if egv_data else {}
-                print(f"Sample reading: {sample}")
-                return True
-            else:
-                print("⚠️  API authenticated but returned no data")
-                return False
-        else:
-            print("❌ Authentication failed")
-            return False
-            
-    except Exception as e:
-        print(f"❌ Error: {e}")
-        return False
-
-if __name__ == "__main__":
-    # Test the original API method
-    test_original_api_method()