Add initial project
This commit is contained in:
@@ -0,0 +1,162 @@
|
||||
"""
|
||||
Health Connect Provider - Implementation for Health Connect API.
|
||||
|
||||
This module provides the Health Connect implementation that reads data
|
||||
pushed from the Android companion app via the /api/health-connect/sync endpoint.
|
||||
"""
|
||||
import datetime
|
||||
from typing import Dict, Any, List, Optional
|
||||
|
||||
from backend.integrations.health_provider import (
|
||||
HealthDataProvider,
|
||||
DataSource,
|
||||
HealthProviderFactory
|
||||
)
|
||||
|
||||
|
||||
class HealthConnectProvider(HealthDataProvider):
|
||||
"""
|
||||
Health Connect implementation of HealthDataProvider.
|
||||
|
||||
This provider reads data that has been pushed from an Android companion app
|
||||
via the /api/health-connect/sync endpoint and stored in the database.
|
||||
"""
|
||||
|
||||
@property
|
||||
def source(self) -> DataSource:
|
||||
return DataSource.HEALTH_CONNECT
|
||||
|
||||
def get_auth_url(self) -> str:
|
||||
"""
|
||||
Health Connect doesn't use OAuth - permissions are granted on-device.
|
||||
|
||||
Returns a URL to download the Android companion app.
|
||||
"""
|
||||
# TODO: Replace with actual Play Store URL when app is published
|
||||
return "https://play.google.com/store/apps/details?id=com.habitforge.companion"
|
||||
|
||||
async def fetch_daily_metrics(
|
||||
self,
|
||||
access_token: Optional[str], # Not used for Health Connect, but required by interface
|
||||
days: int = 7,
|
||||
timezone: str = "UTC",
|
||||
user_id: Optional[int] = None
|
||||
) -> List[Dict[str, Any]]:
|
||||
"""
|
||||
Fetch daily metrics from Health Connect data store.
|
||||
|
||||
Reads data that was pushed from the Android companion app.
|
||||
"""
|
||||
if not user_id:
|
||||
print("[HealthConnect] ERROR: user_id is required")
|
||||
return []
|
||||
|
||||
print(f"[HealthConnect] Fetching daily metrics for user {user_id}, last {days} days")
|
||||
|
||||
from backend.database import SessionLocal
|
||||
from backend.models.health_connect import HealthConnectDailyMetrics
|
||||
from datetime import timedelta, date as date_type
|
||||
|
||||
db = SessionLocal()
|
||||
try:
|
||||
start_date = date_type.today() - timedelta(days=days)
|
||||
|
||||
metrics = db.query(HealthConnectDailyMetrics).filter(
|
||||
HealthConnectDailyMetrics.user_id == user_id,
|
||||
HealthConnectDailyMetrics.date >= start_date
|
||||
).order_by(HealthConnectDailyMetrics.date.desc()).all()
|
||||
|
||||
if not metrics:
|
||||
print(f"[HealthConnect] No data found. Ensure Android app has synced data.")
|
||||
return []
|
||||
|
||||
print(f"[HealthConnect] Found {len(metrics)} days of data")
|
||||
|
||||
# Convert to standard format
|
||||
return [
|
||||
{
|
||||
"date": m.date,
|
||||
"step_count": m.step_count or 0,
|
||||
"calories_burned": m.calories_burned or 0.0,
|
||||
"distance_meters": m.distance_meters or 0.0,
|
||||
"sleep_duration_minutes": m.sleep_duration_minutes or 0,
|
||||
"deep_sleep_minutes": m.deep_sleep_minutes or 0,
|
||||
"light_sleep_minutes": m.light_sleep_minutes or 0,
|
||||
"rem_sleep_minutes": m.rem_sleep_minutes or 0,
|
||||
"awake_duration_minutes": m.awake_duration_minutes or 0,
|
||||
"avg_heart_rate": m.avg_heart_rate or 0,
|
||||
"min_heart_rate": m.min_heart_rate or 0,
|
||||
"max_heart_rate": m.max_heart_rate or 0,
|
||||
"avg_spo2": m.avg_spo2 or 0.0,
|
||||
"pai_score": 0.0, # Calculate if needed
|
||||
"weight": m.weight,
|
||||
"resting_heart_rate": m.resting_heart_rate,
|
||||
"hrv": None
|
||||
}
|
||||
for m in metrics
|
||||
]
|
||||
finally:
|
||||
db.close()
|
||||
|
||||
async def fetch_workouts(
|
||||
self,
|
||||
access_token: Optional[str],
|
||||
days: int = 30
|
||||
) -> List[Dict[str, Any]]:
|
||||
"""
|
||||
Fetch exercise sessions from Health Connect data store.
|
||||
"""
|
||||
# Note: We need user_id but the interface doesn't pass it for workouts
|
||||
# This is a limitation we'll need to address in the interface
|
||||
print(f"[HealthConnect] fetch_workouts called")
|
||||
|
||||
from backend.database import SessionLocal
|
||||
from backend.models.health_connect import HealthConnectExerciseSession
|
||||
from datetime import timedelta
|
||||
|
||||
db = SessionLocal()
|
||||
try:
|
||||
start_date = datetime.datetime.now() - timedelta(days=days)
|
||||
|
||||
# Query all sessions (we can't filter by user without user_id)
|
||||
# In practice, this would be called with user context
|
||||
sessions = db.query(HealthConnectExerciseSession).filter(
|
||||
HealthConnectExerciseSession.start_time >= start_date
|
||||
).order_by(HealthConnectExerciseSession.start_time.desc()).all()
|
||||
|
||||
return [
|
||||
{
|
||||
"start_time": s.start_time,
|
||||
"end_time": s.end_time,
|
||||
"activity_type": s.exercise_type_name,
|
||||
"activity_type_id": s.exercise_type,
|
||||
"duration_seconds": s.duration_seconds,
|
||||
"distance_meters": s.distance_meters or 0.0,
|
||||
"calories": s.calories_burned or 0.0,
|
||||
"avg_hr": s.avg_heart_rate,
|
||||
"max_hr": s.max_heart_rate,
|
||||
"min_hr": s.min_heart_rate
|
||||
}
|
||||
for s in sessions
|
||||
]
|
||||
finally:
|
||||
db.close()
|
||||
|
||||
async def fetch_sleep_sessions(
|
||||
self,
|
||||
access_token: Optional[str],
|
||||
days: int = 30
|
||||
) -> List[Dict[str, Any]]:
|
||||
"""
|
||||
Fetch sleep sessions from Health Connect data store.
|
||||
|
||||
Note: Health Connect stores sleep as part of daily metrics.
|
||||
This returns an empty list as sleep is handled in fetch_daily_metrics.
|
||||
"""
|
||||
print(f"[HealthConnect] fetch_sleep_sessions called")
|
||||
# Sleep data is included in daily metrics for Health Connect
|
||||
return []
|
||||
|
||||
|
||||
# Register the provider with the factory
|
||||
HealthProviderFactory.register("health_connect", HealthConnectProvider)
|
||||
Reference in New Issue
Block a user