owain
67fd4b3c96
- Sleep: store per-epoch stage timestamps in new sleep_stages JSON column; DailySnapshot now renders a proper 4-lane hypnogram (Awake/REM/Light/Deep) instead of the old proportional flat bar - Body battery: replace grey background bars + white line with per-minute bars coloured by inferred type (sleep=indigo, rest=teal, active=orange, stable=grey) derived from sleep window + battery direction; Y-axis fixed 0-100 - Routes: convert sidebar list to tile grid sorted by most completions; tiles colour-bordered by sport type (blue=running, orange=cycling); completion count shown on each tile; detail panel displays below the grid when a tile is clicked - Segments on activity detail: add column headers (This run / Best / Δ) and show signed time delta vs best, green when faster, red when slower Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
357 lines
14 KiB
Python
357 lines
14 KiB
Python
"""
|
|
Garmin wellness FIT file parser using the official Garmin FIT Python SDK.
|
|
The SDK with convert_types_to_strings=True returns snake_case field names.
|
|
|
|
Sleep stages: message 275 (modern) or 269 (older) each carry a start timestamp
|
|
and a stage name. Duration of each stage = gap to the next stage's timestamp.
|
|
The sleep session stop time (from event message 21, event_type='stop') closes
|
|
the last stage.
|
|
"""
|
|
from datetime import datetime, timezone, date
|
|
from typing import Optional
|
|
from garmin_fit_sdk import Decoder, Stream
|
|
|
|
|
|
FIT_EPOCH_S = 631065600
|
|
SLEEP_LEVEL_MAP = {"unmeasurable": 0, "awake": 1, "light": 2, "deep": 3, "rem": 4}
|
|
|
|
|
|
def _fit_ts(raw) -> Optional[datetime]:
|
|
if raw is None:
|
|
return None
|
|
try:
|
|
s = int(raw)
|
|
if s <= 0 or s == 0xFFFFFFFF:
|
|
return None
|
|
return datetime.fromtimestamp(s + FIT_EPOCH_S, tz=timezone.utc)
|
|
except (TypeError, ValueError, OverflowError, OSError):
|
|
return None
|
|
|
|
|
|
def _to_date(val) -> Optional[date]:
|
|
if val is None:
|
|
return None
|
|
if isinstance(val, datetime):
|
|
if val.tzinfo is None:
|
|
val = val.replace(tzinfo=timezone.utc)
|
|
return val.date()
|
|
if isinstance(val, (int, float)):
|
|
dt = _fit_ts(val)
|
|
return dt.date() if dt else None
|
|
return None
|
|
|
|
|
|
def _to_dt(val) -> Optional[datetime]:
|
|
if isinstance(val, datetime):
|
|
return val.replace(tzinfo=timezone.utc) if val.tzinfo is None else val
|
|
if isinstance(val, (int, float)):
|
|
return _fit_ts(val)
|
|
return None
|
|
|
|
|
|
def parse_wellness_fit(file_path: str) -> dict:
|
|
"""
|
|
Parse a Garmin wellness/monitoring FIT file.
|
|
Returns {"days": {date: metrics_dict}, "error": str|None}
|
|
"""
|
|
daily = {}
|
|
last_date_seen = [None]
|
|
|
|
def ensure_day(d: date) -> dict:
|
|
if d not in daily:
|
|
daily[d] = {
|
|
"heart_rates": [],
|
|
"stress_values": [],
|
|
"spo2_readings": [],
|
|
# Each entry: (datetime, level_int) — duration computed from gaps
|
|
"sleep_epochs": [],
|
|
"sleep_start": None,
|
|
"sleep_end": None,
|
|
"steps": None,
|
|
"floors_climbed": None,
|
|
"active_calories": None,
|
|
"bmr": None,
|
|
"resting_hr": None,
|
|
"hrv_nightly_avg": None,
|
|
"hrv_5min_high": None,
|
|
"hrv_status": None,
|
|
"sleep_score": None,
|
|
}
|
|
return daily[d]
|
|
|
|
def _add_sleep_epoch(ts: datetime, level_raw):
|
|
d = _to_date(ts)
|
|
if not d:
|
|
return
|
|
last_date_seen[0] = d
|
|
if isinstance(level_raw, str):
|
|
level = SLEEP_LEVEL_MAP.get(level_raw.lower())
|
|
else:
|
|
level = level_raw
|
|
if level is not None:
|
|
ensure_day(d)["sleep_epochs"].append((ts, int(level)))
|
|
|
|
def listener(mesg_num: int, msg: dict):
|
|
|
|
# ── monitoring_info (147) - older firmware ─────────────────────────
|
|
if mesg_num == 147:
|
|
d = _to_date(msg.get("timestamp") or msg.get("local_timestamp"))
|
|
rhr = msg.get("resting_heart_rate")
|
|
if d and rhr and 20 < rhr < 120:
|
|
last_date_seen[0] = d
|
|
ensure_day(d)["resting_hr"] = int(rhr)
|
|
|
|
# ── monitoring (148) - older firmware ──────────────────────────────
|
|
elif mesg_num == 148:
|
|
d = _to_date(msg.get("timestamp") or msg.get("local_timestamp"))
|
|
if not d:
|
|
return
|
|
last_date_seen[0] = d
|
|
entry = ensure_day(d)
|
|
hr = msg.get("heart_rate")
|
|
if hr and 20 < hr < 250:
|
|
entry["heart_rates"].append(int(hr))
|
|
steps = msg.get("steps") or msg.get("cycles")
|
|
if steps and steps > 0:
|
|
entry["steps"] = max(entry["steps"] or 0, int(steps))
|
|
stress = msg.get("stress_level_value")
|
|
if stress is not None and stress >= 0:
|
|
entry["stress_values"].append(int(stress))
|
|
|
|
# ── monitoring (55) - modern, per-interval running totals ──────────
|
|
elif mesg_num == 55:
|
|
d = _to_date(msg.get("timestamp"))
|
|
if not d:
|
|
return
|
|
last_date_seen[0] = d
|
|
entry = ensure_day(d)
|
|
hr = msg.get("heart_rate")
|
|
if hr and 20 < hr < 250:
|
|
entry["heart_rates"].append(int(hr))
|
|
steps = msg.get("steps")
|
|
if steps and steps > 0:
|
|
entry["steps"] = max(entry["steps"] or 0, int(steps))
|
|
active_cal = msg.get("active_calories")
|
|
if active_cal and active_cal > 0:
|
|
entry["active_calories"] = max(entry["active_calories"] or 0, float(active_cal))
|
|
ascent = msg.get("ascent")
|
|
if ascent and ascent > 0:
|
|
# Garmin counts 1 floor ≈ 3 m of ascent
|
|
floors = max(1, round(float(ascent) / 3))
|
|
entry["floors_climbed"] = max(entry["floors_climbed"] or 0, floors)
|
|
|
|
# ── monitoring_info (103) - calibration; carries BMR ───────────────
|
|
elif mesg_num == 103:
|
|
d = _to_date(msg.get("timestamp"))
|
|
if not d:
|
|
return
|
|
last_date_seen[0] = d
|
|
bmr = msg.get("resting_metabolic_rate")
|
|
if bmr and bmr > 0:
|
|
ensure_day(d)["bmr"] = int(bmr)
|
|
|
|
# ── hrv_status_summary (370) - modern HRV ─────────────────────────
|
|
elif mesg_num == 370:
|
|
d = _to_date(msg.get("timestamp"))
|
|
if not d:
|
|
return
|
|
last_date_seen[0] = d
|
|
entry = ensure_day(d)
|
|
hrv_avg = msg.get("last_night_average")
|
|
if hrv_avg and hrv_avg > 0:
|
|
entry["hrv_nightly_avg"] = float(hrv_avg)
|
|
hrv_high = msg.get("last_night_5_min_high")
|
|
if hrv_high and hrv_high > 0:
|
|
entry["hrv_5min_high"] = float(hrv_high)
|
|
status = msg.get("status")
|
|
if status:
|
|
entry["hrv_status"] = str(status)
|
|
|
|
# ── message 275 - sleep epochs (modern) or HRV (older firmware) ───
|
|
elif mesg_num == 275:
|
|
sleep_level = msg.get("sleep_level")
|
|
ts = _to_dt(msg.get("timestamp"))
|
|
if sleep_level is not None and ts:
|
|
_add_sleep_epoch(ts, sleep_level)
|
|
elif ts:
|
|
# Older firmware: HRV summary in message 275
|
|
d = _to_date(ts)
|
|
if d:
|
|
last_date_seen[0] = d
|
|
entry = ensure_day(d)
|
|
for key in ("weekly_average", "last_night_avg", "hrv_nightly_avg"):
|
|
v = msg.get(key)
|
|
if v and v > 0:
|
|
entry["hrv_nightly_avg"] = float(v)
|
|
break
|
|
high = msg.get("last_night_5_min_high")
|
|
if high:
|
|
entry["hrv_5min_high"] = float(high)
|
|
status = msg.get("hrv_status") or msg.get("status")
|
|
if status:
|
|
entry["hrv_status"] = str(status)
|
|
|
|
# ── sleep_level (269) - older firmware sleep epochs ────────────────
|
|
elif mesg_num == 269:
|
|
ts = _to_dt(msg.get("timestamp"))
|
|
level = msg.get("sleep_level")
|
|
if ts and level is not None:
|
|
_add_sleep_epoch(ts, level)
|
|
|
|
# ── event (21) - sleep session start / stop ────────────────────────
|
|
elif mesg_num == 21:
|
|
ts = _to_dt(msg.get("timestamp"))
|
|
if not ts:
|
|
return
|
|
d = _to_date(ts)
|
|
if not d:
|
|
return
|
|
event_type = msg.get("event_type")
|
|
if event_type == "start":
|
|
last_date_seen[0] = d
|
|
ensure_day(d)["sleep_start"] = ts
|
|
elif event_type == "stop":
|
|
last_date_seen[0] = d
|
|
ensure_day(d)["sleep_end"] = ts
|
|
|
|
# ── sleep_assessment (346) - overall sleep score, no timestamp ────
|
|
elif mesg_num == 346:
|
|
d = last_date_seen[0]
|
|
if not d:
|
|
return
|
|
score = msg.get("overall_sleep_score")
|
|
if score and score > 0:
|
|
ensure_day(d)["sleep_score"] = int(score)
|
|
|
|
# ── stress_level (132) ─────────────────────────────────────────────
|
|
elif mesg_num == 132:
|
|
d = _to_date(msg.get("stress_level_time") or msg.get("timestamp"))
|
|
if not d:
|
|
return
|
|
last_date_seen[0] = d
|
|
stress = msg.get("stress_level_value")
|
|
if stress is not None and stress >= 0:
|
|
ensure_day(d)["stress_values"].append(int(stress))
|
|
|
|
# ── spo2_data (258) ────────────────────────────────────────────────
|
|
elif mesg_num == 258:
|
|
d = _to_date(msg.get("timestamp"))
|
|
if not d:
|
|
return
|
|
last_date_seen[0] = d
|
|
spo2 = msg.get("spo2_percent") or msg.get("reading_spo2")
|
|
if spo2 and 50 < spo2 <= 100:
|
|
ensure_day(d)["spo2_readings"].append(float(spo2))
|
|
|
|
# ── per-minute stress + HR (227) proprietary ───────────────────────
|
|
elif mesg_num == 227:
|
|
d = _to_date(msg.get("stress_level_time") or msg.get("timestamp"))
|
|
if not d:
|
|
return
|
|
last_date_seen[0] = d
|
|
entry = ensure_day(d)
|
|
hr_raw = msg.get(2)
|
|
if hr_raw and isinstance(hr_raw, (int, float)) and 20 < hr_raw < 250:
|
|
entry["heart_rates"].append(int(hr_raw))
|
|
stress = msg.get("stress_level_value")
|
|
if stress is None:
|
|
stress = msg.get(0)
|
|
if stress is not None and isinstance(stress, (int, float)) and stress >= 0:
|
|
entry["stress_values"].append(int(stress))
|
|
|
|
# ── daily resting HR (211) proprietary ─────────────────────────────
|
|
elif mesg_num == 211:
|
|
d = _to_date(msg.get("timestamp"))
|
|
if not d:
|
|
return
|
|
last_date_seen[0] = d
|
|
entry = ensure_day(d)
|
|
rhr = msg.get("resting_heart_rate") or msg.get("current_day_resting_heart_rate")
|
|
if rhr and isinstance(rhr, (int, float)) and 20 < rhr < 120:
|
|
entry["resting_hr"] = int(rhr)
|
|
|
|
try:
|
|
stream = Stream.from_file(file_path)
|
|
decoder = Decoder(stream)
|
|
messages, errors = decoder.read(
|
|
apply_scale_and_offset=True,
|
|
convert_datetimes_to_dates=True,
|
|
convert_types_to_strings=True,
|
|
enable_crc_check=False,
|
|
expand_sub_fields=True,
|
|
expand_components=True,
|
|
merge_heart_rates=False,
|
|
mesg_listener=listener,
|
|
)
|
|
except Exception as e:
|
|
return {"error": str(e), "days": {}}
|
|
|
|
result = {}
|
|
for day_date, data in daily.items():
|
|
hrs = data.pop("heart_rates", [])
|
|
stresses = data.pop("stress_values", [])
|
|
spo2s = data.pop("spo2_readings", [])
|
|
sleep_epochs = data.pop("sleep_epochs", [])
|
|
sleep_end_ts = data.pop("sleep_end", None)
|
|
sleep_start_ts = data.pop("sleep_start", None)
|
|
|
|
avg_hr = round(sum(hrs) / len(hrs), 1) if hrs else None
|
|
max_hr = max(hrs) if hrs else None
|
|
avg_stress = round(sum(s for s in stresses if s >= 0) / len(stresses), 1) if stresses else None
|
|
spo2_avg = round(sum(spo2s) / len(spo2s), 1) if spo2s else None
|
|
|
|
# Compute sleep stage durations from epoch timestamps
|
|
if sleep_epochs:
|
|
epochs_sorted = sorted(sleep_epochs, key=lambda x: x[0])
|
|
level_secs = {1: 0, 2: 0, 3: 0, 4: 0} # awake, light, deep, rem
|
|
for i, (ts, level) in enumerate(epochs_sorted):
|
|
if i + 1 < len(epochs_sorted):
|
|
next_ts = epochs_sorted[i + 1][0]
|
|
elif sleep_end_ts:
|
|
next_ts = sleep_end_ts
|
|
else:
|
|
continue
|
|
dur = (next_ts - ts).total_seconds()
|
|
if level in level_secs and dur > 0:
|
|
level_secs[level] += dur
|
|
sleep_deep_s = level_secs[3] or None
|
|
sleep_light_s = level_secs[2] or None
|
|
sleep_rem_s = level_secs[4] or None
|
|
sleep_awake_s = level_secs[1] or None
|
|
sleep_duration_s = (level_secs[2] + level_secs[3] + level_secs[4]) or None
|
|
sleep_stages = [[int(ts.timestamp() * 1000), level] for ts, level in epochs_sorted]
|
|
else:
|
|
sleep_deep_s = sleep_light_s = sleep_rem_s = sleep_awake_s = sleep_duration_s = None
|
|
sleep_stages = None
|
|
|
|
active_cal = data.get("active_calories")
|
|
bmr = data.get("bmr")
|
|
# Require active_cal so we don't store BMR-only as "total" calories
|
|
total_cal = float(bmr + active_cal) if (bmr and active_cal) else None
|
|
|
|
result[day_date] = {
|
|
"resting_hr": data.get("resting_hr"),
|
|
"avg_hr_day": avg_hr,
|
|
"max_hr_day": max_hr,
|
|
"avg_stress": avg_stress,
|
|
"spo2_avg": spo2_avg,
|
|
"hrv_nightly_avg": data.get("hrv_nightly_avg"),
|
|
"hrv_5min_high": data.get("hrv_5min_high"),
|
|
"hrv_status": data.get("hrv_status"),
|
|
"steps": data.get("steps"),
|
|
"floors_climbed": data.get("floors_climbed"),
|
|
"active_calories": active_cal,
|
|
"total_calories": total_cal,
|
|
"sleep_duration_s": sleep_duration_s,
|
|
"sleep_deep_s": sleep_deep_s,
|
|
"sleep_light_s": sleep_light_s,
|
|
"sleep_rem_s": sleep_rem_s,
|
|
"sleep_awake_s": sleep_awake_s,
|
|
"sleep_score": data.get("sleep_score"),
|
|
"sleep_start": sleep_start_ts,
|
|
"sleep_end": sleep_end_ts,
|
|
"sleep_stages": sleep_stages,
|
|
}
|
|
|
|
return {"days": result, "error": None}
|