Lars
7ac9752c3d
- Refactored the `calculate_lag_correlation` function to normalize lag payloads and improve correlation calculations for various nutrition metrics. - Introduced a new function `build_nutrition_correlation_heuristic_items` to generate heuristic insights based on merged nutrition data, enhancing user understanding of dietary impacts on weight and body composition. - Updated the `get_nutrition_history_viz_bundle` function to include daily calorie balance and protein vs. lean mass data, providing a comprehensive view of nutrition trends. - Enhanced the frontend to visualize calorie balance and protein vs. lean mass insights, improving the user experience with clear graphical representations of dietary correlations.
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
|