""" Health Data Provider - Abstract interface for health data sources. This module provides an abstraction layer that allows switching between different health data backends (Google Fit, Health Connect, etc.) without changing the API layer. """ from abc import ABC, abstractmethod from typing import List, Dict, Any, Optional from datetime import datetime from enum import Enum class DataSource(str, Enum): """Enum for tracking data provenance.""" HEALTH_CONNECT = "health_connect" MANUAL = "manual" class HealthDataProvider(ABC): """ Abstract base class for health data providers. Implementations must provide methods to fetch daily metrics, workout sessions, and sleep data from their respective sources. """ @property @abstractmethod def source(self) -> DataSource: """Return the data source identifier.""" pass @abstractmethod async def fetch_daily_metrics( self, access_token: Optional[str], days: int = 7, timezone: str = "UTC", user_id: Optional[int] = None ) -> List[Dict[str, Any]]: """ Fetch aggregated daily health metrics. Args: access_token: OAuth token or API key days: Number of days to fetch timezone: User's timezone for day boundaries user_id: Optional user ID for logging Returns: List of daily metric dictionaries with keys: - date: date object - step_count: int - calories_burned: float - distance_meters: float - sleep_duration_minutes: int - deep_sleep_minutes: int - light_sleep_minutes: int - rem_sleep_minutes: int - awake_duration_minutes: int - avg_heart_rate: int - min_heart_rate: int - max_heart_rate: int - avg_spo2: float - pai_score: float """ pass @abstractmethod async def fetch_workouts( self, access_token: Optional[str], days: int = 30 ) -> List[Dict[str, Any]]: """ Fetch workout sessions. Args: access_token: OAuth token or API key days: Number of days to fetch Returns: List of workout dictionaries with keys: - start_time: datetime - end_time: datetime - activity_type: str (human readable) - activity_type_id: int (source-specific ID) - duration_seconds: int - distance_meters: float - calories: float - avg_hr: int - max_hr: int - min_hr: int """ pass @abstractmethod async def fetch_sleep_sessions( self, access_token: Optional[str], days: int = 30 ) -> List[Dict[str, Any]]: """ Fetch sleep sessions for total duration validation. Args: access_token: OAuth token or API key days: Number of days to fetch Returns: List of sleep session dictionaries with keys: - date: date object (wake-up date) - duration_minutes: int - start_time: datetime - end_time: datetime - type: str """ pass @abstractmethod def get_auth_url(self) -> str: """ Get the OAuth authorization URL for this provider. Returns: Authorization URL string """ pass class HealthProviderFactory: """ Factory for creating health data provider instances. Uses the HEALTH_PROVIDER environment variable to determine which provider to instantiate. """ _providers: Dict[str, type] = {} @classmethod def register(cls, name: str, provider_class: type): """Register a provider class with a name.""" cls._providers[name] = provider_class @classmethod def get_provider(cls, name: Optional[str] = None) -> HealthDataProvider: """ Get a provider instance by name. Args: name: Provider name. If None, uses HEALTH_PROVIDER env var. Returns: HealthDataProvider instance Raises: ValueError: If provider name is not registered """ from backend.core.config import settings provider_name = name or getattr(settings, 'HEALTH_PROVIDER', 'health_connect') if provider_name not in cls._providers: available = list(cls._providers.keys()) raise ValueError( f"Unknown provider '{provider_name}'. Available: {available}" ) return cls._providers[provider_name]() @classmethod def list_providers(cls) -> List[str]: """List all registered provider names.""" return list(cls._providers.keys())