94 lines
3.4 KiB
Python
94 lines
3.4 KiB
Python
"""Running load v1: personal five-zone minutes, with daily exponential decay."""
|
|
|
|
from collections import defaultdict
|
|
from datetime import UTC, datetime, timedelta
|
|
from math import exp
|
|
from zoneinfo import ZoneInfo
|
|
|
|
from sqlalchemy.orm import undefer
|
|
|
|
from .extensions import db
|
|
from .heart_rate import zones
|
|
from .models import Activity
|
|
|
|
|
|
def effort(activity, bands):
|
|
"""Require 80% timer-time HR coverage; never extrapolate missing HR."""
|
|
result = {"score": None, "coverage": 0, "reason": "Set your HR zones in Settings."}
|
|
if activity.sport != "running":
|
|
result["reason"] = "Running activities only."
|
|
return result
|
|
units = (activity.analysis_summary or {}).get("intervals", [])
|
|
seconds = [0.0] * (len(bands) + 1 if bands else 5)
|
|
for unit in units:
|
|
for hr, duration in unit.get("hr_seconds", {}).items():
|
|
hr = float(hr)
|
|
if 30 <= hr <= 240:
|
|
band = sum(hr >= boundary for boundary in bands) if bands else 0
|
|
seconds[band] += duration
|
|
coverage = sum(seconds) / activity.duration_s if activity.duration_s > 0 else 0
|
|
result["coverage"] = round(min(coverage, 1) * 100, 1)
|
|
if not bands:
|
|
return result
|
|
result["reason"] = "Insufficient recorded HR (80% timer-time coverage required)."
|
|
if coverage >= 0.8:
|
|
result.update(
|
|
score=sum(t * w for w, t in enumerate(seconds, 1)) / 60,
|
|
reason="Recorded HR · zone weights 1 / 2 / 3 / 4 / 5",
|
|
)
|
|
result["zone_minutes"] = [round(t / 60, 2) for t in seconds]
|
|
return result
|
|
|
|
|
|
def history(user, today=None):
|
|
zone = ZoneInfo(user.timezone)
|
|
today = today or datetime.now(zone).date()
|
|
zone_settings = zones(user, today)
|
|
rows = db.session.scalars(
|
|
db.select(Activity)
|
|
.where(Activity.owner_id == user.id, Activity.sport == "running")
|
|
.options(undefer(Activity.analysis_summary))
|
|
.order_by(Activity.started_at, Activity.id)
|
|
).all()
|
|
days = defaultdict(list)
|
|
runs = []
|
|
for row in rows:
|
|
day = row.started_at.replace(tzinfo=UTC).astimezone(zone).date()
|
|
if day > today:
|
|
continue
|
|
scored = effort(row, zone_settings["boundaries"])
|
|
run = {"id": row.id, "title": row.title, "day": day.isoformat(), **scored}
|
|
days[day].append(run)
|
|
runs.append(run)
|
|
series = []
|
|
fitness = fatigue = 0.0
|
|
incomplete = False
|
|
if days:
|
|
day = min(days)
|
|
start = day
|
|
while day <= today:
|
|
entries = days.get(day, [])
|
|
missing = sum(r["score"] is None for r in entries)
|
|
incomplete |= missing > 0
|
|
load = sum(r["score"] for r in entries if r["score"] is not None)
|
|
fitness += (load - fitness) * (1 - exp(-1 / 42))
|
|
fatigue += (load - fatigue) * (1 - exp(-1 / 7))
|
|
series.append(
|
|
{
|
|
"day": day.isoformat(),
|
|
"load": None if missing else load,
|
|
"fitness": round(fitness, 2),
|
|
"fatigue": round(fatigue, 2),
|
|
"form": round(fitness - fatigue, 2),
|
|
"missing": missing,
|
|
"provisional": (day - start).days < 126 or incomplete,
|
|
}
|
|
)
|
|
day += timedelta(days=1)
|
|
return {
|
|
"series": series,
|
|
"runs": list(reversed(runs)),
|
|
"unscored": sum(r["score"] is None for r in runs),
|
|
"zones": zone_settings,
|
|
}
|