#!/usr/bin/env python3 """Import secondary metrics from Google Fit REST into Health Bridge. Handles heart_rate, weight, height, body_fat, nutrition, hydration. Uses adaptive chunk sizes to avoid 'too many points' errors. Posts typed records (heart_rate, weight, height, body_fat, nutrition, hydration) so Health Bridge server classifies and aggregates them correctly. """ import json import os import sys import urllib.request import urllib.error from datetime import datetime, date, time, timedelta from zoneinfo import ZoneInfo FIT_URL = "https://www.googleapis.com/fitness/v1/users/me/dataset:aggregate" INGEST_URL = os.environ.get("HEALTH_BRIDGE_INGEST_URL", "http://100.87.116.90:3007/ingest") HEALTH_TOKEN = os.environ.get("HEALTH_BRIDGE_TOKEN", "XXMYWq-ft146fRL0AlNV_5P9Vg4NNJJa4pHxqDvzRkg") SUBJECT = os.environ.get("HEALTH_BRIDGE_SUBJECT", "chicho") TZ = os.environ.get("HEALTH_BRIDGE_TIMEZONE", "America/Argentina/Buenos_Aires") SOURCE = "google_fit_rest" METRICS = { "heart_rate": { "dataTypeName": "com.google.heart_rate.bpm", "record_type": "heart_rate", }, "weight": { "dataTypeName": "com.google.weight", "record_type": "weight", }, "height": { "dataTypeName": "com.google.height", "record_type": "height", }, "body_fat": { "dataTypeName": "com.google.body.fat.percentage", "record_type": "body_fat", }, "nutrition": { "dataTypeName": "com.google.nutrition", "record_type": "nutrition", }, "hydration": { "dataTypeName": "com.google.hydration", "record_type": "hydration", }, } def millis(d: date, zone: ZoneInfo) -> int: return int(datetime.combine(d, time.min, tzinfo=zone).timestamp() * 1000) def local_date_str(start_ms: int, zone: ZoneInfo) -> str: return datetime.fromtimestamp(int(start_ms) / 1000, zone).date().isoformat() def iso_utc(ms: int) -> str: return datetime.fromtimestamp(int(ms) / 1000, ZoneInfo("UTC")).isoformat().replace("+00:00", "Z") def req_json(url, body=None, headers=None): data = json.dumps(body).encode() if body is not None else None r = urllib.request.Request( url, data=data, headers=headers or {}, method="POST" if body is not None else "GET" ) try: with urllib.request.urlopen(r, timeout=60) as resp: return json.loads(resp.read().decode()) except urllib.error.HTTPError as e: detail = e.read().decode() raise RuntimeError(f"HTTP {e.code}: {detail}") from e def query_aggregate(token, metric_cfg, start, end, zone): body = { "aggregateBy": [{"dataTypeName": metric_cfg["dataTypeName"]}], "bucketByTime": { "period": {"type": "day", "value": 1, "timeZoneId": TZ} }, "startTimeMillis": millis(start, zone), "endTimeMillis": millis(end + timedelta(days=1), zone), } return req_json( FIT_URL, body, { "Authorization": f"Bearer {token}", "Content-Type": "application/json;encoding=utf-8", }, ) def parse_hr_point(p): """com.google.heart_rate.summary -> [avg, max, min]""" vals = p.get("value", []) if len(vals) >= 3: return { "avg_bpm": float(vals[0].get("fpVal", 0)), "max_bpm": float(vals[1].get("fpVal", 0)), "min_bpm": float(vals[2].get("fpVal", 0)), } return None def parse_weight_point(p): """com.google.weight.summary -> [avg, min, max]""" vals = p.get("value", []) if vals: return {"kg": float(vals[0].get("fpVal", 0))} return None def parse_height_point(p): """com.google.height.summary -> [avg, min, max]""" vals = p.get("value", []) if vals: return {"meters": float(vals[0].get("fpVal", 0))} return None def parse_body_fat_point(p): """com.google.body.fat.percentage.summary -> [avg, min, max]""" vals = p.get("value", []) if vals: return {"percentage": float(vals[0].get("fpVal", 0))} return None def parse_nutrition_point(p): """com.google.nutrition.summary -> mapVal with nutrients""" vals = p.get("value", []) if not vals: return None # First value object contains the mapVal; second is usually intVal=0 m = vals[0].get("mapVal", []) out = {} for item in m: key = item.get("key") v = item.get("value", {}) fp = v.get("fpVal", 0) out[key] = fp return { "energy_kcal": out.get("calories", 0), "protein_g": out.get("protein", 0), "carbs_g": out.get("carbs.total", 0), "fat_g": out.get("fat.total", 0), } def parse_hydration_point(p): """com.google.hydration -> fpVal in litres""" vals = p.get("value", []) if vals: return {"volume_ml": float(vals[0].get("fpVal", 0)) * 1000} return None PARSERS = { "heart_rate": parse_hr_point, "weight": parse_weight_point, "height": parse_height_point, "body_fat": parse_body_fat_point, "nutrition": parse_nutrition_point, "hydration": parse_hydration_point, } def import_metric(token, metric_key, metric_cfg, zone, start, end): """Query one metric across the date range with adaptive chunk sizing.""" records = [] chunks = 0 chunk_sizes = [30, 14, 7, 3] # adaptive fallback cur = start while cur <= end: for chunk_days in chunk_sizes: chunk_end = min(cur + timedelta(days=chunk_days - 1), end) try: fit = query_aggregate(token, metric_cfg, cur, chunk_end, zone) chunks += 1 for b in fit.get("bucket", []): has = any(ds.get("point") for ds in b.get("dataset", [])) if not has: continue local = local_date_str(b["startTimeMillis"], zone) for ds in b.get("dataset", []): for p in ds.get("point", []): parsed = PARSERS[metric_key](p) if parsed is None: continue rec = { "type": metric_cfg["record_type"], "subject": SUBJECT, "source": SOURCE, "local_date": local, "record_key": f"{metric_cfg['record_type']}:{SUBJECT}:{local}", } rec.update(parsed) records.append(rec) cur = chunk_end + timedelta(days=1) break # success with this chunk size except RuntimeError as e: if "too many points" in str(e).lower(): if chunk_days == chunk_sizes[-1]: print(f" WARN: even {chunk_days}-day chunk too large for {metric_key} at {cur}", file=sys.stderr) cur = chunk_end + timedelta(days=1) break # try smaller chunk continue raise return records, chunks def main(): token = os.environ.get("GOOGLE_FIT_ACCESS_TOKEN") if not token: print("ERROR: set GOOGLE_FIT_ACCESS_TOKEN", file=sys.stderr) return 2 zone = ZoneInfo(TZ) # Default range per metric to avoid scanning empty years metric_starts = { "heart_rate": "2023-01-01", "weight": "2023-01-01", "height": "2023-01-01", "body_fat": "2024-01-01", "nutrition": "2023-01-01", "hydration": "2023-01-01", } end = date.fromisoformat(os.environ.get("GOOGLE_FIT_END", datetime.now(zone).date().isoformat())) all_records = [] total_chunks = 0 stats = {} for metric_key, metric_cfg in METRICS.items(): start = date.fromisoformat(os.environ.get("GOOGLE_FIT_START", metric_starts.get(metric_key, "2015-01-01"))) print(f"Importing {metric_key} ({start} to {end})...", file=sys.stderr) records, chunks = import_metric(token, metric_key, metric_cfg, zone, start, end) total_chunks += chunks all_records.extend(records) stats[metric_key] = { "records": len(records), "chunks": chunks, } if records: # quick peek print(f" {len(records)} records from {chunks} chunks", file=sys.stderr) if not all_records: print(json.dumps({"status": "ok", "records": 0, "metrics": stats}, indent=2)) return 0 payload = {"subject": SUBJECT, "timezone": TZ, "records": all_records} ingest = req_json( INGEST_URL, payload, { "Authorization": f"Bearer {HEALTH_TOKEN}", "Content-Type": "application/json", }, ) print( json.dumps( { "status": "ok", "total_records": len(all_records), "total_chunks": total_chunks, "metrics": stats, "ingest": ingest, }, indent=2, ) ) return 0 if __name__ == "__main__": raise SystemExit(main())