289 lines
11 KiB
Python
289 lines
11 KiB
Python
"""Bulk day-level fitness metrics without per-day get_daily_summary loops."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from collections import defaultdict
|
|
from datetime import date, datetime, time, timedelta, timezone
|
|
from typing import Any
|
|
|
|
from sqlalchemy import func, select
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.db.models import FitnessProfile, FoodLog, StepLog, WaterLog, WorkoutLog
|
|
from app.fitness.activity_budget import estimate_workout_active_kcal
|
|
from app.fitness.calculators import compute_daily_targets, targets_to_api
|
|
from app.fitness.service import FitnessService
|
|
|
|
|
|
def plan_hit(totals: dict[str, float], targets: dict[str, float], goal: str, *, has_meals: bool) -> bool:
|
|
if not has_meals:
|
|
return False
|
|
actual = float(totals.get("calories", 0))
|
|
target = float(targets.get("calories", 0))
|
|
if target <= 0:
|
|
return False
|
|
if goal == "lose":
|
|
return actual <= target * 1.05
|
|
if goal == "gain":
|
|
return actual >= target * 0.95
|
|
return abs(actual - target) <= target * 0.1
|
|
|
|
|
|
def signed_balance(totals: dict[str, float], targets: dict[str, float], goal: str) -> float:
|
|
actual = float(totals.get("calories", 0))
|
|
target = float(targets.get("calories", 0))
|
|
if goal == "gain":
|
|
return actual - target
|
|
return target - actual
|
|
|
|
|
|
def streak_from_flags(flags_by_day: dict[date, bool], end: date) -> int:
|
|
streak = 0
|
|
cursor = end
|
|
while flags_by_day.get(cursor):
|
|
streak += 1
|
|
cursor -= timedelta(days=1)
|
|
return streak
|
|
|
|
|
|
def best_streak(flags_by_day: dict[date, bool]) -> int:
|
|
if not flags_by_day:
|
|
return 0
|
|
days = sorted(flags_by_day.keys())
|
|
best = 0
|
|
run = 0
|
|
prev: date | None = None
|
|
for d in days:
|
|
if not flags_by_day[d]:
|
|
run = 0
|
|
prev = d
|
|
continue
|
|
if prev and (d - prev).days == 1:
|
|
run += 1
|
|
else:
|
|
run = 1
|
|
best = max(best, run)
|
|
prev = d
|
|
return best
|
|
|
|
|
|
class DayMetricsService:
|
|
def __init__(self, db: Session, user_id: int):
|
|
self.db = db
|
|
self.user_id = user_id
|
|
self._fitness = FitnessService(db, user_id)
|
|
|
|
def first_activity_date(self) -> date | None:
|
|
candidates: list[date] = []
|
|
for model, dt_col in (
|
|
(FoodLog, FoodLog.logged_at),
|
|
(WaterLog, WaterLog.logged_at),
|
|
(StepLog, StepLog.logged_at),
|
|
(WorkoutLog, WorkoutLog.logged_at),
|
|
):
|
|
ts = self.db.scalar(select(func.min(dt_col)).where(model.user_id == self.user_id))
|
|
if ts is not None:
|
|
candidates.append(ts.date() if isinstance(ts, datetime) else ts)
|
|
return min(candidates) if candidates else None
|
|
|
|
def days_with_data_count(self, end: date) -> int:
|
|
first = self.first_activity_date()
|
|
if first is None:
|
|
return 0
|
|
return self._count_days_with_data(first, end)
|
|
|
|
def _count_days_with_data(self, start: date, end: date) -> int:
|
|
start_dt = datetime.combine(start, time.min, tzinfo=timezone.utc)
|
|
end_dt = datetime.combine(end, time.max, tzinfo=timezone.utc)
|
|
days: set[date] = set()
|
|
for model, dt_col in (
|
|
(FoodLog, FoodLog.logged_at),
|
|
(WaterLog, WaterLog.logged_at),
|
|
(StepLog, StepLog.logged_at),
|
|
(WorkoutLog, WorkoutLog.logged_at),
|
|
):
|
|
rows = self.db.scalars(
|
|
select(dt_col).where(
|
|
model.user_id == self.user_id,
|
|
dt_col >= start_dt,
|
|
dt_col <= end_dt,
|
|
)
|
|
).all()
|
|
for ts in rows:
|
|
days.add(ts.date() if isinstance(ts, datetime) else ts)
|
|
return len(days)
|
|
|
|
def _load_range(self, start: date, end: date) -> dict[date, dict[str, Any]]:
|
|
start_dt = datetime.combine(start, time.min, tzinfo=timezone.utc)
|
|
end_dt = datetime.combine(end, time.max, tzinfo=timezone.utc)
|
|
profile_dict = self._fitness.get_profile() or {}
|
|
profile = self._fitness._profile_for_budget(profile_dict)
|
|
weight_kg = float(profile_dict.get("weight_kg") or 70)
|
|
goal = str(profile_dict.get("goal") or "lose")
|
|
|
|
meal_count: dict[date, int] = defaultdict(int)
|
|
totals: dict[date, dict[str, float]] = defaultdict(
|
|
lambda: {
|
|
"calories": 0.0,
|
|
"protein_g": 0.0,
|
|
"fat_g": 0.0,
|
|
"carbs_g": 0.0,
|
|
"fiber_g": 0.0,
|
|
"water_ml": 0.0,
|
|
"steps": 0.0,
|
|
}
|
|
)
|
|
steps_by_day: dict[date, int] = defaultdict(int)
|
|
workout_kcal_by_day: dict[date, float] = defaultdict(float)
|
|
workout_count: dict[date, int] = defaultdict(int)
|
|
|
|
foods = self.db.scalars(
|
|
select(FoodLog).where(
|
|
FoodLog.user_id == self.user_id,
|
|
FoodLog.logged_at >= start_dt,
|
|
FoodLog.logged_at <= end_dt,
|
|
)
|
|
).all()
|
|
for row in foods:
|
|
d = row.logged_at.date()
|
|
meal_count[d] += 1
|
|
totals[d]["calories"] += row.calories
|
|
totals[d]["protein_g"] += row.protein_g
|
|
totals[d]["fat_g"] += row.fat_g
|
|
totals[d]["carbs_g"] += row.carbs_g
|
|
totals[d]["fiber_g"] += row.fiber_g
|
|
|
|
for row in self.db.scalars(
|
|
select(WaterLog).where(
|
|
WaterLog.user_id == self.user_id,
|
|
WaterLog.logged_at >= start_dt,
|
|
WaterLog.logged_at <= end_dt,
|
|
)
|
|
).all():
|
|
totals[row.logged_at.date()]["water_ml"] += float(row.amount_ml)
|
|
|
|
for row in self.db.scalars(
|
|
select(StepLog).where(
|
|
StepLog.user_id == self.user_id,
|
|
StepLog.logged_at >= start_dt,
|
|
StepLog.logged_at <= end_dt,
|
|
)
|
|
).all():
|
|
d = row.logged_at.date()
|
|
steps_by_day[d] += row.steps
|
|
totals[d]["steps"] += float(row.steps)
|
|
|
|
for row in self.db.scalars(
|
|
select(WorkoutLog).where(
|
|
WorkoutLog.user_id == self.user_id,
|
|
WorkoutLog.logged_at >= start_dt,
|
|
WorkoutLog.logged_at <= end_dt,
|
|
)
|
|
).all():
|
|
d = row.logged_at.date()
|
|
workout_count[d] += 1
|
|
kcal = estimate_workout_active_kcal(
|
|
FitnessService._workout_to_dict(row),
|
|
weight_kg=weight_kg,
|
|
)
|
|
workout_kcal_by_day[d] += kcal
|
|
|
|
result: dict[date, dict[str, Any]] = {}
|
|
cursor = start
|
|
while cursor <= end:
|
|
day_totals = dict(totals[cursor])
|
|
steps_total = int(steps_by_day.get(cursor, 0))
|
|
workouts = (
|
|
[{"active_calories": workout_kcal_by_day[cursor]}]
|
|
if workout_kcal_by_day.get(cursor, 0) > 0
|
|
else []
|
|
)
|
|
daily_targets = compute_daily_targets(
|
|
profile,
|
|
steps_total=steps_total,
|
|
workouts=workouts,
|
|
)
|
|
targets = targets_to_api(daily_targets)
|
|
has_meals = meal_count.get(cursor, 0) > 0
|
|
result[cursor] = {
|
|
"date": cursor.isoformat(),
|
|
"totals": day_totals,
|
|
"targets": targets,
|
|
"meal_count": meal_count.get(cursor, 0),
|
|
"workout_count": workout_count.get(cursor, 0),
|
|
"steps_total": steps_total,
|
|
"has_meals": has_meals,
|
|
"plan_hit": plan_hit(day_totals, targets, goal, has_meals=has_meals),
|
|
"balance": signed_balance(day_totals, targets, goal) if has_meals else 0.0,
|
|
}
|
|
cursor += timedelta(days=1)
|
|
return result
|
|
|
|
def day_summaries(self, start: date, end: date) -> dict[date, dict[str, Any]]:
|
|
return self._load_range(start, end)
|
|
|
|
def metrics_bundle(self, end: date) -> dict[str, Any]:
|
|
first = self.first_activity_date()
|
|
start = first if first is not None else end - timedelta(days=364)
|
|
if start > end:
|
|
start = end
|
|
days = self._load_range(start, end)
|
|
|
|
meal_flags = {d: info["meal_count"] > 0 for d, info in days.items()}
|
|
plan_flags = {d: info["plan_hit"] for d, info in days.items()}
|
|
balance_by_day = {d: info["balance"] for d, info in days.items() if info["meal_count"] > 0}
|
|
|
|
week_start = end - timedelta(days=6)
|
|
month_start = end - timedelta(days=29)
|
|
|
|
return {
|
|
"start": start,
|
|
"end": end,
|
|
"days": days,
|
|
"meal_flags": meal_flags,
|
|
"plan_flags": plan_flags,
|
|
"balance_by_day": balance_by_day,
|
|
"meal_log_streak": streak_from_flags(meal_flags, end),
|
|
"meal_log_best_streak": best_streak(meal_flags),
|
|
"plan_hit_streak": streak_from_flags(plan_flags, end),
|
|
"plan_hit_total": sum(1 for hit in plan_flags.values() if hit),
|
|
"plan_hit_week": sum(1 for d, hit in plan_flags.items() if hit and d >= week_start),
|
|
"deficit_week": round(sum(v for d, v in balance_by_day.items() if d >= week_start)),
|
|
"deficit_month": round(sum(v for d, v in balance_by_day.items() if d >= month_start)),
|
|
"deficit_all": round(sum(balance_by_day.values())),
|
|
}
|
|
|
|
def day_fitness_context(self, day: date) -> dict[str, Any]:
|
|
info = self._load_range(day, day).get(day)
|
|
if not info:
|
|
return {
|
|
"date": day.isoformat(),
|
|
"meals_count": 0,
|
|
"workouts_count": 0,
|
|
"steps_total": 0,
|
|
"calories": 0,
|
|
"calorie_target": 0,
|
|
"calorie_delta": 0,
|
|
"protein_g": 0,
|
|
"protein_target": 0,
|
|
"water_l": 0.0,
|
|
"has_fitness_data": False,
|
|
}
|
|
totals = info["totals"]
|
|
targets = info["targets"]
|
|
calorie_delta = float(totals.get("calories", 0)) - float(targets.get("calories", 0))
|
|
has_data = info["meal_count"] > 0 or info["workout_count"] > 0 or info["steps_total"] > 0
|
|
return {
|
|
"date": day.isoformat(),
|
|
"meals_count": info["meal_count"],
|
|
"workouts_count": info["workout_count"],
|
|
"steps_total": info["steps_total"],
|
|
"calories": round(float(totals.get("calories", 0))),
|
|
"calorie_target": round(float(targets.get("calories", 0))),
|
|
"calorie_delta": round(calorie_delta),
|
|
"protein_g": round(float(totals.get("protein_g", 0))),
|
|
"protein_target": round(float(targets.get("protein_g", 0))),
|
|
"water_l": round(float(totals.get("water_ml", 0)) / 1000, 1),
|
|
"has_fitness_data": has_data,
|
|
}
|