Add body battery: sync, storage, and health UI chart
Build and push images / validate (push) Successful in 2s
Build and push images / build-backend (push) Successful in 5s
Build and push images / build-worker (push) Successful in 5s
Build and push images / build-frontend (push) Successful in 10s

Parses Garmin Connect get_body_battery() per day, storing charged/drained/
start+end levels and the fine-grained [[ts_ms, level, type, stress]] values
array in a new body_battery JSONB column on health_metrics.

Frontend adds:
- BatteryRing SVG gauge (color-scaled 0–100)
- BodyBatteryChart: ComposedChart with type-colored bars (REST/ACTIVE/SLEEP/
  STRESS) and battery level overlay line, matching Garmin's layout
- Body battery trend chart in the Trends section (end_level per day)

Also adds avg_hr_day and weight data which now correctly sync with the
intraday_hr JSON serialization fix from the previous commit.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
2026-06-07 11:13:38 +01:00
parent 37ffd4c9e0
commit 616099402b
5 changed files with 198 additions and 10 deletions
+61 -4
View File
@@ -202,6 +202,7 @@ def sync_wellness(garmin, user_id: int, since: Optional[datetime], db,
processed = 0
import time as _time
import json as _json
for i in range(max(days, 1)):
day = start_date + timedelta(days=i)
day_str = day.isoformat()
@@ -212,6 +213,7 @@ def sync_wellness(garmin, user_id: int, since: Optional[datetime], db,
# Intraday HR (requires display_name; skip gracefully if absent)
hr_raw = _safe(garmin.get_heart_rates, day_str) if garmin.display_name else None
bc_data = _safe(garmin.get_body_composition, day_str, day_str)
bb_raw = _safe(garmin.get_body_battery, day_str, day_str)
_time.sleep(0.25) # avoid hammering Garmin's wellness API
row = _parse_day(stats, sleep_data, hrv_data)
@@ -243,6 +245,12 @@ def sync_wellness(garmin, user_id: int, since: Optional[datetime], db,
bwf = float(bw)
_set(row, "weight_kg", round(bwf / 1000 if bwf > 300 else bwf, 2))
# Body battery — store summary + fine-grained timeline
if bb_raw:
bb = _parse_body_battery(bb_raw, day_str)
if bb:
row["body_battery"] = _json.dumps(bb)
# Intraday heart rate — store non-null [epoch_ms, bpm] pairs
if hr_raw:
raw_vals = hr_raw.get("heartRateValues") or []
@@ -253,11 +261,12 @@ def sync_wellness(garmin, user_id: int, since: Optional[datetime], db,
if not row:
continue
# psycopg2 treats Python lists as PostgreSQL arrays; serialize JSON columns
# explicitly so they arrive as a JSON string that the json/jsonb column accepts.
import json as _json
if "intraday_hr" in row and isinstance(row["intraday_hr"], list):
# psycopg2 treats Python lists/dicts as PG arrays/hstore; serialize JSON
# columns as strings so psycopg2 passes them correctly to json/jsonb columns.
if "intraday_hr" in row and not isinstance(row["intraday_hr"], str):
row["intraday_hr"] = _json.dumps(row["intraday_hr"])
if "body_battery" in row and not isinstance(row["body_battery"], str):
row["body_battery"] = _json.dumps(row["body_battery"])
cols = list(row.keys())
col_sql = ", ".join(cols)
@@ -288,6 +297,54 @@ def sync_wellness(garmin, user_id: int, since: Optional[datetime], db,
return processed
def _parse_body_battery(bb_response, day_str: str):
"""Parse get_body_battery() response for a single day into a compact dict."""
if not bb_response:
return None
entry = next((e for e in bb_response if e.get("date") == day_str), None)
if not entry and bb_response:
entry = bb_response[0]
if not entry:
return None
charged = entry.get("charged")
drained = entry.get("drained")
start_lvl = entry.get("startValue")
end_lvl = entry.get("endValue")
# Fine-grained timeline: [[ts_ms, level, type_code, stress], ...]
# type_code: 0=REST, 1=ACTIVE, 2=SLEEP, 3=STRESS, 4=UNMEASURABLE
values = entry.get("bodyBatteryValuesArray") or []
if not values:
# Fall back to bodyBatteryStatList (segment-level data)
type_map = {"REST": 0, "ACTIVE": 1, "SLEEP": 2, "STRESS": 3, "UNMEASURABLE": 4}
for seg in (entry.get("bodyBatteryStatList") or []):
ts_str = seg.get("startTimestampGMT") or seg.get("startTimestampLocal")
if ts_str:
try:
from datetime import datetime as _dt, timezone as _tz
ts = _dt.fromisoformat(ts_str.rstrip("Z")).replace(tzinfo=_tz.utc)
type_code = type_map.get(seg.get("activityType", "UNMEASURABLE"), 4)
values.append([int(ts.timestamp() * 1000),
int(seg.get("bodyBatteryLevel") or 0),
type_code,
int(seg.get("stressLevel") or -1)])
except Exception:
pass
if charged is None and end_lvl is None and not values:
return None
return {
"charged": charged,
"drained": drained,
"start_level": start_lvl,
"end_level": end_lvl,
"values": values, # stripped from list-API, returned in intraday endpoint
}
def _safe(fn, *args):
try:
return fn(*args)