Add VAM (climbing velocity) metric and per-duration curve
Extract pipeline now computes two VAM metrics per activity (cycling, running, hiking, walking): - climbing_vam_mh: VAM on ascending segments only, using 30 s forward lookahead to classify climbing vs. flat/descent (stored in detail JSON) - vam_curve: [[duration_s, vam_mh], ...] best VAM per standard duration (60 s – 1 h), sliding window on 30 s smoothed elevation, only windows with ≥ 10 m net gain count (stored in summary + detail) Athlete JSON aggregates vam_curve across all activities (all_time, last_365d, last_90d), same structure as power_curve. Frontend: - ActivityDetail shows "Climbing VAM" stat (grouped with elevation) - AthleteView adds a "VAM Curve" tab that appears only when the athlete has climbing data; renders VamChart (new component, mirrors MmpChart) vam_curve stripped from combined global feed; kept in user year shards for season-based on-the-fly aggregation in VamChart. Requires bincio reextract to backfill existing activities.
This commit is contained in:
Davide Scaini
@@ -14,6 +14,11 @@ 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]] = {
|
||||
"running": [0.4, 1.0, 1.609, 5.0, 10.0, 21.097, 42.195],
|
||||
@@ -62,6 +67,8 @@ 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], ...]
|
||||
|
||||
|
||||
def compute(activity: ParsedActivity) -> ComputedMetrics:
|
||||
@@ -81,6 +88,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)
|
||||
|
||||
return ComputedMetrics(
|
||||
distance_m=distance_m,
|
||||
@@ -102,6 +110,8 @@ def compute(activity: ParsedActivity) -> ComputedMetrics:
|
||||
mmp=mmp,
|
||||
best_efforts=best_efforts,
|
||||
best_climb_m=best_climb_m,
|
||||
climbing_vam_mh=climbing_vam_mh,
|
||||
vam_curve=vam_curve,
|
||||
)
|
||||
|
||||
|
||||
@@ -161,6 +171,114 @@ def compute_mmp(pts: list[DataPoint], started_at: datetime) -> Optional[list[lis
|
||||
return results if results else None
|
||||
|
||||
|
||||
# ── VAM (Velocità Ascensionale Media) ────────────────────────────────────────
|
||||
|
||||
def _rolling_mean_ele(data: list[float], win: int) -> list[float]:
|
||||
"""O(n) rolling mean via prefix sums."""
|
||||
n = len(data)
|
||||
prefix = [0.0] * (n + 1)
|
||||
for i, v in enumerate(data):
|
||||
prefix[i + 1] = prefix[i] + v
|
||||
half = win // 2
|
||||
result = []
|
||||
for i in range(n):
|
||||
lo = max(0, i - half)
|
||||
hi = min(n, i + half + 1)
|
||||
result.append((prefix[hi] - prefix[lo]) / (hi - lo))
|
||||
return result
|
||||
|
||||
|
||||
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.
|
||||
|
||||
Returns (climbing_vam_mh, vam_curve).
|
||||
climbing_vam_mh: VAM on ascending segments only (m/h), or None.
|
||||
vam_curve: [[duration_s, vam_mh], ...] best VAM per standard duration, or None.
|
||||
Only computed for cycling, running, hiking, walking.
|
||||
"""
|
||||
if sport not in _VAM_SPORTS:
|
||||
return None, None
|
||||
|
||||
# Build dense 1 Hz elevation array, forward-filling gaps
|
||||
sparse: dict[int, Optional[float]] = {}
|
||||
last_t = -1
|
||||
for p in pts:
|
||||
t = int((p.timestamp - started_at).total_seconds())
|
||||
if t < 0 or t == last_t:
|
||||
continue
|
||||
sparse[t] = p.elevation_m
|
||||
last_t = t
|
||||
|
||||
if not sparse:
|
||||
return None, None
|
||||
|
||||
t_min = min(sparse)
|
||||
t_max = max(sparse)
|
||||
if t_max - t_min > 7 * 24 * 3600:
|
||||
return None, None
|
||||
|
||||
ele_raw: list[Optional[float]] = []
|
||||
last_known: Optional[float] = None
|
||||
for t in range(t_min, t_max + 1):
|
||||
v = sparse.get(t)
|
||||
if v is not None:
|
||||
last_known = v
|
||||
ele_raw.append(last_known)
|
||||
|
||||
if sum(1 for e in ele_raw if e is not None) < 60:
|
||||
return None, None
|
||||
|
||||
first_valid = next((e for e in ele_raw if e is not None), None)
|
||||
if first_valid is None:
|
||||
return None, None
|
||||
ele_1hz: list[float] = [e if e is not None else first_valid for e in ele_raw]
|
||||
|
||||
n = len(ele_1hz)
|
||||
ele_smooth = _rolling_mean_ele(ele_1hz, 30)
|
||||
|
||||
# VAM curve: sliding window per duration, only windows with net gain above threshold
|
||||
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).
|
||||
_LOOK = 30
|
||||
_THRESH = 2.0
|
||||
climbing_gain = 0.0
|
||||
climbing_time = 0
|
||||
for i in range(n - 1):
|
||||
look = min(i + _LOOK, n - 1)
|
||||
if ele_smooth[look] - ele_smooth[i] >= _THRESH:
|
||||
inst = ele_smooth[i + 1] - ele_smooth[i]
|
||||
if inst > 0:
|
||||
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
|
||||
|
||||
|
||||
# ── best efforts & best climb ─────────────────────────────────────────────────
|
||||
|
||||
def compute_best_efforts(
|
||||
@@ -524,4 +642,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,
|
||||
)
|
||||
|
||||
@@ -101,6 +101,8 @@ def write_activity(
|
||||
"mmp": metrics.mmp,
|
||||
"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,
|
||||
@@ -257,6 +259,7 @@ def build_summary(
|
||||
"mmp": metrics.mmp,
|
||||
"best_efforts": metrics.best_efforts,
|
||||
"best_climb_m": metrics.best_climb_m,
|
||||
"vam_curve": metrics.vam_curve,
|
||||
"source": _infer_source(activity),
|
||||
"privacy": privacy,
|
||||
"detail_url": f"activities/{activity_id}.json",
|
||||
@@ -300,6 +303,25 @@ 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}
|
||||
@@ -368,6 +390,11 @@ 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