Spaces:
Runtime error
Runtime error
| """ | |
| Health Context - Unified access to all user health data | |
| Central hub for agents to access and update health information | |
| """ | |
| from typing import List, Optional, Dict, Any | |
| from datetime import datetime | |
| from .data_store import HealthDataStore | |
| from .models import ( | |
| UserHealthProfile, HealthRecord, UserPreferences, | |
| FitnessProgress, HealthMetrics | |
| ) | |
| class HealthContext: | |
| """ | |
| Unified access to all user health data | |
| Provides a single interface for agents to access and update health information | |
| """ | |
| def __init__(self, user_id: str, data_store: Optional[HealthDataStore] = None): | |
| self.user_id = user_id | |
| self.data_store = data_store or HealthDataStore() | |
| # Load data from storage | |
| self.profile = self.data_store.get_user_profile(user_id) | |
| if not self.profile: | |
| self.profile = UserHealthProfile(user_id) | |
| self.data_store.save_user_profile(self.profile) | |
| self.preferences = self.data_store.get_preferences(user_id) | |
| if not self.preferences: | |
| self.preferences = UserPreferences(user_id) | |
| self.data_store.save_preferences(self.preferences) | |
| # ===== Profile Access ===== | |
| def get_user_profile(self) -> UserHealthProfile: | |
| """Get user's health profile""" | |
| return self.profile | |
| def update_profile(self, **kwargs) -> None: | |
| """Update user profile fields""" | |
| for key, value in kwargs.items(): | |
| if hasattr(self.profile, key): | |
| setattr(self.profile, key, value) | |
| self.profile.updated_at = datetime.now() | |
| self.data_store.save_user_profile(self.profile) | |
| def get_profile_dict(self) -> Dict[str, Any]: | |
| """Get profile as dictionary""" | |
| return self.profile.to_dict() | |
| # ===== Health History Access ===== | |
| def get_health_history(self, days: int = 30) -> List[HealthRecord]: | |
| """Get health history for last N days""" | |
| return self.data_store.get_health_history(self.user_id, days) | |
| def get_records_by_type(self, record_type: str) -> List[HealthRecord]: | |
| """Get records of specific type""" | |
| return self.data_store.get_records_by_type(self.user_id, record_type) | |
| def add_health_record(self, record_type: str, data: Dict[str, Any], | |
| agent_name: Optional[str] = None, | |
| confidence: float = 0.5) -> None: | |
| """Add health record""" | |
| record = HealthRecord(self.user_id, record_type, data) | |
| record.agent_name = agent_name | |
| record.confidence = confidence | |
| self.data_store.add_health_record(record) | |
| # ===== Preferences Access ===== | |
| def get_preferences(self) -> UserPreferences: | |
| """Get user preferences""" | |
| return self.preferences | |
| def update_preferences(self, **kwargs) -> None: | |
| """Update preferences""" | |
| for key, value in kwargs.items(): | |
| if hasattr(self.preferences, key): | |
| setattr(self.preferences, key, value) | |
| self.preferences.updated_at = datetime.now() | |
| self.data_store.save_preferences(self.preferences) | |
| def add_goal(self, goal: str) -> None: | |
| """Add health goal""" | |
| if goal not in self.preferences.goals: | |
| self.preferences.goals.append(goal) | |
| self.data_store.save_preferences(self.preferences) | |
| def add_exercise_preference(self, exercise_type: str) -> None: | |
| """Add exercise preference""" | |
| if exercise_type not in self.preferences.preferred_exercise_types: | |
| self.preferences.preferred_exercise_types.append(exercise_type) | |
| self.data_store.save_preferences(self.preferences) | |
| # ===== Fitness Access ===== | |
| def get_fitness_history(self, days: int = 30) -> List[FitnessProgress]: | |
| """Get fitness history""" | |
| return self.data_store.get_fitness_history(self.user_id, days) | |
| def add_fitness_record(self, workout_data: Dict[str, Any]) -> None: | |
| """Add fitness record""" | |
| record = FitnessProgress(self.user_id) | |
| for key, value in workout_data.items(): | |
| if hasattr(record, key): | |
| setattr(record, key, value) | |
| self.data_store.add_fitness_record(record) | |
| def get_workout_adherence(self, days: int = 30) -> float: | |
| """Calculate workout adherence rate (0-1)""" | |
| history = self.get_fitness_history(days) | |
| if not history: | |
| return 0.0 | |
| # Simple adherence: workouts completed / expected workouts | |
| # Assuming 3-4 workouts per week is ideal | |
| expected_workouts = (days / 7) * 3.5 | |
| actual_workouts = len(history) | |
| adherence = min(actual_workouts / expected_workouts, 1.0) if expected_workouts > 0 else 0.0 | |
| return round(adherence, 2) | |
| # ===== Metrics Access ===== | |
| def get_metrics(self, metric_type: Optional[str] = None) -> List[HealthMetrics]: | |
| """Get health metrics""" | |
| return self.data_store.get_metrics(self.user_id, metric_type) | |
| def add_metric(self, metric_type: str, value: float, unit: str) -> None: | |
| """Add health metric""" | |
| metric = HealthMetrics(self.user_id, metric_type, value, unit) | |
| self.data_store.add_metric(metric) | |
| def get_latest_metric(self, metric_type: str) -> Optional[HealthMetrics]: | |
| """Get latest metric of specific type""" | |
| metrics = self.get_metrics(metric_type) | |
| return metrics[0] if metrics else None | |
| # ===== Context Summary ===== | |
| def get_context_summary(self) -> str: | |
| """Get summary of current context for agents""" | |
| summary_parts = [] | |
| # Profile summary | |
| profile = self.profile | |
| if profile.age: | |
| summary_parts.append(f"Age: {profile.age}") | |
| if profile.gender: | |
| summary_parts.append(f"Gender: {profile.gender}") | |
| if profile.weight and profile.height: | |
| summary_parts.append(f"Weight: {profile.weight}kg, Height: {profile.height}cm") | |
| if profile.bmi: | |
| summary_parts.append(f"BMI: {profile.bmi}") | |
| # Health conditions | |
| if profile.health_conditions: | |
| summary_parts.append(f"Conditions: {', '.join(profile.health_conditions)}") | |
| # Goals | |
| if self.preferences.goals: | |
| summary_parts.append(f"Goals: {', '.join(self.preferences.goals)}") | |
| # Recent activity | |
| recent_records = self.get_health_history(days=7) | |
| if recent_records: | |
| summary_parts.append(f"Recent interactions: {len(recent_records)}") | |
| return " | ".join(summary_parts) if summary_parts else "No context yet" | |
| def get_personalization_context(self) -> str: | |
| """Get context for personalization""" | |
| context = [] | |
| # User profile | |
| profile = self.profile | |
| context.append(f"User Profile: {profile.age}yo, {profile.gender}, BMI {profile.bmi}") | |
| # Health conditions | |
| if profile.health_conditions: | |
| context.append(f"Health Conditions: {', '.join(profile.health_conditions)}") | |
| # Goals | |
| if self.preferences.goals: | |
| context.append(f"Goals: {', '.join(self.preferences.goals)}") | |
| # Recent interactions | |
| recent = self.get_health_history(days=7) | |
| if recent: | |
| context.append(f"Recent interactions: {len(recent)} in last 7 days") | |
| # Fitness adherence | |
| adherence = self.get_workout_adherence(days=30) | |
| context.append(f"Fitness adherence (30d): {adherence*100:.0f}%") | |
| return "\n".join(context) | |
| # ===== Data Export & Management ===== | |
| def export_data(self) -> Dict[str, Any]: | |
| """Export all user data""" | |
| return self.data_store.export_user_data(self.user_id) | |
| def delete_all_data(self) -> None: | |
| """Delete all user data (GDPR compliance)""" | |
| self.data_store.delete_user_data(self.user_id) | |
| def __repr__(self) -> str: | |
| return f"<HealthContext: {self.user_id} - {self.get_context_summary()}>" | |