""" Garmin wellness FIT file parser using the official Garmin FIT Python SDK. The official SDK (garmin-fit-sdk) correctly handles: - Standard FIT messages (monitoring, hrv_status_summary, sleep_level etc.) - Garmin proprietary messages stored by numeric mesg_num - Unknown fields stored by field definition number - Scale/offset application, component expansion, HR merging Fenix 6X proprietary message numbers identified by binary analysis: 55 - activity accumulation snapshots (cumulative steps, HR per interval) 103 - daily totals summary (total steps, floors, calories) 211 - resting HR + HRV summary 227 - per-minute stress level + heart rate (most valuable for health dashboard) """ from datetime import datetime, timezone, timedelta, date from typing import Optional FIT_EPOCH_S = 631065600 # seconds between Unix epoch and FIT epoch (Dec 31 1989) def fit_ts(seconds) -> Optional[datetime]: """Convert FIT timestamp to UTC datetime.""" if seconds is None: return None try: s = int(seconds) 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 _is_datetime(v) -> bool: return isinstance(v, datetime) def parse_wellness_fit(file_path: str) -> dict: """ Parse a Garmin wellness/monitoring FIT file using the official Garmin SDK. Returns {"days": {date: metrics_dict}, "error": str|None} """ try: from garmin_fit_sdk import Decoder, Stream except ImportError: # Fall back to fitparse-based parser if SDK not installed yet from app.services.wellness_parser_fallback import parse_wellness_fit as _fb return _fb(file_path) daily = {} # date -> aggregation dict def ensure_day(d: date) -> dict: if d not in daily: daily[d] = { "heart_rates": [], "stress_values": [], "spo2_readings": [], "sleep_levels": [], "steps": None, "floors_climbed": None, "active_calories": None, "total_calories": None, "resting_hr": None, "hrv_nightly_avg": None, "hrv_5min_high": None, "hrv_status": None, } return daily[d] def get_date(msg: dict, *keys) -> Optional[date]: """Extract a date from a message, trying multiple field names.""" for key in keys: v = msg.get(key) if v is None: continue if _is_datetime(v): return v.date() if isinstance(v, (int, float)): dt = fit_ts(v) if dt: return dt.date() return None def listener(mesg_num: int, msg: dict): """Called for every message after full decoding.""" # ── Standard: monitoring (148) ──────────────────────────────────── if mesg_num == 148: d = get_date(msg, "timestamp", "local_timestamp") if not d: return 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)) # ── Standard: monitoring_info (147) ─────────────────────────────── elif mesg_num == 147: d = get_date(msg, "timestamp", "local_timestamp") if not d: return rhr = msg.get("resting_heart_rate") if rhr and 20 < rhr < 120: ensure_day(d)["resting_hr"] = int(rhr) # ── Standard: hrv_status_summary (275) ──────────────────────────── elif mesg_num == 275: d = get_date(msg, "timestamp") if not d: return entry = ensure_day(d) for key in ("weekly_average", "last_night_avg", "hrv_nightly_avg"): v = msg.get(key) if v: 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") if status: entry["hrv_status"] = str(status) # ── Standard: stress_level (132) ────────────────────────────────── elif mesg_num == 132: d = get_date(msg, "stress_level_time", "timestamp") if not d: return stress = msg.get("stress_level_value") if stress is not None and stress >= 0: ensure_day(d)["stress_values"].append(int(stress)) # ── Standard: spo2_data (258) ───────────────────────────────────── elif mesg_num == 258: d = get_date(msg, "timestamp") if not d: return spo2 = msg.get("spo2_percent") or msg.get("reading_spo2") if spo2 and 50 < spo2 <= 100: ensure_day(d)["spo2_readings"].append(float(spo2)) # ── Standard: sleep_level (269) ─────────────────────────────────── elif mesg_num == 269: d = get_date(msg, "timestamp") if not d: return level = msg.get("sleep_level") if level is not None: # Convert string level names to numeric codes if SDK decoded them if isinstance(level, str): level_map = {"unmeasurable": 0, "awake": 1, "light": 2, "deep": 3, "rem": 4} level = level_map.get(level.lower()) if level is not None: ensure_day(d)["sleep_levels"].append(int(level)) # ── Proprietary 227: per-minute stress + HR ─────────────────────── # field_1 = FIT timestamp, field_2 = heart rate bpm, field_0 = stress elif mesg_num == 227: # SDK stores unknown fields as "unknown_N" or by def_num ts_raw = msg.get(1) or msg.get("unknown_1") or msg.get("field_1") hr_raw = msg.get(2) or msg.get("unknown_2") or msg.get("field_2") stress_raw = msg.get(0) or msg.get("unknown_0") or msg.get("field_0") ts = fit_ts(ts_raw) if isinstance(ts_raw, (int, float)) else ( ts_raw if _is_datetime(ts_raw) else None ) if not ts: return entry = ensure_day(ts.date()) if hr_raw and isinstance(hr_raw, (int, float)) and 20 < hr_raw < 250: entry["heart_rates"].append(int(hr_raw)) if stress_raw is not None and isinstance(stress_raw, (int, float)) and stress_raw >= 0: entry["stress_values"].append(int(stress_raw)) # ── Proprietary 103: daily totals summary ───────────────────────── # field_253 = timestamp, field_3 = steps, field_4 = floors, field_5/7 = cal elif mesg_num == 103: ts_v = msg.get(253) or msg.get("timestamp") ts = ts_v if _is_datetime(ts_v) else fit_ts(ts_v) if not ts: return entry = ensure_day(ts.date()) steps = msg.get(3) if steps and isinstance(steps, (int, float)) and steps > 0: entry["steps"] = int(steps) floors = msg.get(4) if floors and isinstance(floors, (int, float)) and floors > 0: f = float(floors) if f > 1000: f = f / 100 entry["floors_climbed"] = round(f, 1) active_cal = msg.get(5) if active_cal and isinstance(active_cal, (int, float)) and active_cal > 0: entry["active_calories"] = float(active_cal) total_cal = msg.get(7) if total_cal and isinstance(total_cal, (int, float)) and total_cal > 0: entry["total_calories"] = float(total_cal) # ── Proprietary 211: resting HR + HRV summary ───────────────────── elif mesg_num == 211: ts_v = msg.get(253) or msg.get("timestamp") ts = ts_v if _is_datetime(ts_v) else fit_ts(ts_v) if not ts: return entry = ensure_day(ts.date()) rhr = msg.get(0) if rhr and isinstance(rhr, (int, float)) and 20 < rhr < 120: entry["resting_hr"] = int(rhr) hrv = msg.get(1) if hrv and isinstance(hrv, (int, float)) and 5 < hrv < 300: entry["hrv_nightly_avg"] = float(hrv) # ── Proprietary 55: activity accumulation snapshots ─────────────── elif mesg_num == 55: ts_v = msg.get(253) or msg.get("timestamp") ts = ts_v if _is_datetime(ts_v) else fit_ts(ts_v) if not ts: return entry = ensure_day(ts.date()) steps = msg.get(2) if steps and isinstance(steps, (int, float)) and steps > 0: entry["steps"] = max(entry["steps"] or 0, int(steps)) hr = msg.get(19) if hr and isinstance(hr, (int, float)) and 20 < hr < 250: entry["heart_rates"].append(int(hr)) # Decode the file 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, # wellness files sometimes have bad CRCs expand_sub_fields=True, expand_components=True, merge_heart_rates=True, mesg_listener=listener, ) except Exception as e: return {"error": str(e), "days": {}} # Aggregate per-day result = {} for day_date, data in daily.items(): hrs = data.pop("heart_rates", []) stresses = data.pop("stress_values", []) spo2s = data.pop("spo2_readings", []) sleep_levels = data.pop("sleep_levels", []) 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 # Sleep stage seconds (each level record = 30s epoch) if sleep_levels: sleep_deep_s = sum(30 for l in sleep_levels if l == 3) or None sleep_light_s = sum(30 for l in sleep_levels if l == 2) or None sleep_rem_s = sum(30 for l in sleep_levels if l == 4) or None sleep_awake_s = sum(30 for l in sleep_levels if l == 1) or None sleep_duration_s = (sleep_deep_s or 0) + (sleep_light_s or 0) + (sleep_rem_s or 0) or None else: sleep_deep_s = sleep_light_s = sleep_rem_s = sleep_awake_s = sleep_duration_s = 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": data.get("active_calories"), "total_calories": data.get("total_calories"), "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, } return {"days": result, "error": None}