166 lines
6.3 KiB
Python
166 lines
6.3 KiB
Python
"""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}
|