Fix wellness_parser - had fit_parser content instead of wellness parser
This commit is contained in:
@@ -1,284 +1,257 @@
|
|||||||
"""
|
"""
|
||||||
FIT and GPX file parser using the official Garmin FIT Python SDK.
|
Garmin wellness FIT file parser using the official Garmin FIT Python SDK.
|
||||||
Field names from the SDK are camelCase as per the SDK documentation.
|
SDK field names are camelCase as per the SDK documentation.
|
||||||
"""
|
"""
|
||||||
import math
|
from datetime import datetime, timezone, date
|
||||||
from datetime import datetime, timezone
|
|
||||||
from typing import Optional
|
from typing import Optional
|
||||||
import gpxpy
|
|
||||||
import polyline as polyline_lib
|
|
||||||
from garmin_fit_sdk import Decoder, Stream
|
from garmin_fit_sdk import Decoder, Stream
|
||||||
|
|
||||||
|
|
||||||
def haversine_distance(lat1, lon1, lat2, lon2) -> float:
|
FIT_EPOCH_S = 631065600
|
||||||
R = 6371000
|
|
||||||
phi1, phi2 = math.radians(lat1), math.radians(lat2)
|
|
||||||
dphi = math.radians(lat2 - lat1)
|
|
||||||
dlam = math.radians(lon2 - lon1)
|
|
||||||
a = math.sin(dphi/2)**2 + math.cos(phi1)*math.cos(phi2)*math.sin(dlam/2)**2
|
|
||||||
return 2 * R * math.asin(math.sqrt(a))
|
|
||||||
|
|
||||||
|
|
||||||
def _safe_float(val) -> Optional[float]:
|
def _fit_ts(raw) -> Optional[datetime]:
|
||||||
|
if raw is None:
|
||||||
|
return None
|
||||||
try:
|
try:
|
||||||
return float(val) if val is not None else None
|
s = int(raw)
|
||||||
except (TypeError, ValueError):
|
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
|
return None
|
||||||
|
|
||||||
|
|
||||||
def _bounding_box(coords: list) -> Optional[dict]:
|
def _to_date(val) -> Optional[date]:
|
||||||
if not coords:
|
if val is None:
|
||||||
return None
|
return None
|
||||||
lats = [c[0] for c in coords]
|
if isinstance(val, datetime):
|
||||||
lons = [c[1] for c in coords]
|
if val.tzinfo is None:
|
||||||
return {"min_lat": min(lats), "max_lat": max(lats),
|
val = val.replace(tzinfo=timezone.utc)
|
||||||
"min_lon": min(lons), "max_lon": max(lons)}
|
return val.date()
|
||||||
|
if isinstance(val, (int, float)):
|
||||||
|
dt = _fit_ts(val)
|
||||||
def _ensure_utc(dt) -> Optional[datetime]:
|
return dt.date() if dt else None
|
||||||
if dt is None:
|
|
||||||
return None
|
|
||||||
if isinstance(dt, datetime):
|
|
||||||
if dt.tzinfo is None:
|
|
||||||
return dt.replace(tzinfo=timezone.utc)
|
|
||||||
return dt
|
|
||||||
return None
|
return None
|
||||||
|
|
||||||
|
|
||||||
def parse_fit_file(filepath: str) -> dict:
|
def parse_wellness_fit(file_path: str) -> dict:
|
||||||
"""Parse a Garmin .fit activity file using the official Garmin SDK."""
|
"""
|
||||||
stream = Stream.from_file(filepath)
|
Parse a Garmin wellness/monitoring FIT file.
|
||||||
decoder = Decoder(stream)
|
Returns {"days": {date: metrics_dict}, "error": str|None}
|
||||||
|
"""
|
||||||
|
daily = {}
|
||||||
|
|
||||||
messages, errors = decoder.read(
|
def ensure_day(d: date) -> dict:
|
||||||
apply_scale_and_offset=True,
|
if d not in daily:
|
||||||
convert_datetimes_to_dates=True,
|
daily[d] = {
|
||||||
convert_types_to_strings=True,
|
"heart_rates": [],
|
||||||
enable_crc_check=False,
|
"stress_values": [],
|
||||||
expand_sub_fields=True,
|
"spo2_readings": [],
|
||||||
expand_components=True,
|
"sleep_levels": [],
|
||||||
merge_heart_rates=True,
|
"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]
|
||||||
|
|
||||||
# SDK returns camelCase keys
|
def listener(mesg_num: int, msg: dict):
|
||||||
sessions = messages.get("session", [{}])
|
|
||||||
session = sessions[0] if sessions else {}
|
|
||||||
records = messages.get("record", [])
|
|
||||||
laps = messages.get("lap", [])
|
|
||||||
|
|
||||||
sport = str(session.get("sport", "generic")).lower()
|
# monitoring_info (147)
|
||||||
sport_map = {
|
if mesg_num == 147:
|
||||||
"running": "running", "cycling": "cycling",
|
d = _to_date(msg.get("timestamp") or msg.get("localTimestamp"))
|
||||||
"hiking": "hiking", "walking": "walking",
|
rhr = msg.get("restingHeartRate")
|
||||||
"generic": "other", "trail_running": "running",
|
if d and rhr and 20 < rhr < 120:
|
||||||
"e_biking": "cycling", "open_water_swimming": "other",
|
ensure_day(d)["resting_hr"] = int(rhr)
|
||||||
}
|
|
||||||
sport_type = sport_map.get(sport, sport)
|
|
||||||
|
|
||||||
start_time = _ensure_utc(session.get("startTime"))
|
# monitoring (148)
|
||||||
|
elif mesg_num == 148:
|
||||||
|
d = _to_date(msg.get("timestamp") or msg.get("localTimestamp"))
|
||||||
|
if not d:
|
||||||
|
return
|
||||||
|
entry = ensure_day(d)
|
||||||
|
hr = msg.get("heartRate")
|
||||||
|
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("stressLevelValue")
|
||||||
|
if stress is not None and stress >= 0:
|
||||||
|
entry["stress_values"].append(int(stress))
|
||||||
|
|
||||||
coords = []
|
# hrv_status_summary (275)
|
||||||
for r in records:
|
elif mesg_num == 275:
|
||||||
lat = r.get("positionLat")
|
d = _to_date(msg.get("timestamp"))
|
||||||
lon = r.get("positionLong")
|
if not d:
|
||||||
if lat is not None and lon is not None:
|
return
|
||||||
if -90 <= lat <= 90 and -180 <= lon <= 180:
|
entry = ensure_day(d)
|
||||||
coords.append((lat, lon))
|
for key in ("weeklyAverage", "lastNightAvg", "hrvNightlyAvg"):
|
||||||
|
v = msg.get(key)
|
||||||
|
if v and v > 0:
|
||||||
|
entry["hrv_nightly_avg"] = float(v)
|
||||||
|
break
|
||||||
|
high = msg.get("lastNight5MinHigh")
|
||||||
|
if high:
|
||||||
|
entry["hrv_5min_high"] = float(high)
|
||||||
|
status = msg.get("hrvStatus")
|
||||||
|
if status:
|
||||||
|
entry["hrv_status"] = str(status)
|
||||||
|
|
||||||
encoded_polyline = polyline_lib.encode(coords) if coords else None
|
# stress_level (132)
|
||||||
bounding_box = _bounding_box(coords)
|
elif mesg_num == 132:
|
||||||
|
d = _to_date(msg.get("stressLevelTime") or msg.get("timestamp"))
|
||||||
|
if not d:
|
||||||
|
return
|
||||||
|
stress = msg.get("stressLevelValue")
|
||||||
|
if stress is not None and stress >= 0:
|
||||||
|
ensure_day(d)["stress_values"].append(int(stress))
|
||||||
|
|
||||||
normalized_points = []
|
# spo2_data (258)
|
||||||
for r in records:
|
elif mesg_num == 258:
|
||||||
ts = _ensure_utc(r.get("timestamp"))
|
d = _to_date(msg.get("timestamp"))
|
||||||
lat = r.get("positionLat")
|
if not d:
|
||||||
lon = r.get("positionLong")
|
return
|
||||||
|
spo2 = msg.get("spo2Percent") or msg.get("readingSpo2")
|
||||||
|
if spo2 and 50 < spo2 <= 100:
|
||||||
|
ensure_day(d)["spo2_readings"].append(float(spo2))
|
||||||
|
|
||||||
if lat is not None and not (-90 <= lat <= 90):
|
# sleep_level (269)
|
||||||
lat = None
|
elif mesg_num == 269:
|
||||||
if lon is not None and not (-180 <= lon <= 180):
|
d = _to_date(msg.get("timestamp"))
|
||||||
lon = None
|
if not d:
|
||||||
|
return
|
||||||
|
level = msg.get("sleepLevel")
|
||||||
|
if level is not None:
|
||||||
|
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))
|
||||||
|
|
||||||
normalized_points.append({
|
# Proprietary 227: per-minute stress + HR
|
||||||
"timestamp": ts.isoformat() if ts else None,
|
elif mesg_num == 227:
|
||||||
"latitude": _safe_float(lat),
|
ts_raw = msg.get(1) or msg.get("1")
|
||||||
"longitude": _safe_float(lon),
|
hr_raw = msg.get(2) or msg.get("2")
|
||||||
"altitude_m": _safe_float(r.get("altitude") or r.get("enhancedAltitude")),
|
stress_raw = msg.get(0) or msg.get("0")
|
||||||
"heart_rate": _safe_float(r.get("heartRate")),
|
d = _to_date(ts_raw)
|
||||||
"cadence": _safe_float(r.get("cadence")),
|
if not d:
|
||||||
"speed_ms": _safe_float(r.get("speed") or r.get("enhancedSpeed")),
|
return
|
||||||
"power": _safe_float(r.get("power")),
|
entry = ensure_day(d)
|
||||||
"temperature_c": _safe_float(r.get("temperature")),
|
if hr_raw and isinstance(hr_raw, (int, float)) and 20 < hr_raw < 250:
|
||||||
"distance_m": _safe_float(r.get("distance")),
|
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))
|
||||||
|
|
||||||
normalized_laps = []
|
# Proprietary 103: daily totals
|
||||||
for i, lap in enumerate(laps):
|
elif mesg_num == 103:
|
||||||
ls = _ensure_utc(lap.get("startTime"))
|
ts_raw = msg.get(253) or msg.get("253") or msg.get("timestamp")
|
||||||
normalized_laps.append({
|
d = _to_date(ts_raw)
|
||||||
"lap_number": i + 1,
|
if not d:
|
||||||
"start_time": ls.isoformat() if ls else None,
|
return
|
||||||
"duration_s": _safe_float(lap.get("totalElapsedTime")),
|
entry = ensure_day(d)
|
||||||
"distance_m": _safe_float(lap.get("totalDistance")),
|
steps = msg.get(3) or msg.get("3")
|
||||||
"avg_heart_rate": _safe_float(lap.get("avgHeartRate")),
|
if steps and isinstance(steps, (int, float)) and steps > 0:
|
||||||
"avg_cadence": _safe_float(lap.get("avgCadence")),
|
entry["steps"] = int(steps)
|
||||||
"avg_speed_ms": _safe_float(lap.get("avgSpeed") or lap.get("enhancedAvgSpeed")),
|
floors = msg.get(4) or msg.get("4")
|
||||||
"avg_power": _safe_float(lap.get("avgPower")),
|
if floors and isinstance(floors, (int, float)) and floors > 0:
|
||||||
})
|
f = float(floors)
|
||||||
|
entry["floors_climbed"] = round(f / 100 if f > 1000 else f, 1)
|
||||||
|
active_cal = msg.get(5) or 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) or msg.get("7")
|
||||||
|
if total_cal and isinstance(total_cal, (int, float)) and total_cal > 0:
|
||||||
|
entry["total_calories"] = float(total_cal)
|
||||||
|
|
||||||
name = sport_type.title()
|
# Proprietary 211: resting HR + HRV
|
||||||
if start_time:
|
elif mesg_num == 211:
|
||||||
name += " " + start_time.strftime("%Y-%m-%d")
|
ts_raw = msg.get(253) or msg.get("253") or msg.get("timestamp")
|
||||||
|
d = _to_date(ts_raw)
|
||||||
|
if not d:
|
||||||
|
return
|
||||||
|
entry = ensure_day(d)
|
||||||
|
rhr = msg.get(0) or msg.get("0")
|
||||||
|
if rhr and isinstance(rhr, (int, float)) and 20 < rhr < 120:
|
||||||
|
entry["resting_hr"] = int(rhr)
|
||||||
|
hrv = msg.get(1) or msg.get("1")
|
||||||
|
if hrv and isinstance(hrv, (int, float)) and 5 < hrv < 300:
|
||||||
|
entry["hrv_nightly_avg"] = float(hrv)
|
||||||
|
|
||||||
return {
|
# Proprietary 55: activity accumulations
|
||||||
"name": name,
|
elif mesg_num == 55:
|
||||||
"sport_type": sport_type,
|
ts_raw = msg.get(253) or msg.get("253") or msg.get("timestamp")
|
||||||
"start_time": start_time.isoformat() if start_time else None,
|
d = _to_date(ts_raw)
|
||||||
"distance_m": _safe_float(session.get("totalDistance")),
|
if not d:
|
||||||
"duration_s": _safe_float(session.get("totalElapsedTime")),
|
return
|
||||||
"elevation_gain_m": _safe_float(session.get("totalAscent")),
|
entry = ensure_day(d)
|
||||||
"elevation_loss_m": _safe_float(session.get("totalDescent")),
|
steps = msg.get(2) or msg.get("2")
|
||||||
"avg_heart_rate": _safe_float(session.get("avgHeartRate")),
|
if steps and isinstance(steps, (int, float)) and steps > 0:
|
||||||
"max_heart_rate": _safe_float(session.get("maxHeartRate")),
|
entry["steps"] = max(entry["steps"] or 0, int(steps))
|
||||||
"avg_cadence": _safe_float(session.get("avgCadence")),
|
hr = msg.get(19) or msg.get("19")
|
||||||
"avg_power": _safe_float(session.get("avgPower")),
|
if hr and isinstance(hr, (int, float)) and 20 < hr < 250:
|
||||||
"normalized_power": _safe_float(session.get("normalizedPower")),
|
entry["heart_rates"].append(int(hr))
|
||||||
"avg_speed_ms": _safe_float(session.get("avgSpeed") or session.get("enhancedAvgSpeed")),
|
|
||||||
"max_speed_ms": _safe_float(session.get("maxSpeed") or session.get("enhancedMaxSpeed")),
|
|
||||||
"avg_temperature_c": _safe_float(session.get("avgTemperature")),
|
|
||||||
"calories": _safe_float(session.get("totalCalories")),
|
|
||||||
"training_stress_score": _safe_float(session.get("trainingStressScore")),
|
|
||||||
"vo2max_estimate": _safe_float(session.get("totalTrainingEffect")),
|
|
||||||
"polyline": encoded_polyline,
|
|
||||||
"bounding_box": bounding_box,
|
|
||||||
"source_type": "fit",
|
|
||||||
"data_points": normalized_points,
|
|
||||||
"laps": normalized_laps,
|
|
||||||
}
|
|
||||||
|
|
||||||
|
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": {}}
|
||||||
|
|
||||||
def parse_gpx_file(filepath: str) -> dict:
|
result = {}
|
||||||
"""Parse a GPX file."""
|
for day_date, data in daily.items():
|
||||||
with open(filepath) as f:
|
hrs = data.pop("heart_rates", [])
|
||||||
gpx = gpxpy.parse(f)
|
stresses = data.pop("stress_values", [])
|
||||||
|
spo2s = data.pop("spo2_readings", [])
|
||||||
|
sleep_levels = data.pop("sleep_levels", [])
|
||||||
|
|
||||||
data_points = []
|
avg_hr = round(sum(hrs) / len(hrs), 1) if hrs else None
|
||||||
track = gpx.tracks[0] if gpx.tracks else None
|
max_hr = max(hrs) if hrs else None
|
||||||
if not track:
|
avg_stress = round(sum(s for s in stresses if s >= 0) / len(stresses), 1) if stresses else None
|
||||||
raise ValueError("No tracks found in GPX file")
|
spo2_avg = round(sum(spo2s) / len(spo2s), 1) if spo2s else None
|
||||||
|
|
||||||
for segment in track.segments:
|
if sleep_levels:
|
||||||
for pt in segment.points:
|
sleep_deep_s = sum(30 for l in sleep_levels if l == 3) or None
|
||||||
ts = pt.time
|
sleep_light_s = sum(30 for l in sleep_levels if l == 2) or None
|
||||||
if ts and ts.tzinfo is None:
|
sleep_rem_s = sum(30 for l in sleep_levels if l == 4) or None
|
||||||
ts = ts.replace(tzinfo=timezone.utc)
|
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
|
||||||
extensions = {}
|
|
||||||
if pt.extensions:
|
|
||||||
for ext in pt.extensions:
|
|
||||||
for child in ext:
|
|
||||||
tag = child.tag.split("}")[-1] if "}" in child.tag else child.tag
|
|
||||||
try:
|
|
||||||
extensions[tag] = float(child.text)
|
|
||||||
except (ValueError, TypeError):
|
|
||||||
pass
|
|
||||||
|
|
||||||
data_points.append({
|
|
||||||
"timestamp": ts.isoformat() if ts else None,
|
|
||||||
"latitude": pt.latitude,
|
|
||||||
"longitude": pt.longitude,
|
|
||||||
"altitude_m": pt.elevation,
|
|
||||||
"heart_rate": extensions.get("hr"),
|
|
||||||
"cadence": extensions.get("cad"),
|
|
||||||
"speed_ms": extensions.get("speed"),
|
|
||||||
"power": extensions.get("power"),
|
|
||||||
"temperature_c": extensions.get("temp") or extensions.get("atemp"),
|
|
||||||
"distance_m": None,
|
|
||||||
})
|
|
||||||
|
|
||||||
coords = [(p["latitude"], p["longitude"]) for p in data_points
|
|
||||||
if p["latitude"] and p["longitude"]]
|
|
||||||
encoded_polyline = polyline_lib.encode(coords) if coords else None
|
|
||||||
bounding_box = _bounding_box(coords)
|
|
||||||
|
|
||||||
total_dist = 0.0
|
|
||||||
prev = None
|
|
||||||
for p in data_points:
|
|
||||||
if p["latitude"] and p["longitude"]:
|
|
||||||
if prev:
|
|
||||||
total_dist += haversine_distance(prev[0], prev[1], p["latitude"], p["longitude"])
|
|
||||||
prev = (p["latitude"], p["longitude"])
|
|
||||||
p["distance_m"] = total_dist
|
|
||||||
|
|
||||||
uphill, downhill = 0.0, 0.0
|
|
||||||
alts = [p["altitude_m"] for p in data_points if p["altitude_m"]]
|
|
||||||
for i in range(1, len(alts)):
|
|
||||||
diff = alts[i] - alts[i-1]
|
|
||||||
if diff > 0:
|
|
||||||
uphill += diff
|
|
||||||
else:
|
else:
|
||||||
downhill += abs(diff)
|
sleep_deep_s = sleep_light_s = sleep_rem_s = sleep_awake_s = sleep_duration_s = None
|
||||||
|
|
||||||
hrs = [p["heart_rate"] for p in data_points if p["heart_rate"]]
|
result[day_date] = {
|
||||||
start_time_str = data_points[0]["timestamp"] if data_points else None
|
"resting_hr": data.get("resting_hr"),
|
||||||
start_dt = datetime.fromisoformat(start_time_str) if start_time_str else None
|
"avg_hr_day": avg_hr,
|
||||||
end_dt = datetime.fromisoformat(data_points[-1]["timestamp"]) if data_points else None
|
"max_hr_day": max_hr,
|
||||||
duration = (end_dt - start_dt).total_seconds() if (start_dt and end_dt) else None
|
"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,
|
||||||
|
}
|
||||||
|
|
||||||
sport = "running"
|
return {"days": result, "error": None}
|
||||||
if track.type:
|
|
||||||
sport = track.type.lower()
|
|
||||||
|
|
||||||
return {
|
|
||||||
"name": track.name or gpx.name or f"Activity {start_dt.date() if start_dt else ''}",
|
|
||||||
"sport_type": sport,
|
|
||||||
"start_time": start_time_str,
|
|
||||||
"distance_m": total_dist,
|
|
||||||
"duration_s": duration,
|
|
||||||
"elevation_gain_m": uphill,
|
|
||||||
"elevation_loss_m": downhill,
|
|
||||||
"avg_heart_rate": (sum(hrs) / len(hrs)) if hrs else None,
|
|
||||||
"max_heart_rate": max(hrs) if hrs else None,
|
|
||||||
"avg_cadence": None,
|
|
||||||
"avg_power": None,
|
|
||||||
"normalized_power": None,
|
|
||||||
"avg_speed_ms": (total_dist / duration) if (total_dist and duration) else None,
|
|
||||||
"max_speed_ms": None,
|
|
||||||
"avg_temperature_c": None,
|
|
||||||
"calories": None,
|
|
||||||
"training_stress_score": None,
|
|
||||||
"vo2max_estimate": None,
|
|
||||||
"polyline": encoded_polyline,
|
|
||||||
"bounding_box": bounding_box,
|
|
||||||
"source_type": "gpx",
|
|
||||||
"data_points": data_points,
|
|
||||||
"laps": [],
|
|
||||||
}
|
|
||||||
|
|
||||||
|
|
||||||
def calculate_hr_zones(data_points: list, user_max_hr: float) -> dict:
|
|
||||||
"""Calculate % time in each HR zone using user's configured max HR."""
|
|
||||||
if not user_max_hr or user_max_hr < 100:
|
|
||||||
return {}
|
|
||||||
|
|
||||||
zone_bounds = [0.0, 0.60, 0.70, 0.80, 0.90, 1.01]
|
|
||||||
zone_keys = ["z1", "z2", "z3", "z4", "z5"]
|
|
||||||
zones = {k: 0 for k in zone_keys}
|
|
||||||
total = 0
|
|
||||||
|
|
||||||
for p in data_points:
|
|
||||||
hr = p.get("heart_rate")
|
|
||||||
if not hr or hr < 20:
|
|
||||||
continue
|
|
||||||
pct = hr / user_max_hr
|
|
||||||
total += 1
|
|
||||||
for i, key in enumerate(zone_keys):
|
|
||||||
if zone_bounds[i] <= pct < zone_bounds[i+1]:
|
|
||||||
zones[key] += 1
|
|
||||||
break
|
|
||||||
else:
|
|
||||||
zones["z5"] += 1
|
|
||||||
|
|
||||||
if total:
|
|
||||||
return {k: round(v / total * 100, 1) for k, v in zones.items()}
|
|
||||||
return {}
|
|
||||||
Reference in New Issue
Block a user