"""Health check and metric endpoints for API v1.""" from fastapi import APIRouter, Depends, HTTPException, status from fastapi.responses import FileResponse import os from sqlalchemy.orm import Session from sqlalchemy import text from typing import List, Optional import datetime from backend.database import get_db from backend.models.daily_health_metrics import DailyHealthMetrics from backend.models.user import User from backend.models.workout_session import WorkoutSession from backend.schemas.daily_health_metrics import DailyHealthMetricsResponse from backend import auth router = APIRouter(prefix="/api/health", tags=["health"]) @router.get("", status_code=status.HTTP_200_OK) def health_check(db: Session = Depends(get_db)): """Check if the application and database are healthy.""" try: # Check database connection db.execute(text("SELECT 1")) return { "status": "healthy", "database": "connected" } except Exception as e: raise HTTPException(status_code=503, detail=f"Unhealthy: {str(e)}") # Valid endpoints start here @router.get("/stats", response_model=List[DailyHealthMetricsResponse]) def get_health_stats( period: str = "week", ref_date: Optional[datetime.date] = None, db: Session = Depends(get_db), current_user: User = Depends(auth.get_current_user) ): """Retrieve health statistics for various periods.""" # Imports for Raw Data from backend.models.health_connect_raw import HCRawWeight from sqlalchemy import func # Helper to fetch raw weights def get_raw_weights(start, end): return db.query(HCRawWeight).filter( HCRawWeight.user_id == current_user.id, HCRawWeight.time >= start, HCRawWeight.time <= end ).order_by(HCRawWeight.time.asc()).all() if ref_date is None: ref_date = datetime.date.today() start_date = None end_date = datetime.datetime.combine(ref_date, datetime.time.max) # End of ref day if period == "day": start_date = datetime.datetime.combine(ref_date, datetime.time.min) elif period == "week": start_date = datetime.datetime.combine(ref_date - datetime.timedelta(days=6), datetime.time.min) elif period == "month": start_date = datetime.datetime.combine(ref_date - datetime.timedelta(days=29), datetime.time.min) elif period == "year": start_date = datetime.datetime.combine(ref_date - datetime.timedelta(days=365), datetime.time.min) # Base Metrics Query metrics_query = db.query(DailyHealthMetrics).filter( DailyHealthMetrics.user_id == current_user.id ) if period == "day": metrics_query = metrics_query.filter(DailyHealthMetrics.date == ref_date) else: metrics_query = metrics_query.filter(DailyHealthMetrics.date.between(start_date.date(), ref_date)) metrics_data = metrics_query.order_by(DailyHealthMetrics.date.asc()).all() # If Year, handle aggregation specially if period == "year": # Raw weights for the year raw_weights = get_raw_weights(start_date, end_date) # Month-based aggregation for metrics aggregated = [] import itertools from statistics import mean # Merge raw weights into metrics_data conceptually? # Actually easier to aggregate metrics_data first, then overlay weight. metrics_by_month = {} for m in metrics_data: key = (m.date.year, m.date.month) if key not in metrics_by_month: metrics_by_month[key] = [] metrics_by_month[key].append(m) weights_by_month = {} for w in raw_weights: key = (w.time.year, w.time.month) if key not in weights_by_month: weights_by_month[key] = [] weights_by_month[key].append(w.weight_kg) # distinct months from both sources all_months = set(metrics_by_month.keys()) | set(weights_by_month.keys()) for key in sorted(all_months): year, month = key g_metrics = metrics_by_month.get(key, []) g_weights = weights_by_month.get(key, []) # Use first metric date or construct one if g_metrics: base_date = g_metrics[0].date.replace(day=1) else: base_date = datetime.date(year, month, 1) agg = { "id": int(f"{year}{month:02d}"), # Fake ID "user_id": current_user.id, "date": base_date, # Defaults if no metrics "step_count": (sum(x.step_count for x in g_metrics) // len(g_metrics)) if g_metrics else 0, "step_count_total": sum(x.step_count for x in g_metrics) if g_metrics else 0, "calories_burned": (sum(x.calories_burned for x in g_metrics) / len(g_metrics)) if g_metrics else 0, "distance_meters": (sum(x.distance_meters for x in g_metrics) / len(g_metrics)) if g_metrics else 0, "sleep_duration_minutes": (sum(x.sleep_duration_minutes for x in g_metrics) // len(g_metrics)) if g_metrics else 0, "avg_heart_rate": (sum(x.avg_heart_rate or 0 for x in g_metrics) // len(g_metrics)) if g_metrics else 0, # ... skip other fields for brevity unless critical ... "min_heart_rate": min((x.min_heart_rate for x in g_metrics if x.min_heart_rate), default=0) if g_metrics else 0, "max_heart_rate": max((x.max_heart_rate for x in g_metrics if x.max_heart_rate), default=0) if g_metrics else 0, # WEIGHT from RAW (schema uses 'weight') "weight": mean(g_weights) if g_weights else 0 } # If weight missing in raw but present in aggregated (unlikely but possible) if agg["weight"] == 0 and g_metrics: # Fallback to avg of metrics weight if available w_vals = [x.weight for x in g_metrics if x.weight] if w_vals: agg["weight"] = mean(w_vals) aggregated.append(agg) return aggregated # For Day/Week/Month # Fetch Raw Weights raw_weights = get_raw_weights(start_date, end_date) # Map raw weights to date (take latest per date) weights_by_date = {} for w in raw_weights: d = w.time.date() weights_by_date[d] = w.weight_kg # overwrites with latest # Merge into metrics_data # 1. Update existing processed_dates = set() final_list = [] for m in metrics_data: processed_dates.add(m.date) # Overlay raw weight if available if m.date in weights_by_date: m.weight = weights_by_date[m.date] # set attribute dynamically final_list.append(m) # 2. Add missing dates (if we have weight but no daily metric) for d, w in weights_by_date.items(): if d not in processed_dates: # Create dummy object dummy = DailyHealthMetricsResponse( id=0, user_id=current_user.id, date=d, step_count=0, calories_burned=0, distance_meters=0, sleep_duration_minutes=0, weight=w, # Our raw weight # defaults deep_sleep_minutes=0, light_sleep_minutes=0, rem_sleep_minutes=0, awake_duration_minutes=0, pai_score=0, avg_heart_rate=0, min_heart_rate=0, max_heart_rate=0, avg_spo2=0, resting_heart_rate=0, hrv=0 ) final_list.append(dummy) # Sort final_list.sort(key=lambda x: x.date) return final_list @router.get("/workouts") def get_workouts( limit: int = 100, offset: int = 0, activity_type: Optional[str] = None, start_date: Optional[datetime.date] = None, end_date: Optional[datetime.date] = None, db: Session = Depends(get_db), current_user: User = Depends(auth.get_current_user) ): """Retrieve workout sessions for the current user (merges manual and raw data).""" from backend.models.workout_session import WorkoutSession from backend.models.health_connect_raw import HCRawExerciseSession start_dt = datetime.datetime.combine(start_date, datetime.time.min) if start_date else None end_dt = datetime.datetime.combine(end_date, datetime.time.max) if end_date else None # 1. Fetch from WorkoutSession query_ws = db.query(WorkoutSession).filter(WorkoutSession.user_id == current_user.id) if activity_type: query_ws = query_ws.filter(WorkoutSession.activity_type == activity_type) if start_dt: query_ws = query_ws.filter(WorkoutSession.start_time >= start_dt) if end_dt: query_ws = query_ws.filter(WorkoutSession.start_time <= end_dt) workouts_ws = query_ws.order_by(WorkoutSession.start_time.desc()).offset(offset).limit(limit).all() # 2. Fetch from HCRawExerciseSession query_raw = db.query(HCRawExerciseSession).filter(HCRawExerciseSession.user_id == current_user.id) if activity_type: query_raw = query_raw.filter(HCRawExerciseSession.exercise_type_name == activity_type) if start_dt: query_raw = query_raw.filter(HCRawExerciseSession.start_time >= start_dt) if end_dt: query_raw = query_raw.filter(HCRawExerciseSession.start_time <= end_dt) workouts_raw = query_raw.order_by(HCRawExerciseSession.start_time.desc()).offset(offset).limit(limit).all() # Combine and Map combined = [] for w in workouts_ws: combined.append({ "id": f"ws_{w.id}", "user_id": w.user_id, "start_time": w.start_time.isoformat() if w.start_time else None, "end_time": w.end_time.isoformat() if w.end_time else None, "activity_type": w.activity_type, "activity_type_id": w.activity_type_id, "duration_seconds": w.duration_seconds or 0, "distance_meters": w.distance_meters or 0.0, "calories": w.calories or 0.0, "avg_pace": w.avg_pace, "max_pace": w.max_pace, "min_pace": w.min_pace, "avg_hr": w.avg_hr, "max_hr": w.max_hr, "min_hr": w.min_hr, "notes": w.notes, "source": w.data_source }) for w in workouts_raw: duration = 0 if w.start_time and w.end_time: duration = int((w.end_time - w.start_time).total_seconds()) combined.append({ "id": f"raw_{w.id}", "user_id": w.user_id, "start_time": w.start_time.isoformat() if w.start_time else None, "end_time": w.end_time.isoformat() if w.end_time else None, "activity_type": w.exercise_type_name or f"Activity {w.exercise_type}", "activity_type_id": w.exercise_type, "duration_seconds": duration, "distance_meters": w.distance_meters or 0.0, "calories": w.calories or 0.0, "avg_pace": None, "max_pace": None, "min_pace": None, "avg_hr": w.avg_heart_rate, "max_hr": w.max_heart_rate, "min_hr": w.min_heart_rate, "notes": w.notes, "source": "health_connect" }) # Sort combined list by start_time DESC combined.sort(key=lambda x: x["start_time"] if x["start_time"] else "", reverse=True) return combined[:limit] @router.get("/workouts/summary") def get_workout_summary( start_date: Optional[datetime.date] = None, end_date: Optional[datetime.date] = None, db: Session = Depends(get_db), current_user: User = Depends(auth.get_current_user) ): """Get summary statistics for workouts (combines manual and raw data).""" from backend.models.workout_session import WorkoutSession from backend.models.health_connect_raw import HCRawExerciseSession from sqlalchemy import func start_dt = datetime.datetime.combine(start_date, datetime.time.min) if start_date else None end_dt = datetime.datetime.combine(end_date, datetime.time.max) if end_date else None # 1. Stats from WorkoutSession query_ws = db.query( WorkoutSession.activity_type, func.count(WorkoutSession.id).label('count'), func.sum(WorkoutSession.duration_seconds).label('total_duration'), func.sum(WorkoutSession.calories).label('total_calories'), func.sum(WorkoutSession.distance_meters).label('total_distance') ).filter(WorkoutSession.user_id == current_user.id) if start_dt: query_ws = query_ws.filter(WorkoutSession.start_time >= start_dt) if end_dt: query_ws = query_ws.filter(WorkoutSession.start_time <= end_dt) by_type_ws = query_ws.group_by(WorkoutSession.activity_type).all() # 2. Stats from HCRawExerciseSession # We fetch the sessions and aggregate in Python to avoid SQLite date extraction issues query_raw = db.query(HCRawExerciseSession).filter(HCRawExerciseSession.user_id == current_user.id) if start_dt: query_raw = query_raw.filter(HCRawExerciseSession.start_time >= start_dt) if end_dt: query_raw = query_raw.filter(HCRawExerciseSession.start_time <= end_dt) sessions_raw = query_raw.all() # Merge results summary_map = {} # Process manual sessions for row in by_type_ws: name = row[0] or "Unknown" summary_map[name] = { "count": row[1], "duration": row[2] or 0, "calories": row[3] or 0, "distance": row[4] or 0 } # Process raw sessions for w in sessions_raw: name = w.exercise_type_name or f"Activity {w.exercise_type}" duration = 0 if w.start_time and w.end_time: duration = (w.end_time - w.start_time).total_seconds() if name in summary_map: summary_map[name]["count"] += 1 summary_map[name]["duration"] += duration summary_map[name]["calories"] += (w.calories or 0) summary_map[name]["distance"] += (w.distance_meters or 0) else: summary_map[name] = { "count": 1, "duration": duration, "calories": w.calories or 0, "distance": w.distance_meters or 0 } total_workouts = sum(s["count"] for s in summary_map.values()) return { "total_workouts": total_workouts, "by_activity_type": [ { "activity_type": name, "count": stats["count"], "total_duration_seconds": int(stats["duration"]), "total_calories": round(stats["calories"], 1), "total_distance_meters": round(stats["distance"], 1) } for name, stats in summary_map.items() ] }