826 lines
29 KiB
Python
826 lines
29 KiB
Python
import json
|
|
from collections import defaultdict
|
|
from datetime import date, datetime, time, timedelta, timezone
|
|
from typing import Any
|
|
|
|
from sqlalchemy import select
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.db.models import (
|
|
BodyMetric,
|
|
FitnessProfile,
|
|
FitnessReminder,
|
|
FoodLog,
|
|
StepLog,
|
|
WaterLog,
|
|
WorkoutLog,
|
|
)
|
|
from app.fitness.activity_budget import estimate_workout_active_kcal
|
|
from app.fitness.calculators import (
|
|
compute_daily_targets,
|
|
compute_expected_targets,
|
|
compute_targets,
|
|
one_rep_max,
|
|
targets_to_api,
|
|
tdee_breakdown_to_api,
|
|
tdee_expected_to_api,
|
|
EXPECTED_LOOKBACK_DAYS,
|
|
)
|
|
from app.fitness.body_composition import compute_body_composition
|
|
|
|
DEFAULT_REMINDERS = [
|
|
{"kind": "water", "hour": 9, "minute": 0, "interval_hours": 2},
|
|
{"kind": "meal", "hour": 13, "minute": 0, "interval_hours": None},
|
|
{"kind": "workout", "hour": 18, "minute": 0, "interval_hours": None},
|
|
{"kind": "weigh_in", "hour": 8, "minute": 0, "interval_hours": None},
|
|
]
|
|
|
|
|
|
class FitnessService:
|
|
def __init__(self, db: Session, user_id: int):
|
|
self.db = db
|
|
self.user_id = user_id
|
|
|
|
def _get_profile_row(self) -> FitnessProfile | None:
|
|
return self.db.scalar(select(FitnessProfile).where(FitnessProfile.user_id == self.user_id).limit(1))
|
|
|
|
def get_profile(self) -> dict[str, Any] | None:
|
|
row = self._get_profile_row()
|
|
if not row:
|
|
return None
|
|
return self._profile_to_dict(row)
|
|
|
|
def _profile_params(self, row: FitnessProfile) -> dict[str, Any]:
|
|
return {
|
|
"sex": row.sex,
|
|
"age": row.age,
|
|
"height_cm": row.height_cm,
|
|
"weight_kg": row.weight_kg,
|
|
"goal": row.goal,
|
|
"neat_base_kcal": row.neat_base_kcal,
|
|
"activity_level": row.activity_level,
|
|
"weekly_workouts": row.weekly_workouts,
|
|
"baseline_steps": row.baseline_steps,
|
|
"baseline_workout_kcal": row.baseline_workout_kcal,
|
|
}
|
|
|
|
def _profile_to_dict(self, row: FitnessProfile) -> dict[str, Any]:
|
|
targets = compute_targets(self._profile_params(row))
|
|
return {
|
|
"sex": row.sex,
|
|
"age": row.age,
|
|
"height_cm": row.height_cm,
|
|
"weight_kg": row.weight_kg,
|
|
"goal": row.goal,
|
|
"target_weight_kg": row.target_weight_kg,
|
|
"neat_base_kcal": row.neat_base_kcal,
|
|
"activity_level": row.activity_level,
|
|
"weekly_workouts": row.weekly_workouts,
|
|
"baseline_steps": row.baseline_steps,
|
|
"baseline_workout_kcal": row.baseline_workout_kcal,
|
|
"calorie_target": row.calorie_target,
|
|
"protein_g": row.protein_g,
|
|
"fat_g": row.fat_g,
|
|
"carbs_g": row.carbs_g,
|
|
"water_l": row.water_l,
|
|
"computed": targets,
|
|
"updated_at": row.updated_at.isoformat() if row.updated_at else None,
|
|
}
|
|
|
|
def set_profile(self, updates: dict[str, Any]) -> dict[str, Any]:
|
|
row = self._get_profile_row()
|
|
is_new = row is None
|
|
if is_new:
|
|
row = FitnessProfile(user_id=self.user_id)
|
|
self.db.add(row)
|
|
self.db.flush()
|
|
|
|
for key in (
|
|
"sex", "age", "height_cm", "weight_kg",
|
|
"goal", "target_weight_kg", "neat_base_kcal",
|
|
"activity_level", "weekly_workouts",
|
|
"baseline_steps", "baseline_workout_kcal",
|
|
):
|
|
if key in updates and updates[key] is not None:
|
|
setattr(row, key, updates[key])
|
|
|
|
targets = compute_targets(self._profile_params(row))
|
|
row.calorie_target = targets["calorie_target"]
|
|
row.protein_g = targets["protein_g"]
|
|
row.fat_g = targets["fat_g"]
|
|
row.carbs_g = targets["carbs_g"]
|
|
row.water_l = targets["water_l"]
|
|
row.updated_at = datetime.now(timezone.utc)
|
|
|
|
if is_new:
|
|
self._ensure_default_reminders()
|
|
|
|
self.db.commit()
|
|
self.db.refresh(row)
|
|
return {"ok": True, "profile": self._profile_to_dict(row)}
|
|
|
|
def _ensure_default_reminders(self) -> None:
|
|
existing = self.db.scalars(select(FitnessReminder).where(FitnessReminder.user_id == self.user_id)).all()
|
|
if existing:
|
|
return
|
|
for item in DEFAULT_REMINDERS:
|
|
self.db.add(FitnessReminder(user_id=self.user_id, **item))
|
|
|
|
def calc_targets(self, params: dict[str, Any]) -> dict[str, Any]:
|
|
return compute_targets(params)
|
|
|
|
def calc_body_composition(self, params: dict[str, Any]) -> dict[str, Any]:
|
|
profile = self.get_profile() or {}
|
|
sex = params.get("sex") or profile.get("sex") or "male"
|
|
height_cm = float(params.get("height_cm") or profile.get("height_cm") or 170)
|
|
weight_kg = float(params.get("weight_kg") or profile.get("weight_kg") or 70)
|
|
return compute_body_composition(
|
|
sex=str(sex),
|
|
height_cm=height_cm,
|
|
weight_kg=weight_kg,
|
|
neck_cm=params.get("neck_cm"),
|
|
waist_cm=params.get("waist_cm"),
|
|
hip_cm=params.get("hip_cm"),
|
|
body_fat_pct=params.get("body_fat_pct"),
|
|
)
|
|
|
|
def get_latest_body_composition(self) -> dict[str, Any] | None:
|
|
rows = self.list_body_metrics(limit=1)
|
|
return rows[0] if rows else None
|
|
|
|
@staticmethod
|
|
def _body_metric_to_dict(row: BodyMetric) -> dict[str, Any]:
|
|
return {
|
|
"id": row.id,
|
|
"weight_kg": row.weight_kg,
|
|
"body_fat_pct": row.body_fat_pct,
|
|
"body_fat_method": row.body_fat_method,
|
|
"chest_cm": row.chest_cm,
|
|
"waist_cm": row.waist_cm,
|
|
"neck_cm": row.neck_cm,
|
|
"hip_cm": row.hip_cm,
|
|
"whr": row.whr,
|
|
"lbm_kg": row.lbm_kg,
|
|
"ffmi": row.ffmi,
|
|
"notes": row.notes,
|
|
"recorded_at": row.recorded_at.isoformat() if row.recorded_at else None,
|
|
}
|
|
|
|
@staticmethod
|
|
def _resolve_logged_at(
|
|
*,
|
|
logged_at: datetime | str | None = None,
|
|
day: date | None = None,
|
|
days_ago: int | None = None,
|
|
) -> datetime:
|
|
if logged_at is not None:
|
|
if isinstance(logged_at, str):
|
|
dt = datetime.fromisoformat(logged_at.replace("Z", "+00:00"))
|
|
if dt.tzinfo is None:
|
|
dt = dt.replace(tzinfo=timezone.utc)
|
|
return dt
|
|
if logged_at.tzinfo is None:
|
|
return logged_at.replace(tzinfo=timezone.utc)
|
|
return logged_at
|
|
|
|
target_day = day
|
|
if target_day is None and days_ago is not None:
|
|
target_day = datetime.now(timezone.utc).date() - timedelta(days=int(days_ago))
|
|
|
|
if target_day is None:
|
|
return datetime.now(timezone.utc)
|
|
|
|
return datetime.combine(target_day, time(12, 0), tzinfo=timezone.utc)
|
|
|
|
def _profile_for_budget(self, profile: dict[str, Any] | None) -> dict[str, Any]:
|
|
if profile:
|
|
return profile
|
|
return {
|
|
"weight_kg": 70,
|
|
"height_cm": 170,
|
|
"age": 30,
|
|
"sex": "male",
|
|
"goal": "maintain",
|
|
"neat_base_kcal": 200,
|
|
"activity_level": "moderate",
|
|
"weekly_workouts": 3,
|
|
"baseline_steps": None,
|
|
"baseline_workout_kcal": None,
|
|
}
|
|
|
|
def _activity_history(
|
|
self,
|
|
end_day: date,
|
|
*,
|
|
days: int = EXPECTED_LOOKBACK_DAYS,
|
|
) -> list[dict[str, Any]]:
|
|
"""Daily steps and workout kcal for `days` calendar days before end_day (exclusive)."""
|
|
if days <= 0:
|
|
return []
|
|
start = end_day - timedelta(days=days)
|
|
range_start = datetime.combine(start, time.min, tzinfo=timezone.utc)
|
|
range_end = datetime.combine(end_day - timedelta(days=1), time.max, tzinfo=timezone.utc)
|
|
|
|
profile_row = self._get_profile_row()
|
|
weight_kg = float(profile_row.weight_kg) if profile_row else 70.0
|
|
|
|
steps_by_day: dict[date, int] = defaultdict(int)
|
|
workout_kcal_by_day: dict[date, float] = defaultdict(float)
|
|
|
|
steps_rows = self.db.scalars(
|
|
select(StepLog).where(
|
|
StepLog.user_id == self.user_id,
|
|
StepLog.logged_at >= range_start,
|
|
StepLog.logged_at <= range_end,
|
|
)
|
|
).all()
|
|
for row in steps_rows:
|
|
steps_by_day[row.logged_at.date()] += row.steps
|
|
|
|
workouts_rows = self.db.scalars(
|
|
select(WorkoutLog).where(
|
|
WorkoutLog.user_id == self.user_id,
|
|
WorkoutLog.logged_at >= range_start,
|
|
WorkoutLog.logged_at <= range_end,
|
|
)
|
|
).all()
|
|
for row in workouts_rows:
|
|
d = row.logged_at.date()
|
|
workout_kcal_by_day[d] += estimate_workout_active_kcal(
|
|
self._workout_to_dict(row),
|
|
weight_kg=weight_kg,
|
|
)
|
|
|
|
history: list[dict[str, Any]] = []
|
|
cursor = start
|
|
while cursor < end_day:
|
|
history.append(
|
|
{
|
|
"date": cursor.isoformat(),
|
|
"steps": steps_by_day.get(cursor, 0),
|
|
"workout_kcal": round(workout_kcal_by_day.get(cursor, 0.0), 1),
|
|
}
|
|
)
|
|
cursor += timedelta(days=1)
|
|
return history
|
|
|
|
def _maybe_update_baseline(self, profile_row: FitnessProfile | None, expected: dict[str, Any]) -> None:
|
|
if profile_row is None:
|
|
return
|
|
if expected.get("source") != "weekly_avg":
|
|
return
|
|
if int(expected.get("days_with_data") or 0) < 5:
|
|
return
|
|
profile_row.baseline_steps = int(expected.get("expected_steps") or 0)
|
|
profile_row.baseline_workout_kcal = round(
|
|
float(expected.get("expected_workout_kcal") or 0) * 7,
|
|
1,
|
|
)
|
|
|
|
def _expected_payload(
|
|
self,
|
|
profile: dict[str, Any],
|
|
day: date,
|
|
*,
|
|
profile_row: FitnessProfile | None = None,
|
|
update_baseline: bool = False,
|
|
) -> tuple[dict[str, Any], dict[str, float]]:
|
|
history = self._activity_history(day, days=EXPECTED_LOOKBACK_DAYS)
|
|
expected_daily = compute_expected_targets(profile, history=history)
|
|
if update_baseline:
|
|
self._maybe_update_baseline(profile_row, expected_daily)
|
|
return tdee_expected_to_api(expected_daily), targets_to_api(expected_daily)
|
|
|
|
|
|
def _day_bounds(self, day: date | None = None) -> tuple[datetime, datetime]:
|
|
d = day or datetime.now(timezone.utc).date()
|
|
start = datetime.combine(d, time.min, tzinfo=timezone.utc)
|
|
end = datetime.combine(d, time.max, tzinfo=timezone.utc)
|
|
return start, end
|
|
|
|
def get_daily_summary(self, day: date | None = None) -> dict[str, Any]:
|
|
start, end = self._day_bounds(day)
|
|
profile_row = self._get_profile_row()
|
|
profile_dict = self.get_profile()
|
|
profile = self._profile_for_budget(profile_dict)
|
|
|
|
foods = self.db.scalars(
|
|
select(FoodLog)
|
|
.where(FoodLog.user_id == self.user_id, FoodLog.logged_at >= start, FoodLog.logged_at <= end)
|
|
.order_by(FoodLog.logged_at)
|
|
).all()
|
|
waters = self.db.scalars(
|
|
select(WaterLog)
|
|
.where(WaterLog.user_id == self.user_id, WaterLog.logged_at >= start, WaterLog.logged_at <= end)
|
|
.order_by(WaterLog.logged_at)
|
|
).all()
|
|
workouts_rows = self.db.scalars(
|
|
select(WorkoutLog)
|
|
.where(WorkoutLog.user_id == self.user_id, WorkoutLog.logged_at >= start, WorkoutLog.logged_at <= end)
|
|
.order_by(WorkoutLog.logged_at)
|
|
).all()
|
|
steps_rows = self.db.scalars(
|
|
select(StepLog)
|
|
.where(StepLog.user_id == self.user_id, StepLog.logged_at >= start, StepLog.logged_at <= end)
|
|
.order_by(StepLog.logged_at)
|
|
).all()
|
|
|
|
workouts = [self._workout_to_dict(w) for w in workouts_rows]
|
|
steps_total = sum(s.steps for s in steps_rows)
|
|
|
|
totals = {
|
|
"calories": sum(f.calories for f in foods),
|
|
"protein_g": sum(f.protein_g for f in foods),
|
|
"fat_g": sum(f.fat_g for f in foods),
|
|
"carbs_g": sum(f.carbs_g for f in foods),
|
|
"water_ml": sum(w.amount_ml for w in waters),
|
|
"steps": steps_total,
|
|
}
|
|
|
|
daily = compute_daily_targets(
|
|
profile,
|
|
steps_total=steps_total,
|
|
workouts=workouts,
|
|
)
|
|
targets = targets_to_api(daily)
|
|
target_day = day or datetime.now(timezone.utc).date()
|
|
tdee_expected, targets_expected = self._expected_payload(
|
|
profile,
|
|
target_day,
|
|
profile_row=profile_row,
|
|
update_baseline=target_day == datetime.now(timezone.utc).date(),
|
|
)
|
|
if profile_row is not None and target_day == datetime.now(timezone.utc).date():
|
|
self.db.commit()
|
|
|
|
return {
|
|
"date": target_day.isoformat(),
|
|
"profile_configured": profile_row is not None,
|
|
"totals": totals,
|
|
"targets": targets,
|
|
"targets_expected": targets_expected,
|
|
"tdee_breakdown": tdee_breakdown_to_api(daily),
|
|
"tdee_expected": tdee_expected,
|
|
"meals": [self._food_to_dict(f) for f in foods],
|
|
"water": [self._water_to_dict(w) for w in waters],
|
|
"workouts": workouts,
|
|
"steps": [self._step_to_dict(s) for s in steps_rows],
|
|
"steps_total": steps_total,
|
|
}
|
|
|
|
def log_meal(
|
|
self,
|
|
*,
|
|
description: str,
|
|
meal_type: str = "snack",
|
|
calories: float = 0,
|
|
protein_g: float = 0,
|
|
fat_g: float = 0,
|
|
carbs_g: float = 0,
|
|
source: str = "llm",
|
|
estimated: bool = True,
|
|
) -> dict[str, Any]:
|
|
row = FoodLog(
|
|
user_id=self.user_id,
|
|
meal_type=meal_type[:32],
|
|
description=description[:2000],
|
|
calories=calories,
|
|
protein_g=protein_g,
|
|
fat_g=fat_g,
|
|
carbs_g=carbs_g,
|
|
source=source[:32],
|
|
estimated=estimated,
|
|
)
|
|
self.db.add(row)
|
|
self.db.commit()
|
|
self.db.refresh(row)
|
|
return {"ok": True, "meal": self._food_to_dict(row)}
|
|
|
|
def log_water(self, amount_ml: int) -> dict[str, Any]:
|
|
row = WaterLog(user_id=self.user_id, amount_ml=max(0, amount_ml))
|
|
self.db.add(row)
|
|
self.db.commit()
|
|
self.db.refresh(row)
|
|
return {"ok": True, "water": self._water_to_dict(row)}
|
|
|
|
def log_steps(
|
|
self,
|
|
steps: int,
|
|
*,
|
|
active_calories: float | None = None,
|
|
logged_at: datetime | str | None = None,
|
|
day: date | None = None,
|
|
days_ago: int | None = None,
|
|
notes: str = "",
|
|
source: str = "manual",
|
|
) -> dict[str, Any]:
|
|
row = StepLog(
|
|
user_id=self.user_id,
|
|
steps=max(0, int(steps)),
|
|
active_calories=active_calories,
|
|
notes=notes[:2000],
|
|
source=source[:32],
|
|
logged_at=self._resolve_logged_at(
|
|
logged_at=logged_at,
|
|
day=day,
|
|
days_ago=days_ago,
|
|
),
|
|
)
|
|
self.db.add(row)
|
|
self.db.commit()
|
|
self.db.refresh(row)
|
|
return {"ok": True, "step_log": self._step_to_dict(row)}
|
|
|
|
def log_weight(
|
|
self,
|
|
weight_kg: float,
|
|
*,
|
|
body_fat_pct: float | None = None,
|
|
chest_cm: float | None = None,
|
|
waist_cm: float | None = None,
|
|
neck_cm: float | None = None,
|
|
hip_cm: float | None = None,
|
|
notes: str = "",
|
|
recorded_at: datetime | str | None = None,
|
|
day: date | None = None,
|
|
days_ago: int | None = None,
|
|
) -> dict[str, Any]:
|
|
profile = self.get_profile() or {}
|
|
sex = profile.get("sex") or "male"
|
|
height_cm = float(profile.get("height_cm") or 170)
|
|
|
|
computed = compute_body_composition(
|
|
sex=str(sex),
|
|
height_cm=height_cm,
|
|
weight_kg=weight_kg,
|
|
neck_cm=neck_cm,
|
|
waist_cm=waist_cm,
|
|
hip_cm=hip_cm,
|
|
body_fat_pct=body_fat_pct,
|
|
)
|
|
|
|
row = BodyMetric(
|
|
user_id=self.user_id,
|
|
weight_kg=weight_kg,
|
|
body_fat_pct=computed.get("body_fat_pct"),
|
|
body_fat_method=computed.get("body_fat_method"),
|
|
chest_cm=chest_cm,
|
|
waist_cm=waist_cm,
|
|
neck_cm=neck_cm,
|
|
hip_cm=hip_cm,
|
|
whr=computed.get("whr"),
|
|
lbm_kg=computed.get("lbm_kg"),
|
|
ffmi=computed.get("ffmi"),
|
|
notes=notes[:1000],
|
|
recorded_at=self._resolve_logged_at(
|
|
logged_at=recorded_at,
|
|
day=day,
|
|
days_ago=days_ago,
|
|
),
|
|
)
|
|
self.db.add(row)
|
|
profile_row = self._get_profile_row()
|
|
if profile_row:
|
|
profile_row.weight_kg = weight_kg
|
|
targets = compute_targets(
|
|
{
|
|
"sex": profile_row.sex,
|
|
"age": profile_row.age,
|
|
"height_cm": profile_row.height_cm,
|
|
"weight_kg": weight_kg,
|
|
"goal": profile_row.goal,
|
|
"neat_base_kcal": profile_row.neat_base_kcal,
|
|
}
|
|
)
|
|
profile_row.calorie_target = targets["calorie_target"]
|
|
profile_row.protein_g = targets["protein_g"]
|
|
profile_row.fat_g = targets["fat_g"]
|
|
profile_row.carbs_g = targets["carbs_g"]
|
|
profile_row.water_l = targets["water_l"]
|
|
self.db.commit()
|
|
self.db.refresh(row)
|
|
metric = self._body_metric_to_dict(row)
|
|
return {
|
|
"ok": True,
|
|
"metric": metric,
|
|
"computed": {
|
|
"body_fat_pct": computed.get("body_fat_pct"),
|
|
"body_fat_method": computed.get("body_fat_method"),
|
|
"whr": computed.get("whr"),
|
|
"lbm_kg": computed.get("lbm_kg"),
|
|
"ffmi": computed.get("ffmi"),
|
|
"warnings": computed.get("warnings") or [],
|
|
},
|
|
}
|
|
|
|
def log_workout(
|
|
self,
|
|
*,
|
|
title: str,
|
|
notes: str = "",
|
|
duration_min: int | None = None,
|
|
exercises: list[dict[str, Any]] | None = None,
|
|
active_calories: float | None = None,
|
|
total_calories: float | None = None,
|
|
steps: int | None = None,
|
|
activity_type: str | None = None,
|
|
met: float | None = None,
|
|
logged_at: datetime | str | None = None,
|
|
day: date | None = None,
|
|
days_ago: int | None = None,
|
|
) -> dict[str, Any]:
|
|
profile = self.get_profile() or {}
|
|
weight_kg = float(profile.get("weight_kg") or 70)
|
|
|
|
if active_calories is None and duration_min and met is not None:
|
|
active_calories = round(met * weight_kg * (float(duration_min) / 60.0), 1)
|
|
elif active_calories is None and duration_min:
|
|
draft = {
|
|
"title": title,
|
|
"notes": notes,
|
|
"activity_type": activity_type,
|
|
"met": met,
|
|
"duration_min": duration_min,
|
|
}
|
|
active_calories = estimate_workout_active_kcal(draft, weight_kg=weight_kg) or None
|
|
|
|
row = WorkoutLog(
|
|
user_id=self.user_id,
|
|
title=title[:255],
|
|
notes=notes[:2000],
|
|
duration_min=duration_min,
|
|
active_calories=active_calories,
|
|
total_calories=total_calories,
|
|
steps=steps,
|
|
exercises_json=json.dumps(exercises or [], ensure_ascii=False),
|
|
logged_at=self._resolve_logged_at(
|
|
logged_at=logged_at,
|
|
day=day,
|
|
days_ago=days_ago,
|
|
),
|
|
)
|
|
self.db.add(row)
|
|
self.db.commit()
|
|
self.db.refresh(row)
|
|
return {"ok": True, "workout": self._workout_to_dict(row)}
|
|
|
|
def get_workout_stats(
|
|
self,
|
|
*,
|
|
days: int = 7,
|
|
end_day: date | None = None,
|
|
) -> dict[str, Any]:
|
|
days = max(1, min(days, 90))
|
|
end = end_day or datetime.now(timezone.utc).date()
|
|
start = end - timedelta(days=days - 1)
|
|
start_dt, _ = self._day_bounds(start)
|
|
_, end_dt = self._day_bounds(end)
|
|
|
|
rows = self.db.scalars(
|
|
select(WorkoutLog)
|
|
.where(WorkoutLog.user_id == self.user_id, WorkoutLog.logged_at >= start_dt, WorkoutLog.logged_at <= end_dt)
|
|
.order_by(WorkoutLog.logged_at)
|
|
).all()
|
|
|
|
profile = self.get_profile() or {}
|
|
weight_kg = float(profile.get("weight_kg") or 70)
|
|
weekly_target = 3
|
|
|
|
count = len(rows)
|
|
duration_min = sum(r.duration_min or 0 for r in rows)
|
|
active_kcal = round(
|
|
sum(
|
|
estimate_workout_active_kcal(self._workout_to_dict(r), weight_kg=weight_kg)
|
|
for r in rows
|
|
),
|
|
1,
|
|
)
|
|
|
|
days_with_workout: set[date] = set()
|
|
for row in rows:
|
|
if row.logged_at:
|
|
days_with_workout.add(row.logged_at.astimezone(timezone.utc).date())
|
|
|
|
streak = 0
|
|
cursor = end
|
|
while cursor >= start:
|
|
if cursor in days_with_workout:
|
|
streak += 1
|
|
cursor -= timedelta(days=1)
|
|
else:
|
|
break
|
|
|
|
return {
|
|
"days": days,
|
|
"start_date": start.isoformat(),
|
|
"end_date": end.isoformat(),
|
|
"count": count,
|
|
"duration_min": duration_min,
|
|
"active_kcal": active_kcal,
|
|
"weekly_target": weekly_target,
|
|
"streak": streak,
|
|
}
|
|
|
|
|
|
def list_body_metrics(self, limit: int = 30) -> list[dict[str, Any]]:
|
|
rows = self.db.scalars(
|
|
select(BodyMetric).where(BodyMetric.user_id == self.user_id).order_by(BodyMetric.recorded_at.desc()).limit(limit)
|
|
).all()
|
|
return [self._body_metric_to_dict(r) for r in rows]
|
|
|
|
def delete_food_log(self, log_id: int) -> bool:
|
|
row = self.db.get(FoodLog, log_id)
|
|
if not row or row.user_id != self.user_id:
|
|
return False
|
|
self.db.delete(row)
|
|
self.db.commit()
|
|
return True
|
|
|
|
def delete_water_log(self, log_id: int) -> bool:
|
|
row = self.db.get(WaterLog, log_id)
|
|
if not row or row.user_id != self.user_id:
|
|
return False
|
|
self.db.delete(row)
|
|
self.db.commit()
|
|
return True
|
|
|
|
def delete_workout_log(self, log_id: int) -> bool:
|
|
row = self.db.get(WorkoutLog, log_id)
|
|
if not row or row.user_id != self.user_id:
|
|
return False
|
|
self.db.delete(row)
|
|
self.db.commit()
|
|
return True
|
|
|
|
def delete_step_log(self, log_id: int) -> bool:
|
|
row = self.db.get(StepLog, log_id)
|
|
if not row or row.user_id != self.user_id:
|
|
return False
|
|
self.db.delete(row)
|
|
self.db.commit()
|
|
return True
|
|
|
|
def list_reminders(self) -> list[dict[str, Any]]:
|
|
rows = self.db.scalars(select(FitnessReminder).where(FitnessReminder.user_id == self.user_id).order_by(FitnessReminder.kind)).all()
|
|
return [self._reminder_to_dict(r) for r in rows]
|
|
|
|
def set_reminder(
|
|
self,
|
|
kind: str,
|
|
*,
|
|
enabled: bool | None = None,
|
|
hour: int | None = None,
|
|
minute: int | None = None,
|
|
interval_hours: int | None = None,
|
|
) -> dict[str, Any]:
|
|
row = self.db.scalar(
|
|
select(FitnessReminder).where(FitnessReminder.user_id == self.user_id, FitnessReminder.kind == kind)
|
|
)
|
|
if not row:
|
|
row = FitnessReminder(user_id=self.user_id, kind=kind)
|
|
self.db.add(row)
|
|
if enabled is not None:
|
|
row.enabled = enabled
|
|
if hour is not None:
|
|
row.hour = hour
|
|
if minute is not None:
|
|
row.minute = minute
|
|
if interval_hours is not None:
|
|
row.interval_hours = interval_hours
|
|
self.db.commit()
|
|
self.db.refresh(row)
|
|
return {"ok": True, "reminder": self._reminder_to_dict(row)}
|
|
|
|
def calc_one_rm(self, weight_kg: float, reps: int) -> dict[str, Any]:
|
|
return {"ok": True, "one_rm_kg": one_rep_max(weight_kg, reps)}
|
|
|
|
def get_history(
|
|
self,
|
|
*,
|
|
days: int = 7,
|
|
end_day: date | None = None,
|
|
include_tdee_breakdown: bool = True,
|
|
) -> dict[str, Any]:
|
|
days = max(1, min(days, 90))
|
|
end = end_day or datetime.now(timezone.utc).date()
|
|
start = end - timedelta(days=days - 1)
|
|
summaries: list[dict[str, Any]] = []
|
|
|
|
for offset in range(days):
|
|
d = start + timedelta(days=offset)
|
|
full = self.get_daily_summary(d)
|
|
totals = full["totals"]
|
|
has_data = bool(full["meals"] or full["water"] or full["workouts"] or full["steps"])
|
|
item: dict[str, Any] = {
|
|
"date": full["date"],
|
|
"has_data": has_data,
|
|
"totals": totals,
|
|
"targets": full["targets"],
|
|
"targets_expected": full.get("targets_expected"),
|
|
"meal_count": len(full["meals"]),
|
|
"workout_count": len(full["workouts"]),
|
|
}
|
|
if include_tdee_breakdown:
|
|
item["tdee_breakdown"] = full.get("tdee_breakdown")
|
|
item["tdee_expected"] = full.get("tdee_expected")
|
|
summaries.append(item)
|
|
|
|
return {
|
|
"start_date": start.isoformat(),
|
|
"end_date": end.isoformat(),
|
|
"days": days,
|
|
"summaries": summaries,
|
|
}
|
|
|
|
def snapshot(self) -> dict[str, Any]:
|
|
today = datetime.now(timezone.utc).date()
|
|
return {
|
|
"profile": self.get_profile(),
|
|
"today": self.get_daily_summary(today),
|
|
"history": self.get_history(days=7, end_day=today),
|
|
"workout_stats": self.get_workout_stats(days=7, end_day=today),
|
|
"body_metrics": self.list_body_metrics(limit=10),
|
|
"reminders": self.list_reminders(),
|
|
}
|
|
|
|
def get_charts(
|
|
self,
|
|
*,
|
|
weeks: int = 52,
|
|
trend: bool = True,
|
|
end_day: date | None = None,
|
|
) -> dict[str, Any]:
|
|
from app.fitness.charts import build_fitness_charts
|
|
|
|
return build_fitness_charts(
|
|
self.db,
|
|
self.user_id,
|
|
weeks=weeks,
|
|
trend=trend,
|
|
end_day=end_day,
|
|
)
|
|
|
|
@staticmethod
|
|
def _food_to_dict(row: FoodLog) -> dict[str, Any]:
|
|
return {
|
|
"id": row.id,
|
|
"meal_type": row.meal_type,
|
|
"description": row.description,
|
|
"calories": row.calories,
|
|
"protein_g": row.protein_g,
|
|
"fat_g": row.fat_g,
|
|
"carbs_g": row.carbs_g,
|
|
"source": row.source,
|
|
"estimated": row.estimated,
|
|
"logged_at": row.logged_at.isoformat() if row.logged_at else None,
|
|
}
|
|
|
|
@staticmethod
|
|
def _water_to_dict(row: WaterLog) -> dict[str, Any]:
|
|
return {
|
|
"id": row.id,
|
|
"amount_ml": row.amount_ml,
|
|
"logged_at": row.logged_at.isoformat() if row.logged_at else None,
|
|
}
|
|
|
|
@staticmethod
|
|
def _step_to_dict(row: StepLog) -> dict[str, Any]:
|
|
return {
|
|
"id": row.id,
|
|
"steps": row.steps,
|
|
"active_calories": row.active_calories,
|
|
"source": row.source,
|
|
"notes": row.notes,
|
|
"logged_at": row.logged_at.isoformat() if row.logged_at else None,
|
|
}
|
|
|
|
@staticmethod
|
|
def _workout_to_dict(row: WorkoutLog) -> dict[str, Any]:
|
|
try:
|
|
exercises = json.loads(row.exercises_json or "[]")
|
|
except json.JSONDecodeError:
|
|
exercises = []
|
|
return {
|
|
"id": row.id,
|
|
"title": row.title,
|
|
"notes": row.notes,
|
|
"duration_min": row.duration_min,
|
|
"active_calories": row.active_calories,
|
|
"total_calories": row.total_calories,
|
|
"steps": row.steps,
|
|
"exercises": exercises,
|
|
"logged_at": row.logged_at.isoformat() if row.logged_at else None,
|
|
}
|
|
|
|
@staticmethod
|
|
def _reminder_to_dict(row: FitnessReminder) -> dict[str, Any]:
|
|
return {
|
|
"id": row.id,
|
|
"kind": row.kind,
|
|
"hour": row.hour,
|
|
"minute": row.minute,
|
|
"interval_hours": row.interval_hours,
|
|
"enabled": row.enabled,
|
|
"last_fired_at": row.last_fired_at.isoformat() if row.last_fired_at else None,
|
|
}
|