brut/bincio-activity
1
0
Fork 0
Code Issues Pull Requests Actions 6 Packages Projects Releases Wiki Activity

personal records tab into athlete page

Browse Source
This commit is contained in:
Davide Scaini
2026-03-30 10:53:51 +02:00
parent 2cc53dece4
commit a6a81f9421
6 changed files with 692 additions and 43 deletions
Download Patch File Download Diff File Expand all files Collapse all files
bincio/extract/metrics.py
+108 -1
View File
@@ -14,6 +14,17 @@ 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]
# 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],
"cycling": [5.0, 10.0, 20.0, 50.0, 100.0],
"swimming": [0.1, 0.2, 0.5, 1.0, 2.0],
"hiking": [], # no sliding-window records; aggregate from summaries only
"walking": [],
"skiing": [],
"other": [],
}
# Speed below which we consider the athlete stopped (km/h)
_STOPPED_THRESHOLD_KMH = 1.0
_EARTH_R = 6_371_000.0 # metres
@@ -47,6 +58,9 @@ class ComputedMetrics:
start_latlng: Optional[tuple[float, float]]
end_latlng: Optional[tuple[float, float]]
mmp: Optional[list[list[int]]] # [[duration_s, avg_watts], ...] — None if no power data
# [[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)
def compute(activity: ParsedActivity) -> ComputedMetrics:
@@ -64,6 +78,7 @@ def compute(activity: ParsedActivity) -> ComputedMetrics:
bbox = _bbox(pts)
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)
return ComputedMetrics(
distance_m=distance_m,
@@ -82,6 +97,8 @@ def compute(activity: ParsedActivity) -> ComputedMetrics:
start_latlng=start_ll,
end_latlng=end_ll,
mmp=mmp,
best_efforts=best_efforts,
best_climb_m=best_climb_m,
)
@@ -132,6 +149,96 @@ def compute_mmp(pts: list[DataPoint], started_at: datetime) -> Optional[list[lis
return results if results else None
# ── best efforts & best climb ─────────────────────────────────────────────────
def compute_best_efforts(
pts: list[DataPoint],
started_at: datetime,
sport: str,
) -> tuple[Optional[list[list[float]]], Optional[float]]:
"""Return (best_efforts, best_climb_m) for this activity.
best_efforts: [[distance_km, time_s], ...] — one entry per target distance
where the activity was long enough to contain that effort.
best_climb_m: maximum net elevation gain over any contiguous window (cycling).
Both use the same 1 Hz downsampled series as the timeseries writer.
"""
targets = BEST_EFFORT_DISTANCES.get(sport, [])
# Build 1 Hz speed (km/h) and elevation (m) arrays — same downsampling as timeseries.py
speed_1hz: list[float] = []
ele_1hz: list[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
last_t = t
speed_1hz.append(p.speed_kmh if p.speed_kmh is not None else 0.0)
ele_1hz.append(p.elevation_m)
best_efforts: Optional[list[list[float]]] = None
if targets and speed_1hz:
results = []
for d_km in targets:
t_s = _fastest_time_for_distance(speed_1hz, d_km)
if t_s is not None:
results.append([d_km, t_s])
best_efforts = results if results else None
best_climb_m: Optional[float] = None
if sport == "cycling":
best_climb_m = _best_climb(ele_1hz)
return best_efforts, best_climb_m
def _fastest_time_for_distance(speed_1hz: list[float], target_km: float) -> Optional[int]:
"""Minimum number of seconds to cover target_km using a two-pointer sliding window.
Each sample contributes speed_kmh / 3600 km (one second at that speed).
Nulls/zeros extend the window without adding distance — naturally deprioritised.
"""
n = len(speed_1hz)
left = 0
window_dist = 0.0
best_s: Optional[int] = None
for right in range(n):
window_dist += speed_1hz[right] / 3600.0
# Shrink from the left while we still cover the target
while window_dist >= target_km and left <= right:
window_s = right - left + 1
if best_s is None or window_s < best_s:
best_s = window_s
window_dist -= speed_1hz[left] / 3600.0
left += 1
return best_s
def _best_climb(ele_1hz: list[Optional[float]]) -> Optional[float]:
"""Maximum net elevation gain over any contiguous window (Kadane's on deltas).
Ignores samples where elevation is None. Returns None if fewer than two
valid elevation samples exist.
"""
valid = [e for e in ele_1hz if e is not None]
if len(valid) < 2:
return None
max_gain = 0.0
current = 0.0
for a, b in zip(valid, valid[1:]):
current = max(0.0, current + (b - a))
if current > max_gain:
max_gain = current
return round(max_gain, 1) if max_gain > 0 else None
# ── single-pass GPS stats ──────────────────────────────────────────────────────
# distance, moving time, avg speed, and max speed are all derived from the same
# per-segment loop, so we compute them in one pass instead of four.
@@ -263,5 +370,5 @@ def _empty() -> ComputedMetrics:
avg_hr_bpm=None, max_hr_bpm=None,
avg_cadence_rpm=None, avg_power_w=None, max_power_w=None,
bbox=None, start_latlng=None, end_latlng=None,
mmp=None,
mmp=None, best_efforts=None, best_climb_m=None,
)
bincio/extract/writer.py
+77 -17
View File
@@ -69,6 +69,8 @@ def write_activity(
"start_latlng": list(metrics.start_latlng) if metrics.start_latlng else None,
"end_latlng": list(metrics.end_latlng) if metrics.end_latlng else None,
"mmp": metrics.mmp,
"best_efforts": metrics.best_efforts,
"best_climb_m": metrics.best_climb_m,
"laps": [_serialise_lap(lap) for lap in activity.laps],
"timeseries": build_timeseries(activity.points, activity.started_at, privacy),
"source": source,
@@ -117,6 +119,8 @@ def build_summary(
"avg_cadence_rpm": metrics.avg_cadence_rpm,
"avg_power_w": metrics.avg_power_w,
"mmp": metrics.mmp,
"best_efforts": metrics.best_efforts,
"best_climb_m": metrics.best_climb_m,
"source": _infer_source(activity),
"privacy": privacy,
"detail_url": f"activities/{activity_id}.json",
@@ -127,16 +131,7 @@ def build_summary(
def write_athlete_json(summaries: list[dict], output_dir: Path, athlete_config: dict) -> None:
"""Aggregate per-activity MMP curves into athlete.json.
Computes element-wise max MMP for:
- all_time
- last_365d
- last_90d
The site reads this single file for the athlete/power-curve page.
Per-activity mmp is already in each summary for client-side season filtering.
"""
"""Aggregate per-activity MMP curves and personal records into athlete.json."""
from datetime import datetime, timezone
now = datetime.now(timezone.utc)
@@ -148,8 +143,9 @@ def write_athlete_json(summaries: list[dict], output_dir: Path, athlete_config:
cutoff_365 = _cutoff_iso(365)
cutoff_90 = _cutoff_iso(90)
# ── MMP aggregation ───────────────────────────────────────────────────────
def _merge_mmps(activity_mmps: list[list[list[int]]]) -> list[list[int]]:
"""Element-wise max across a list of mmp arrays."""
best: dict[int, int] = {}
for mmp in activity_mmps:
for d, w in mmp:
@@ -157,18 +153,82 @@ def write_athlete_json(summaries: list[dict], output_dir: Path, athlete_config:
best[d] = w
return [[d, w] for d, w in sorted(best.items())]
all_mmps = [s["mmp"] for s in summaries if s.get("mmp")]
mmps_365 = [s["mmp"] for s in summaries if s.get("mmp") and s["started_at"] >= cutoff_365]
mmps_90 = [s["mmp"] for s in summaries if s.get("mmp") and s["started_at"] >= cutoff_90]
all_mmps = [s["mmp"] for s in summaries if s.get("mmp")]
mmps_365 = [s["mmp"] for s in summaries if s.get("mmp") and s["started_at"] >= cutoff_365]
mmps_90 = [s["mmp"] for s in summaries if s.get("mmp") 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}
SPORTS = ["running", "cycling", "swimming", "hiking", "walking", "skiing", "other"]
records: dict[str, dict[float, dict]] = {s: {} for s in SPORTS}
best_climb: list[dict] = [] # top 10 best climbs for cycling
for s in summaries:
sport = s.get("sport", "other")
act_id = s.get("id", "")
started = s.get("started_at", "")
title = s.get("title", "")
# Distance-based best efforts
for d_km, t_s in (s.get("best_efforts") or []):
bucket = records.get(sport, {})
existing = bucket.get(d_km)
if existing is None or t_s < existing["time_s"]:
bucket[d_km] = {
"time_s": t_s,
"activity_id": act_id,
"started_at": started,
"title": title,
}
records[sport] = bucket
# Best climb (cycling only) — collect all, trim to top 10 after loop
climb = s.get("best_climb_m")
if climb and sport == "cycling":
best_climb.append({
"climb_m": climb,
"activity_id": act_id,
"started_at": started,
"title": title,
})
# Hiking / walking: track longest distance and most elevation from summaries
if sport in ("hiking", "walking"):
dist = s.get("distance_m") or 0
elev = s.get("elevation_gain_m") or 0
for metric, key, val in [("longest_m", "distance_m", dist),
("most_elevation_m", "elevation_gain_m", elev)]:
bucket = records[sport]
existing = bucket.get(metric)
if val and (existing is None or val > existing.get("value", 0)):
bucket[metric] = {
"value": val,
"activity_id": act_id,
"started_at": started,
"title": title,
}
records[sport] = bucket
# Serialise records: convert float keys to strings for JSON
def _serialise_sport_records(bucket: dict) -> dict:
return {str(k): v for k, v in bucket.items()}
athlete = {
"bas_version": "1.0",
"generated_at": now.isoformat(),
"power_curve": {
"all_time": _merge_mmps(all_mmps) if all_mmps else None,
"last_365d": _merge_mmps(mmps_365) if mmps_365 else None,
"last_90d": _merge_mmps(mmps_90) if mmps_90 else None,
"all_time": _merge_mmps(all_mmps) if all_mmps else None,
"last_365d": _merge_mmps(mmps_365) if mmps_365 else None,
"last_90d": _merge_mmps(mmps_90) if mmps_90 else None,
},
"records": {
sport: _serialise_sport_records(records[sport])
for sport in SPORTS
if records[sport]
},
"best_climbs": sorted(best_climb, key=lambda x: x["climb_m"], reverse=True)[:10],
**athlete_config,
}
(output_dir / "athlete.json").write_text(