diff --git a/bincio/extract/metrics.py b/bincio/extract/metrics.py index d464321..280b8ab 100644 --- a/bincio/extract/metrics.py +++ b/bincio/extract/metrics.py @@ -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( diff --git a/bincio/render/cli.py b/bincio/render/cli.py index a4b75d9..c1d8be0 100644 --- a/bincio/render/cli.py +++ b/bincio/render/cli.py @@ -377,6 +377,58 @@ def _link_data(site: Path, data: Path) -> None: console.print(f"Linked data: [cyan]{target}[/cyan] → [cyan]{public_data}[/cyan]") +def _recompute_vam(data: Path, handle: str | None = None) -> None: + """Recompute climbing_vam_mh and vam_curve for all activities from stored timeseries.""" + import json + from bincio.extract.metrics import compute_vam_from_timeseries + + targets = [data / handle] if handle else _user_dirs(data) + for user_dir in targets: + acts_dir = user_dir / "activities" + index_path = user_dir / "index.json" + if not acts_dir.exists() or not index_path.exists(): + continue + try: + index_data = json.loads(index_path.read_text(encoding="utf-8")) + except Exception: + continue + + updated = 0 + for act_path in acts_dir.glob("*.json"): + if act_path.stem.endswith((".timeseries", ".geojson")): + continue + ts_path = acts_dir / f"{act_path.stem}.timeseries.json" + if not ts_path.exists(): + continue + try: + detail = json.loads(act_path.read_text(encoding="utf-8")) + sport = detail.get("sport", "other") + ts = json.loads(ts_path.read_text(encoding="utf-8")) + new_vam, new_curve = compute_vam_from_timeseries(ts, sport) + if (new_vam == detail.get("climbing_vam_mh") + and new_curve == detail.get("vam_curve")): + continue + detail["climbing_vam_mh"] = new_vam + detail["vam_curve"] = new_curve + act_path.write_text( + json.dumps(detail, indent=2, ensure_ascii=False), encoding="utf-8" + ) + act_id = act_path.stem + for s in index_data.get("activities", []): + if s.get("id") == act_id: + s["vam_curve"] = new_curve + break + updated += 1 + except Exception: + pass + + if updated: + index_path.write_text( + json.dumps(index_data, indent=2, ensure_ascii=False), encoding="utf-8" + ) + console.print(f" [cyan]{user_dir.name}[/cyan]: {updated} activity(ies) updated") + + @click.command() @click.option("--config", "config_path", default=None, help="Path to extract_config.yaml (reads output.dir from it).") @@ -400,6 +452,9 @@ def _link_data(site: Path, data: Path) -> None: @click.option("--recompute-elevation", "recompute_elevation", is_flag=True, help="Recompute elevation_gain_m/loss_m for all activities from stored timeseries " "(run once after upgrading the dropout-skip fix).") +@click.option("--recompute-vam", "recompute_vam", is_flag=True, + help="Recompute climbing_vam_mh and vam_curve for all activities from stored " + "timeseries (run once after adding VAM support).") def render( config_path: Optional[str], data_dir: Optional[str], @@ -411,6 +466,7 @@ def render( no_build: bool, recompute_climbs: bool, recompute_elevation: bool, + recompute_vam: bool, ) -> None: """Build (or serve) the BincioActivity static site from a BAS data store.""" @@ -428,6 +484,10 @@ def render( console.print("Recomputing elevation gain/loss from timeseries…") _recompute_elevation(data, handle=handle) + if recompute_vam: + console.print("Recomputing VAM from timeseries…") + _recompute_vam(data, handle=handle) + _merge_edits(data, handle=handle) _rebuild_athlete_json(data, handle=handle) _bake_tracks(data, handle=handle)