""" Query service for health data analysis. Provides methods to query and analyze health data from storage. """ import json from datetime import date, timedelta from pathlib import Path from typing import Dict, List, Optional, Any, Union from health import config from health.db.repository import HealthRepository from health.services.storage import HealthStorage from health.utils.logging_config import setup_logger logger = setup_logger(__name__) class HealthDataQuery: """Service for querying and analyzing health data.""" def __init__( self, storage: Optional[HealthStorage] = None, repo: Optional[HealthRepository] = None ) -> None: """Initialize query service. Args: storage: Optional storage service instance repo: Optional repository instance """ self.storage = storage or HealthStorage() self.repo = repo or HealthRepository() def get_daily_summary(self, target_date: date) -> Dict[str, Any]: """Get a summary of all health metrics for a specific day. Args: target_date: Date to query Returns: Dictionary containing all available metrics for the day """ logger.info(f"Querying health data summary for {target_date}") summary = { "date": target_date.isoformat(), "metrics": {}, "activities": [], } # Query all daily metrics for metric_type in config.DATA_TYPE_CONFIG.keys(): # Skip activities (handled separately) and weight (body metric) if metric_type in ["activities", "weight"]: continue data = self.storage.load_daily_metric(metric_type, target_date) if data: summary["metrics"][metric_type] = data # Query activities for the day activities = self.repo.get_activities_by_date_range(target_date, target_date) for activity in activities: activity_data = self.storage.load_activity(activity["activity_id"]) if activity_data: summary["activities"].append(activity_data) # Query weight (if available) weight_data = self.storage.load_daily_metric("weight", target_date) if weight_data: summary["metrics"]["weight"] = weight_data logger.info( f"Found {len(summary['metrics'])} metrics and {len(summary['activities'])} activities" ) return summary def get_metric_range( self, metric_type: str, start_date: date, end_date: date ) -> List[Dict[str, Any]]: """Get a specific metric's data over a date range. Args: metric_type: Type of metric (e.g., "steps", "sleep", "heart_rate") start_date: Start date of range end_date: End date of range Returns: List of metric data dictionaries, ordered by date Raises: ValueError: If metric_type is invalid """ if metric_type not in config.DATA_TYPE_CONFIG: valid_types = ", ".join(config.DATA_TYPE_CONFIG.keys()) raise ValueError( f"Invalid metric type: {metric_type}. Valid types: {valid_types}" ) logger.info(f"Querying {metric_type} from {start_date} to {end_date}") results = [] current_date = start_date while current_date <= end_date: data = self.storage.load_daily_metric(metric_type, current_date) if data: results.append(data) current_date += timedelta(days=1) logger.info(f"Found {len(results)} records for {metric_type}") return results def get_activities_range( self, start_date: date, end_date: date, activity_type: Optional[str] = None ) -> List[Dict[str, Any]]: """Get all activities within a date range. Args: start_date: Start date of range end_date: End date of range activity_type: Optional filter by activity type Returns: List of activity data dictionaries """ logger.info(f"Querying activities from {start_date} to {end_date}") # Get activity index entries activity_indices = self.repo.get_activities_by_date_range(start_date, end_date) # Load full activity data activities = [] for idx in activity_indices: # Filter by activity type if specified if activity_type and idx.get("activity_type") != activity_type: continue activity_data = self.storage.load_activity(idx["activity_id"]) if activity_data: activities.append(activity_data) logger.info(f"Found {len(activities)} activities") return activities def get_metric_statistics( self, metric_type: str, start_date: date, end_date: date, fields: List[str] ) -> Dict[str, Dict[str, Any]]: """Calculate statistics for specific metric fields over a date range. Args: metric_type: Type of metric start_date: Start date of range end_date: End date of range fields: List of field names to calculate statistics for Returns: Dictionary mapping field names to their statistics (min, max, avg, count) """ logger.info( f"Calculating statistics for {metric_type} fields: {fields} from {start_date} to {end_date}" ) # Get all data for the range data_list = self.get_metric_range(metric_type, start_date, end_date) # Calculate statistics for each field stats = {} for field in fields: values = [] for data in data_list: if field in data and data[field] is not None: values.append(data[field]) if values: stats[field] = { "min": min(values), "max": max(values), "avg": sum(values) / len(values), "count": len(values), "sum": sum(values), } else: stats[field] = { "min": None, "max": None, "avg": None, "count": 0, "sum": None, } return stats def get_available_dates(self, metric_type: str) -> List[str]: """Get all available dates for a metric type. Args: metric_type: Type of metric Returns: List of date strings in ISO format """ # This is a simplified implementation - in a real scenario, # you might want to query the database index for better performance storage_path = config.DATA_TYPE_CONFIG[metric_type]["storage_path"] base_path = config.DATA_DIR / storage_path dates = [] if base_path.exists(): # Walk through year/month directories for year_dir in sorted(base_path.iterdir()): if not year_dir.is_dir(): continue for month_dir in sorted(year_dir.iterdir()): if not month_dir.is_dir(): continue for file_path in sorted(month_dir.glob("*.json")): # Extract date from filename (YYYY-MM-DD.json) date_str = file_path.stem dates.append(date_str) return sorted(dates) def get_latest_data(self, metric_type: str, days: int = 7) -> List[Dict[str, Any]]: """Get the most recent data for a metric. Args: metric_type: Type of metric days: Number of days to look back (default: 7) Returns: List of metric data dictionaries for the last N days """ end_date = date.today() start_date = end_date - timedelta(days=days - 1) return self.get_metric_range(metric_type, start_date, end_date)