VAM: add --recompute-vam flag and compute_vam_from_timeseries helper
Refactors core VAM logic into _vam_from_ele_1hz() and _build_ele_1hz() so both the DataPoint-based extract path and the timeseries-based backfill path share the same implementation. render --recompute-vam reads stored *.timeseries.json files and updates climbing_vam_mh + vam_curve in activities/*.json and index.json in-place, without re-parsing the original FIT/GPX files.
This commit is contained in:
Davide Scaini
+71
-54
@@ -188,59 +188,14 @@ def _rolling_mean_ele(data: list[float], win: int) -> list[float]:
|
||||
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]
|
||||
|
||||
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."""
|
||||
n = len(ele_1hz)
|
||||
if n < 60:
|
||||
return None, None
|
||||
ele_smooth = _rolling_mean_ele(ele_1hz, 30)
|
||||
|
||||
# VAM curve: sliding window per duration, only windows with net gain above threshold
|
||||
# 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:
|
||||
@@ -260,13 +215,11 @@ def compute_vam(
|
||||
# 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:
|
||||
look = min(i + 30, n - 1)
|
||||
if ele_smooth[look] - ele_smooth[i] >= 2.0:
|
||||
inst = ele_smooth[i + 1] - ele_smooth[i]
|
||||
if inst > 0:
|
||||
climbing_gain += inst
|
||||
@@ -279,6 +232,70 @@ def compute_vam(
|
||||
return climbing_vam_mh, vam_curve
|
||||
|
||||
|
||||
def _build_ele_1hz(sparse: dict[int, Optional[float]]) -> Optional[list[float]]:
|
||||
"""Build a dense 1 Hz elevation array from a {t: ele} sparse dict, forward-filling gaps."""
|
||||
if not sparse:
|
||||
return None
|
||||
t_min = min(sparse)
|
||||
t_max = max(sparse)
|
||||
if t_max - t_min > 7 * 24 * 3600:
|
||||
return 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
|
||||
first_valid = next((e for e in ele_raw if e is not None), None)
|
||||
if first_valid is None:
|
||||
return None
|
||||
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.
|
||||
|
||||
Returns (climbing_vam_mh, vam_curve).
|
||||
Only computed for cycling, running, hiking, walking.
|
||||
"""
|
||||
if sport not in _VAM_SPORTS:
|
||||
return None, None
|
||||
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
|
||||
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 _vam_from_ele_1hz(ele_1hz)
|
||||
|
||||
|
||||
# ── best efforts & best climb ─────────────────────────────────────────────────
|
||||
|
||||
def compute_best_efforts(
|
||||
|
||||
Reference in New Issue
Block a user