VAM: drop duration curve, show avg climbing VAM in Nerd Corner
Remove the per-duration VAM curve everywhere (metrics, summaries, detail JSON, athlete.json, VamChart.svelte, AthleteView VAM tab). Keep only climbing_vam_mh per activity. Add it to activity summaries so NerdCorner can plot average climbing VAM per week/month year-over-year alongside distance/elevation/time. Add --backfill-vam-summary flag to copy the field from existing detail JSONs into index.json without re-extracting.
This commit is contained in:
Davide Scaini
+18
-59
@@ -14,10 +14,7 @@ from bincio.extract.models import DataPoint, ParsedActivity
|
||||
# Standard MMP durations (seconds). Log-spaced so the curve looks good on a log-x axis.
|
||||
MMP_DURATIONS_S = [1, 2, 5, 10, 15, 20, 30, 60, 120, 180, 300, 600, 1200, 1800, 3600]
|
||||
|
||||
# VAM curve durations — start at 60 s (shorter windows are too noisy for elevation data).
|
||||
VAM_DURATIONS_S = [60, 120, 180, 300, 600, 1200, 1800, 3600]
|
||||
_VAM_SPORTS = frozenset({"cycling", "running", "hiking", "walking"})
|
||||
_MIN_CLIMB_GAIN_M = 10.0 # minimum net gain in a window for VAM to be meaningful
|
||||
|
||||
# Standard best-effort distances (km) per sport.
|
||||
BEST_EFFORT_DISTANCES: dict[str, list[float]] = {
|
||||
@@ -67,8 +64,7 @@ class ComputedMetrics:
|
||||
# [[distance_km, time_s], ...] sorted by distance — None if sport has no distance targets
|
||||
best_efforts: Optional[list[list[float]]]
|
||||
best_climb_m: Optional[float] # max net elevation gain in one contiguous window (cycling only)
|
||||
climbing_vam_mh: Optional[int] # VAM on ascending segments only (m/h)
|
||||
vam_curve: Optional[list[list[int]]] # [[duration_s, vam_mh], ...]
|
||||
climbing_vam_mh: Optional[int] # average VAM on ascending segments only (m/h)
|
||||
|
||||
|
||||
def compute(activity: ParsedActivity) -> ComputedMetrics:
|
||||
@@ -88,7 +84,7 @@ def compute(activity: ParsedActivity) -> ComputedMetrics:
|
||||
start_ll, end_ll = _endpoints(pts)
|
||||
mmp = compute_mmp(pts, activity.started_at)
|
||||
best_efforts, best_climb_m = compute_best_efforts(pts, activity.started_at, activity.sport)
|
||||
climbing_vam_mh, vam_curve = compute_vam(pts, activity.started_at, activity.sport)
|
||||
climbing_vam_mh = compute_vam(pts, activity.started_at, activity.sport)
|
||||
|
||||
return ComputedMetrics(
|
||||
distance_m=distance_m,
|
||||
@@ -111,7 +107,6 @@ def compute(activity: ParsedActivity) -> ComputedMetrics:
|
||||
best_efforts=best_efforts,
|
||||
best_climb_m=best_climb_m,
|
||||
climbing_vam_mh=climbing_vam_mh,
|
||||
vam_curve=vam_curve,
|
||||
)
|
||||
|
||||
|
||||
@@ -188,33 +183,18 @@ def _rolling_mean_ele(data: list[float], win: int) -> list[float]:
|
||||
return result
|
||||
|
||||
|
||||
def _vam_from_ele_1hz(ele_1hz: list[float]) -> tuple[Optional[int], Optional[list[list[int]]]]:
|
||||
"""Core VAM computation from a dense 1 Hz elevation array."""
|
||||
def _vam_from_ele_1hz(ele_1hz: list[float]) -> Optional[int]:
|
||||
"""Climbing VAM from a dense 1 Hz elevation array.
|
||||
|
||||
Accumulates gain and time only on ascending seconds, identified by a 30 s
|
||||
forward-lookahead on the smoothed elevation signal.
|
||||
Returns climbing_vam_mh (m/h), or None when there is too little climbing data.
|
||||
"""
|
||||
n = len(ele_1hz)
|
||||
if n < 60:
|
||||
return None, None
|
||||
return None
|
||||
ele_smooth = _rolling_mean_ele(ele_1hz, 30)
|
||||
|
||||
# VAM curve: best VAM per standard duration, windows with net gain ≥ threshold only
|
||||
vam_results: list[list[int]] = []
|
||||
for d in VAM_DURATIONS_S:
|
||||
if d >= n:
|
||||
break
|
||||
best_vam: Optional[float] = None
|
||||
for i in range(n - d):
|
||||
net_gain = ele_smooth[i + d] - ele_smooth[i]
|
||||
if net_gain < _MIN_CLIMB_GAIN_M:
|
||||
continue
|
||||
vam = net_gain * 3600.0 / d
|
||||
if best_vam is None or vam > best_vam:
|
||||
best_vam = vam
|
||||
if best_vam is not None:
|
||||
vam_results.append([d, round(best_vam)])
|
||||
vam_curve: Optional[list[list[int]]] = vam_results if vam_results else None
|
||||
|
||||
# Climbing VAM: accumulate gain and time only on ascending seconds.
|
||||
# A second is climbing if the 30 s forward elevation gain exceeds 2 m
|
||||
# (roughly 1 % gradient at 7 km/h).
|
||||
climbing_gain = 0.0
|
||||
climbing_time = 0
|
||||
for i in range(n - 1):
|
||||
@@ -225,11 +205,9 @@ def _vam_from_ele_1hz(ele_1hz: list[float]) -> tuple[Optional[int], Optional[lis
|
||||
climbing_gain += inst
|
||||
climbing_time += 1
|
||||
|
||||
climbing_vam_mh: Optional[int] = None
|
||||
if climbing_time >= 60 and climbing_gain >= 5.0:
|
||||
climbing_vam_mh = round(climbing_gain * 3600.0 / climbing_time)
|
||||
|
||||
return climbing_vam_mh, vam_curve
|
||||
return round(climbing_gain * 3600.0 / climbing_time)
|
||||
return None
|
||||
|
||||
|
||||
def _build_ele_1hz(sparse: dict[int, Optional[float]]) -> Optional[list[float]]:
|
||||
@@ -255,18 +233,14 @@ def _build_ele_1hz(sparse: dict[int, Optional[float]]) -> Optional[list[float]]:
|
||||
return [e if e is not None else first_valid for e in ele_raw]
|
||||
|
||||
|
||||
def compute_vam(
|
||||
pts: list[DataPoint],
|
||||
started_at: datetime,
|
||||
sport: str,
|
||||
) -> tuple[Optional[int], Optional[list[list[int]]]]:
|
||||
"""Compute climbing VAM and VAM duration curve from DataPoints.
|
||||
def compute_vam(pts: list[DataPoint], started_at: datetime, sport: str) -> Optional[int]:
|
||||
"""Compute average climbing VAM (m/h) from DataPoints.
|
||||
|
||||
Returns (climbing_vam_mh, vam_curve).
|
||||
Only computed for cycling, running, hiking, walking.
|
||||
Returns None when the activity has insufficient climbing data.
|
||||
"""
|
||||
if sport not in _VAM_SPORTS:
|
||||
return None, None
|
||||
return None
|
||||
sparse: dict[int, Optional[float]] = {}
|
||||
last_t = -1
|
||||
for p in pts:
|
||||
@@ -277,22 +251,7 @@ def compute_vam(
|
||||
last_t = t
|
||||
ele_1hz = _build_ele_1hz(sparse)
|
||||
if ele_1hz is None:
|
||||
return None, None
|
||||
return _vam_from_ele_1hz(ele_1hz)
|
||||
|
||||
|
||||
def compute_vam_from_timeseries(ts: dict, sport: str) -> tuple[Optional[int], Optional[list[list[int]]]]:
|
||||
"""Compute VAM from a stored timeseries dict (used for backfill without re-parsing files)."""
|
||||
if sport not in _VAM_SPORTS:
|
||||
return None, None
|
||||
t_vals = ts.get("t") or []
|
||||
ele_vals = ts.get("elevation_m") or []
|
||||
if not t_vals or not ele_vals:
|
||||
return None, None
|
||||
sparse: dict[int, Optional[float]] = {int(t): e for t, e in zip(t_vals, ele_vals)}
|
||||
ele_1hz = _build_ele_1hz(sparse)
|
||||
if ele_1hz is None:
|
||||
return None, None
|
||||
return None
|
||||
return _vam_from_ele_1hz(ele_1hz)
|
||||
|
||||
|
||||
@@ -659,5 +618,5 @@ def _empty() -> ComputedMetrics:
|
||||
avg_cadence_rpm=None, avg_power_w=None, np_power_w=None, max_power_w=None,
|
||||
bbox=None, start_latlng=None, end_latlng=None,
|
||||
mmp=None, best_efforts=None, best_climb_m=None,
|
||||
climbing_vam_mh=None, vam_curve=None,
|
||||
climbing_vam_mh=None,
|
||||
)
|
||||
|
||||
@@ -102,7 +102,6 @@ def write_activity(
|
||||
"best_efforts": metrics.best_efforts,
|
||||
"best_climb_m": metrics.best_climb_m,
|
||||
"climbing_vam_mh": metrics.climbing_vam_mh,
|
||||
"vam_curve": metrics.vam_curve,
|
||||
"laps": [_serialise_lap(lap) for lap in activity.laps],
|
||||
"timeseries_url": f"activities/{activity_id}.timeseries.json" if timeseries else None,
|
||||
"source": source,
|
||||
@@ -259,7 +258,7 @@ def build_summary(
|
||||
"mmp": metrics.mmp,
|
||||
"best_efforts": metrics.best_efforts,
|
||||
"best_climb_m": metrics.best_climb_m,
|
||||
"vam_curve": metrics.vam_curve,
|
||||
"climbing_vam_mh": metrics.climbing_vam_mh,
|
||||
"source": _infer_source(activity),
|
||||
"privacy": privacy,
|
||||
"detail_url": f"activities/{activity_id}.json",
|
||||
@@ -303,25 +302,6 @@ def write_athlete_json(summaries: list[dict], output_dir: Path, athlete_config:
|
||||
mmps_365 = [s["mmp"] for s in summaries if s.get("mmp") and _is_outdoor(s) and s["started_at"] >= cutoff_365]
|
||||
mmps_90 = [s["mmp"] for s in summaries if s.get("mmp") and _is_outdoor(s) and s["started_at"] >= cutoff_90]
|
||||
|
||||
# ── VAM curve aggregation ─────────────────────────────────────────────────
|
||||
|
||||
def _merge_vam_curves(vam_lists: list[list[list[int]]]) -> list[list[int]]:
|
||||
best: dict[int, int] = {}
|
||||
for vc in vam_lists:
|
||||
for d, v in vc:
|
||||
if d not in best or v > best[d]:
|
||||
best[d] = v
|
||||
return [[d, v] for d, v in sorted(best.items())]
|
||||
|
||||
_VAM_SPORTS = {"cycling", "running", "hiking", "walking"}
|
||||
|
||||
def _has_vam(s: dict) -> bool:
|
||||
return bool(s.get("vam_curve")) and s.get("sport") in _VAM_SPORTS and _is_outdoor(s)
|
||||
|
||||
all_vams = [s["vam_curve"] for s in summaries if _has_vam(s)]
|
||||
vams_365 = [s["vam_curve"] for s in summaries if _has_vam(s) and s["started_at"] >= cutoff_365]
|
||||
vams_90 = [s["vam_curve"] for s in summaries if _has_vam(s) and s["started_at"] >= cutoff_90]
|
||||
|
||||
# ── Personal records aggregation ──────────────────────────────────────────
|
||||
# records[sport][distance_km] = {time_s, activity_id, started_at, title}
|
||||
# best_climb[activity_id] = {climb_m, started_at, title}
|
||||
@@ -390,11 +370,6 @@ def write_athlete_json(summaries: list[dict], output_dir: Path, athlete_config:
|
||||
"last_365d": _merge_mmps(mmps_365) if mmps_365 else None,
|
||||
"last_90d": _merge_mmps(mmps_90) if mmps_90 else None,
|
||||
},
|
||||
"vam_curve": {
|
||||
"all_time": _merge_vam_curves(all_vams) if all_vams else None,
|
||||
"last_365d": _merge_vam_curves(vams_365) if vams_365 else None,
|
||||
"last_90d": _merge_vam_curves(vams_90) if vams_90 else None,
|
||||
},
|
||||
"records": {
|
||||
sport: _serialise_sport_records(records[sport])
|
||||
for sport in SPORTS
|
||||
|
||||
Reference in New Issue
Block a user