442 lines
14 KiB
Plaintext
442 lines
14 KiB
Plaintext
import json
|
|
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 (
|
|
BodyMetric,
|
|
FitnessProfile,
|
|
FitnessReminder,
|
|
FoodLog,
|
|
WaterLog,
|
|
WorkoutLog,
|
|
)
|
|
from app.fitness.calculators import compute_targets, one_rep_max
|
|
|
|
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):
|
|
self.db = db
|
|
|
|
def _get_profile_row(self) -> FitnessProfile | None:
|
|
return self.db.scalar(select(FitnessProfile).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_to_dict(self, row: FitnessProfile) -> dict[str, Any]:
|
|
targets = compute_targets(
|
|
{
|
|
"sex": row.sex,
|
|
"age": row.age,
|
|
"height_cm": row.height_cm,
|
|
"weight_kg": row.weight_kg,
|
|
"activity_level": row.activity_level,
|
|
"goal": row.goal,
|
|
}
|
|
)
|
|
return {
|
|
"sex": row.sex,
|
|
"age": row.age,
|
|
"height_cm": row.height_cm,
|
|
"weight_kg": row.weight_kg,
|
|
"activity_level": row.activity_level,
|
|
"goal": row.goal,
|
|
"target_weight_kg": row.target_weight_kg,
|
|
"weekly_workouts": row.weekly_workouts,
|
|
"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()
|
|
self.db.add(row)
|
|
self.db.flush()
|
|
|
|
for key in (
|
|
"sex", "age", "height_cm", "weight_kg", "activity_level",
|
|
"goal", "target_weight_kg", "weekly_workouts",
|
|
):
|
|
if key in updates and updates[key] is not None:
|
|
setattr(row, key, updates[key])
|
|
|
|
targets = compute_targets(
|
|
{
|
|
"sex": row.sex,
|
|
"age": row.age,
|
|
"height_cm": row.height_cm,
|
|
"weight_kg": row.weight_kg,
|
|
"activity_level": row.activity_level,
|
|
"goal": row.goal,
|
|
}
|
|
)
|
|
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)).all()
|
|
if existing:
|
|
return
|
|
for item in DEFAULT_REMINDERS:
|
|
self.db.add(FitnessReminder(**item))
|
|
|
|
def calc_targets(self, params: dict[str, Any]) -> dict[str, Any]:
|
|
return compute_targets(params)
|
|
|
|
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 = self.get_profile()
|
|
|
|
foods = self.db.scalars(
|
|
select(FoodLog)
|
|
.where(FoodLog.logged_at >= start, FoodLog.logged_at <= end)
|
|
.order_by(FoodLog.logged_at)
|
|
).all()
|
|
waters = self.db.scalars(
|
|
select(WaterLog)
|
|
.where(WaterLog.logged_at >= start, WaterLog.logged_at <= end)
|
|
.order_by(WaterLog.logged_at)
|
|
).all()
|
|
workouts = self.db.scalars(
|
|
select(WorkoutLog)
|
|
.where(WorkoutLog.logged_at >= start, WorkoutLog.logged_at <= end)
|
|
.order_by(WorkoutLog.logged_at)
|
|
).all()
|
|
|
|
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),
|
|
}
|
|
|
|
targets = profile or {
|
|
"calorie_target": 2000,
|
|
"protein_g": 140,
|
|
"fat_g": 65,
|
|
"carbs_g": 200,
|
|
"water_l": 2.5,
|
|
}
|
|
|
|
return {
|
|
"date": (day or datetime.now(timezone.utc).date()).isoformat(),
|
|
"profile_configured": profile is not None,
|
|
"totals": totals,
|
|
"targets": {
|
|
"calories": targets.get("calorie_target", 2000),
|
|
"protein_g": targets.get("protein_g", 140),
|
|
"fat_g": targets.get("fat_g", 65),
|
|
"carbs_g": targets.get("carbs_g", 200),
|
|
"water_ml": targets.get("water_l", 2.5) * 1000,
|
|
},
|
|
"meals": [self._food_to_dict(f) for f in foods],
|
|
"water": [self._water_to_dict(w) for w in waters],
|
|
"workouts": [self._workout_to_dict(w) for w in workouts],
|
|
}
|
|
|
|
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(
|
|
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(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_weight(
|
|
self,
|
|
weight_kg: float,
|
|
*,
|
|
body_fat_pct: float | None = None,
|
|
chest_cm: float | None = None,
|
|
waist_cm: float | None = None,
|
|
notes: str = "",
|
|
) -> dict[str, Any]:
|
|
row = BodyMetric(
|
|
weight_kg=weight_kg,
|
|
body_fat_pct=body_fat_pct,
|
|
chest_cm=chest_cm,
|
|
waist_cm=waist_cm,
|
|
notes=notes[:1000],
|
|
)
|
|
self.db.add(row)
|
|
profile = self._get_profile_row()
|
|
if profile:
|
|
profile.weight_kg = weight_kg
|
|
targets = compute_targets(
|
|
{
|
|
"sex": profile.sex,
|
|
"age": profile.age,
|
|
"height_cm": profile.height_cm,
|
|
"weight_kg": weight_kg,
|
|
"activity_level": profile.activity_level,
|
|
"goal": profile.goal,
|
|
}
|
|
)
|
|
profile.calorie_target = targets["calorie_target"]
|
|
profile.protein_g = targets["protein_g"]
|
|
profile.fat_g = targets["fat_g"]
|
|
profile.carbs_g = targets["carbs_g"]
|
|
profile.water_l = targets["water_l"]
|
|
self.db.commit()
|
|
self.db.refresh(row)
|
|
return {
|
|
"ok": True,
|
|
"metric": {
|
|
"id": row.id,
|
|
"weight_kg": row.weight_kg,
|
|
"recorded_at": row.recorded_at.isoformat() if row.recorded_at else None,
|
|
},
|
|
}
|
|
|
|
def log_workout(
|
|
self,
|
|
*,
|
|
title: str,
|
|
notes: str = "",
|
|
duration_min: int | None = None,
|
|
exercises: list[dict[str, Any]] | None = None,
|
|
) -> dict[str, Any]:
|
|
row = WorkoutLog(
|
|
title=title[:255],
|
|
notes=notes[:2000],
|
|
duration_min=duration_min,
|
|
exercises_json=json.dumps(exercises or [], ensure_ascii=False),
|
|
)
|
|
self.db.add(row)
|
|
self.db.commit()
|
|
self.db.refresh(row)
|
|
return {"ok": True, "workout": self._workout_to_dict(row)}
|
|
|
|
def list_body_metrics(self, limit: int = 30) -> list[dict[str, Any]]:
|
|
rows = self.db.scalars(
|
|
select(BodyMetric).order_by(BodyMetric.recorded_at.desc()).limit(limit)
|
|
).all()
|
|
return [
|
|
{
|
|
"id": r.id,
|
|
"weight_kg": r.weight_kg,
|
|
"body_fat_pct": r.body_fat_pct,
|
|
"chest_cm": r.chest_cm,
|
|
"waist_cm": r.waist_cm,
|
|
"notes": r.notes,
|
|
"recorded_at": r.recorded_at.isoformat() if r.recorded_at else None,
|
|
}
|
|
for r in rows
|
|
]
|
|
|
|
def delete_food_log(self, log_id: int) -> bool:
|
|
row = self.db.get(FoodLog, log_id)
|
|
if not row:
|
|
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:
|
|
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:
|
|
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).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.kind == kind)
|
|
)
|
|
if not row:
|
|
row = FitnessReminder(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,
|
|
) -> 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"])
|
|
summaries.append(
|
|
{
|
|
"date": full["date"],
|
|
"has_data": has_data,
|
|
"totals": totals,
|
|
"targets": full["targets"],
|
|
"meal_count": len(full["meals"]),
|
|
"workout_count": len(full["workouts"]),
|
|
}
|
|
)
|
|
|
|
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),
|
|
"body_metrics": self.list_body_metrics(limit=10),
|
|
"reminders": self.list_reminders(),
|
|
}
|
|
|
|
@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 _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,
|
|
"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,
|
|
}
|