Add route heat colouring and detailed running analytics

This commit is contained in:
Peter Stockings
2026-10-01 18:43:33 +10:00
parent e37569ad09
commit 24c7fc2c6a
11 changed files with 526 additions and 31 deletions
+165
View File
@@ -0,0 +1,165 @@
"""Recorded metrics and explicitly modelled grade-adjusted pace for running.
Minetti et al., J Appl Physiol 93 (2002), doi:10.1152/japplphysiol.01177.2001.
GAP = elapsed moving time / equivalent flat distance. No heart-rate adjustment.
"""
import bisect
import itertools
import math
import statistics
def numeric(value):
return isinstance(value, (int, float)) and not isinstance(value, bool) and math.isfinite(value)
def running_cost(grade):
return (
155.4 * grade**5 - 30.4 * grade**4 - 43.3 * grade**3 + 46.3 * grade**2 + 19.5 * grade + 3.6
)
def pace_text(seconds):
if not numeric(seconds) or seconds <= 0:
return "—"
seconds = round(seconds)
return f"{seconds // 60}:{seconds % 60:02d}"
def enrich(samples, running=True):
rows = [dict(sample, grade=None, gap=None, pace=None) for sample in samples]
# Break smoothing at missing elevation/distance, distance resets and recording gaps.
chunks, chunk = [], []
for index, row in enumerate(rows):
valid = numeric(row.get("distance")) and numeric(row.get("elevation"))
if chunk:
previous = rows[chunk[-1]]
dt = (
row["time"] - previous["time"]
if numeric(row.get("time")) and numeric(previous.get("time"))
else 1
)
if not valid or row["distance"] < previous["distance"] or not 0 < dt <= 10:
chunks.append(chunk)
chunk = []
if valid:
chunk.append(index)
if chunk:
chunks.append(chunk)
for indices in chunks:
distances = [rows[i]["distance"] for i in indices]
altitude = [
statistics.median(rows[j]["elevation"] for j in indices[max(0, k - 2) : k + 3])
for k in range(len(indices))
]
for k, index in enumerate(indices):
low = max(0, bisect.bisect_left(distances, distances[k] - 30))
high = min(len(indices) - 1, bisect.bisect_left(distances, distances[k] + 30))
distance = distances[high] - distances[low]
if distance >= 20:
grade = (altitude[high] - altitude[low]) / distance
if abs(grade) <= 0.30:
rows[index]["grade"] = grade * 100
for row in rows:
speed = row.get("speed")
if numeric(speed) and 0.5 < speed < 15:
row["pace"] = 1000 / speed
if running and row["grade"] is not None:
row["gap"] = row["pace"] * running_cost(0) / running_cost(row["grade"] / 100)
return rows
def analyse(activity):
rows = enrich(activity.samples or [], activity.sport == "running")
weighted = {
key: [0.0, 0.0]
for key in (
"heart_rate",
"cadence",
"power",
"stride_length",
"vertical_oscillation",
"ground_contact",
)
}
terrains = {
key: {"label": label, "distance": 0.0, "time": 0.0, "hr_sum": 0.0, "hr_time": 0.0}
for key, label in (
("downhill", "Downhill < −2%"),
("flat", "Flat −2% to 2%"),
("uphill", "Uphill > 2%"),
)
}
equivalent, gap_time, covered = 0.0, 0.0, 0.0
laps = [dict(lap, gap=None, equivalent=0.0, gap_time=0.0) for lap in (activity.laps or [])]
ends, cumulative = [], 0.0
for lap in laps:
cumulative += lap.get("distance", 0)
ends.append(cumulative)
for previous, row in itertools.pairwise(rows):
if not numeric(row.get("time")) or not numeric(previous.get("time")):
continue
dt = row["time"] - previous["time"]
if not 0 < dt <= 10:
continue
for key, values in weighted.items():
value = row.get(key)
if numeric(value) and value > 0:
values[0] += value * dt
values[1] += dt
if not numeric(row.get("distance")) or not numeric(previous.get("distance")):
continue
ds = row["distance"] - previous["distance"]
if not 0 < ds <= dt * 15 or row["pace"] is None:
continue
grade = row["grade"]
if grade is not None:
terrain = terrains["uphill" if grade > 2 else "downhill" if grade < -2 else "flat"]
terrain["distance"] += ds
terrain["time"] += dt
if numeric(row.get("heart_rate")) and row["heart_rate"] > 0:
terrain["hr_sum"] += row["heart_rate"] * dt
terrain["hr_time"] += dt
if row["gap"] is not None:
flat_distance = ds * running_cost(grade / 100) / running_cost(0)
equivalent += flat_distance
gap_time += dt
covered += ds
# Split a sample crossing a lap boundary proportionally to distance.
start = previous["distance"]
while start < row["distance"]:
lap_index = bisect.bisect_right(ends, start)
if lap_index >= len(laps):
break
end = min(ends[lap_index], row["distance"])
fraction = (end - start) / ds
laps[lap_index]["equivalent"] += flat_distance * fraction
laps[lap_index]["gap_time"] += dt * fraction
start = end
for lap in laps:
lap["gap"] = 1000 * lap["gap_time"] / lap["equivalent"] if lap["equivalent"] > 0 else None
for terrain in terrains.values():
terrain["heart_rate"] = (
terrain["hr_sum"] / terrain["hr_time"] if terrain["hr_time"] else None
)
terrain["pace"] = (
terrain["time"] * 1000 / terrain["distance"] if terrain["distance"] else None
)
summary = {
key: total / duration if duration else None for key, (total, duration) in weighted.items()
}
summary.update(
{
"gap": 1000 * gap_time / equivalent if equivalent else None,
"gap_coverage": min(100, covered / activity.distance_m * 100)
if activity.distance_m
else 0,
"max_hr": max(
(r["heart_rate"] for r in rows if numeric(r.get("heart_rate"))), default=None
),
"max_power": max((r["power"] for r in rows if numeric(r.get("power"))), default=None),
"samples": len(rows),
}
)
return {"samples": rows, "summary": summary, "terrain": list(terrains.values()), "laps": laps}