65 lines
2.5 KiB
Python
65 lines
2.5 KiB
Python
|
"""
|
||
|
|
Layer 1 Hilfslogik: Ernährung + Gewicht + Caliper (forward-filled Magermasse).
|
||
|
|
|
||
|
|
Genutzt von Layer 2b (nutrition_viz) und vom Router GET /api/nutrition/correlations.
|
||
|
|
"""
|
||
|
|
|
||
|
|
from __future__ import annotations
|
||
|
|
|
||
|
|
from typing import Any, Dict, List, Optional
|
||
|
|
|
||
|
|
from db import get_db, get_cursor, r2d
|
||
|
|
from caliper_composition import compute_lean_fat_kg, nearest_weight_kg_from_map
|
||
|
|
|
||
|
|
|
||
|
|
def build_merged_daily_nutrition_body_rows(profile_id: str) -> List[Dict[str, Any]]:
|
||
|
|
"""
|
||
|
|
Pro Kalendertag: Makros aus nutrition_log, Gewicht, forward-filled Caliper (lean_mass, bf%).
|
||
|
|
Gleiche Semantik wie bisher ``GET /api/nutrition/correlations``.
|
||
|
|
"""
|
||
|
|
with get_db() as conn:
|
||
|
|
cur = get_cursor(conn)
|
||
|
|
cur.execute("SELECT * FROM nutrition_log WHERE profile_id=%s ORDER BY date", (profile_id,))
|
||
|
|
nutr = {r["date"]: r2d(r) for r in cur.fetchall()}
|
||
|
|
cur.execute("SELECT date, weight FROM weight_log WHERE profile_id=%s ORDER BY date", (profile_id,))
|
||
|
|
wlog = {r["date"]: r["weight"] for r in cur.fetchall()}
|
||
|
|
cur.execute(
|
||
|
|
"SELECT date, lean_mass, body_fat_pct FROM caliper_log WHERE profile_id=%s ORDER BY date",
|
||
|
|
(profile_id,),
|
||
|
|
)
|
||
|
|
cals = sorted([r2d(r) for r in cur.fetchall()], key=lambda x: x["date"])
|
||
|
|
|
||
|
|
all_dates = sorted(set(list(nutr.keys()) + list(wlog.keys())))
|
||
|
|
mi = 0
|
||
|
|
last_cal: Dict[str, Any] = {}
|
||
|
|
cal_by_date: Dict[Any, Dict[str, Any]] = {}
|
||
|
|
for d in all_dates:
|
||
|
|
while mi < len(cals) and cals[mi]["date"] <= d:
|
||
|
|
last_cal = cals[mi]
|
||
|
|
mi += 1
|
||
|
|
if last_cal:
|
||
|
|
cal_by_date[d] = last_cal
|
||
|
|
|
||
|
|
result: List[Dict[str, Any]] = []
|
||
|
|
for d in all_dates:
|
||
|
|
if d not in nutr and d not in wlog:
|
||
|
|
continue
|
||
|
|
row: Dict[str, Any] = {"date": d}
|
||
|
|
if d in nutr:
|
||
|
|
for k in ("kcal", "protein_g", "fat_g", "carbs_g"):
|
||
|
|
v = nutr[d].get(k)
|
||
|
|
row[k] = float(v) if v is not None else None
|
||
|
|
if d in wlog:
|
||
|
|
row["weight"] = float(wlog[d])
|
||
|
|
if d in cal_by_date:
|
||
|
|
lm = cal_by_date[d].get("lean_mass")
|
||
|
|
bf = cal_by_date[d].get("body_fat_pct")
|
||
|
|
if bf is not None and lm is None:
|
||
|
|
wkg = nearest_weight_kg_from_map(wlog, d)
|
||
|
|
if wkg is not None:
|
||
|
|
lm, _fat = compute_lean_fat_kg(wkg, float(bf))
|
||
|
|
row["lean_mass"] = float(lm) if lm is not None else None
|
||
|
|
row["body_fat_pct"] = float(bf) if bf is not None else None
|
||
|
|
result.append(row)
|
||
|
|
return result
|