158 lines
7.3 KiB
Python
158 lines
7.3 KiB
Python
"""Owner-scoped daily observations and matched recording reductions."""
|
|
|
|
from collections import defaultdict
|
|
from copy import copy
|
|
from datetime import datetime, timedelta
|
|
from types import SimpleNamespace
|
|
|
|
from .extensions import db
|
|
from .models import DailyHealth
|
|
from .stats_analysis import evaluate, expression_tree, valid
|
|
|
|
|
|
def daily(history, owner_id, zone, start, end, q, catalog):
|
|
by_day = defaultdict(list)
|
|
for row in history:
|
|
by_day[row.day].append(row)
|
|
observations = {r.day: r for r in db.session.scalars(
|
|
db.select(DailyHealth).where(DailyHealth.owner_id == owner_id,
|
|
DailyHealth.day >= start, DailyHealth.day < end))}
|
|
result = []
|
|
last_matching = {}
|
|
streak = 0
|
|
earlier = [r for r in history if r.day < start]
|
|
if earlier:
|
|
day = start - timedelta(days=1)
|
|
while by_day[day]:
|
|
streak += 1
|
|
day -= timedelta(days=1)
|
|
from .analysis_engine import matches
|
|
|
|
since_specs = [s for s in q["derived"] if "since" in s]
|
|
for spec in since_specs:
|
|
qualifying = [r.day for r in earlier if matches(r, spec["since"])]
|
|
last_matching[spec["name"]] = max(qualifying) if qualifying else None
|
|
for index in range((end - start).days):
|
|
day = start + timedelta(days=index)
|
|
rows = by_day[day]
|
|
if rows:
|
|
streak += 1
|
|
else:
|
|
streak = 0
|
|
values = {key: None for key in catalog}
|
|
distance = sum(r.distance_m for r in rows)
|
|
duration = sum(r.duration_s for r in rows)
|
|
values.update(distance=distance / 1000, duration=duration / 3600,
|
|
sessions=len(rows), count=1, run_day_streak=streak)
|
|
window = [r for d in range(7) for r in by_day[day - timedelta(days=d)]]
|
|
total = sum(r.distance_m for r in window)
|
|
values["long_run_share"] = max(r.distance_m for r in window) / total * 100 if total else None
|
|
for spec in since_specs:
|
|
name = spec["name"]
|
|
if any(matches(r, spec["since"]) for r in rows):
|
|
last_matching[name] = day
|
|
last = last_matching[name]
|
|
values[name] = (day - last).days if last else None
|
|
for field in ("perceived_effort", "fatigue"):
|
|
known = [r.values[field] for r in rows if valid(r.values.get(field))]
|
|
values[field] = sum(known) / len(known) if known else None
|
|
health = observations.get(day)
|
|
if health:
|
|
values.update({key: value for key, value in health.values.items() if key in catalog})
|
|
if q["source"] == "health" and not (health and any(valid(v) for v in health.values.values())):
|
|
continue
|
|
instant = datetime.combine(day, datetime.min.time(), zone)
|
|
row = SimpleNamespace(id=day.toordinal(), day=day, local=instant, instant=instant,
|
|
started_at=instant, title=f"Training day · {day}",
|
|
distance_m=distance, duration_s=duration, elevation_m=None,
|
|
avg_hr=None, analysis_summary=values, values=values, previous=None,
|
|
activity_id=None, activity_ids=[r.id for r in rows])
|
|
result.append(row)
|
|
for spec in q["derived"]:
|
|
if "since" in spec:
|
|
continue
|
|
if "window" in spec:
|
|
config = spec["window"]
|
|
from .stats_analysis import statistic
|
|
|
|
for row in result:
|
|
window = [r.values.get(config["field"]) for r in result
|
|
if row.day - timedelta(days=config["days"]) < r.day <= row.day]
|
|
row.values[spec["name"]] = statistic(window, config["function"])
|
|
else:
|
|
tree = expression_tree(spec["expression"], catalog)[0]
|
|
for row in result:
|
|
row.values[spec["name"]] = evaluate(tree, row.values)
|
|
return result
|
|
|
|
|
|
def reduce_runs(rows, minimum, catalog):
|
|
"""Each session contributes exactly one observation after recording filters."""
|
|
groups = defaultdict(list)
|
|
for row in rows:
|
|
groups[row.activity_id].append(row)
|
|
result = []
|
|
for activity_id, units in groups.items():
|
|
units.sort(key=lambda r: r.values.get("elapsed") or r.values.get("moving_offset") or 0)
|
|
duration = sum(r.duration_s for r in units)
|
|
if duration + 1e-7 < minimum * 60:
|
|
continue
|
|
row = copy(units[0])
|
|
row.id = row.activity_id = activity_id
|
|
row.title = getattr(row, "parent_title", row.title)
|
|
row.distance_m = sum(r.distance_m for r in units)
|
|
row.duration_s = duration
|
|
values = {key: None for key in catalog}
|
|
for key in catalog:
|
|
coverage_key = "sensor_coverage" if key == "heart_rate" else key + "_coverage"
|
|
pairs = [(r.values.get(key), r.duration_s * r.values.get(coverage_key, 100) / 100)
|
|
for r in units if valid(r.values.get(key))]
|
|
weight = sum(w for _, w in pairs)
|
|
values[key] = sum(v * w for v, w in pairs) / weight if weight else None
|
|
values.update(distance=row.distance_m / 1000, duration=duration / 3600,
|
|
pace=duration * 1000 / row.distance_m, count=1,
|
|
matched_minutes=duration / 60, matched_windows=len(units))
|
|
# Post-filter halves of matched moving time. Require >=10 minutes each,
|
|
# comparable pace (10%) and mean signed grade (one percentage point).
|
|
halves = [[0.0, 0.0, 0.0, 0.0, 0.0] for _ in range(2)]
|
|
elapsed = 0.0
|
|
for unit in units:
|
|
hr, grade = unit.values.get("heart_rate"), unit.values.get("grade")
|
|
recorded = getattr(unit, "recording", {})
|
|
fraction = recorded.get("joint_coverage", 100) / 100
|
|
hr = recorded.get("joint_hr", hr)
|
|
grade = recorded.get("joint_grade", grade)
|
|
speed = recorded.get("joint_speed", unit.distance_m / unit.duration_s)
|
|
for index, low, high in ((0, 0, duration / 2), (1, duration / 2, duration)):
|
|
dt = max(0, min(elapsed + unit.duration_s, high) - max(elapsed, low))
|
|
if dt and valid(hr) and valid(grade) and valid(speed):
|
|
dt *= fraction
|
|
part = halves[index]
|
|
part[0] += dt
|
|
part[1] += speed * dt
|
|
part[2] += hr * dt
|
|
part[3] += grade * dt
|
|
part[4] += dt
|
|
elapsed += unit.duration_s
|
|
drift = None
|
|
if all(p[0] >= 600 and p[0] >= duration * .4 and p[1] > 0 and p[2] > 0 for p in halves):
|
|
first, second = halves
|
|
speeds = [p[1] / p[0] for p in halves]
|
|
grades = [p[3] / p[0] for p in halves]
|
|
if abs(speeds[1] / speeds[0] - 1) <= .1 and abs(grades[1] - grades[0]) <= 1:
|
|
drift = (1 - (second[1] / second[2]) / (first[1] / first[2])) * 100
|
|
values["heart_rate_drift"] = drift
|
|
row.avg_hr = values["heart_rate"]
|
|
row.values = row.analysis_summary = values
|
|
result.append(row)
|
|
return result
|
|
|
|
|
|
def evidence(rows, source):
|
|
ids = {getattr(row, "activity_id", row.id) for row in rows}
|
|
ids.discard(None)
|
|
if source in {"days", "health"}:
|
|
ids = {i for row in rows for i in row.activity_ids}
|
|
return {"sessions": len(ids), "inputs": len(rows),
|
|
"minutes": round(sum(row.duration_s for row in rows) / 60, 1)}
|