94 lines
3.7 KiB
Python
94 lines
3.7 KiB
Python
"""Estimated distance best efforts from full-resolution recorded samples."""
|
|
|
|
from bisect import bisect_left, bisect_right
|
|
|
|
from .analytics import numeric
|
|
|
|
DISTANCES = (
|
|
("1 km", 1000),
|
|
("2 km", 2000),
|
|
("5 km", 5000),
|
|
("10 km", 10000),
|
|
("Half marathon", 21097.5),
|
|
("Marathon", 42195),
|
|
)
|
|
|
|
|
|
def best_efforts(activity):
|
|
"""Search piecewise-linear distance/time curves, counting internal stops.
|
|
|
|
A minimum starts or ends at a recorded distance breakpoint. Evaluate both
|
|
sets, interpolating the opposite boundary. At stationary boundaries use
|
|
latest departure and earliest arrival, excluding stops outside the effort.
|
|
Missing samples, distance resets and non-increasing times break the curve.
|
|
Timestamp gaps remain elapsed time, including pauses in watch recording.
|
|
"""
|
|
if activity.sport != "running":
|
|
return []
|
|
segments, segment = [], []
|
|
for sample in activity.samples:
|
|
distance, time = sample.get("distance"), sample.get("time")
|
|
if not numeric(distance) or not numeric(time) or distance < 0:
|
|
if segment:
|
|
segments.append(segment)
|
|
segment = []
|
|
continue
|
|
if segment and (distance < segment[-1][0] or time <= segment[-1][1]):
|
|
segments.append(segment)
|
|
segment = []
|
|
segment.append((distance, time))
|
|
if segment:
|
|
segments.append(segment)
|
|
best = {}
|
|
for segment in segments:
|
|
distances = [point[0] for point in segment]
|
|
times = [point[1] for point in segment]
|
|
|
|
def at(distance, departure=False, distances=distances, times=times):
|
|
index = (
|
|
bisect_right(distances, distance) - 1
|
|
if departure
|
|
else bisect_left(distances, distance)
|
|
)
|
|
if distances[index] == distance:
|
|
return times[index]
|
|
if departure:
|
|
index += 1
|
|
fraction = (distance - distances[index - 1]) / (distances[index] - distances[index - 1])
|
|
return times[index - 1] + fraction * (times[index] - times[index - 1])
|
|
|
|
for label, length in DISTANCES:
|
|
if distances[-1] - distances[0] < length:
|
|
continue
|
|
starts = {d for d in distances if d + length <= distances[-1]}
|
|
starts.update(d - length for d in distances if d - length >= distances[0])
|
|
for start in sorted(starts):
|
|
seconds = at(start + length) - at(start, departure=True)
|
|
if seconds > 0 and (label not in best or seconds < best[label]["seconds"]):
|
|
best[label] = {
|
|
"label": label,
|
|
"distance": length,
|
|
"seconds": seconds,
|
|
"start_m": start,
|
|
"end_m": start + length,
|
|
}
|
|
return [best[label] for label, _ in DISTANCES if label in best]
|
|
|
|
|
|
def record_book(activities):
|
|
"""Recompute in activity chronology so backfills and removals stay correct.
|
|
|
|
First efforts establish a record. Strict improvements earn historical PR
|
|
badges, which remain on the run after a later activity beats that record.
|
|
"""
|
|
records, runs = {}, {}
|
|
for activity in sorted(activities, key=lambda row: (row.started_at, row.id)):
|
|
efforts = best_efforts(activity)
|
|
for effort in efforts:
|
|
previous = records.get(effort["label"])
|
|
effort["pr"] = previous is None or effort["seconds"] < previous["seconds"] - 1e-6
|
|
if effort["pr"]:
|
|
records[effort["label"]] = dict(effort, activity=activity)
|
|
runs[activity.id] = efforts
|
|
return {"records": [records[label] for label, _ in DISTANCES if label in records], "runs": runs}
|