Lars
0365d9eb52
- Added `safe_float` utility to enhance float handling in correlation calculations, preventing potential errors. - Refactored lag correlation logic in `get_history_overview_viz_bundle` to utilize absolute values safely, improving accuracy in metric comparisons. - Enhanced nutrition body merge logic to ensure proper date handling and data integrity, optimizing the retrieval of nutrition and weight logs. - Introduced new functions in the frontend for processing lag details, improving the visualization of correlation data in the History page.
86 lines
3.1 KiB
Python
86 lines
3.1 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 as_date, 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: Dict[Any, Dict[str, Any]] = {}
|
|
for r in cur.fetchall():
|
|
rd = r2d(r)
|
|
dk = as_date(rd.get("date"))
|
|
if dk is not None:
|
|
nutr[dk] = rd
|
|
cur.execute("SELECT date, weight FROM weight_log WHERE profile_id=%s ORDER BY date", (profile_id,))
|
|
wlog: Dict[Any, Any] = {}
|
|
for r in cur.fetchall():
|
|
rd = r2d(r)
|
|
dk = as_date(rd.get("date"))
|
|
if dk is not None:
|
|
wlog[dk] = rd["weight"]
|
|
cur.execute(
|
|
"SELECT date, lean_mass, body_fat_pct FROM caliper_log WHERE profile_id=%s ORDER BY date",
|
|
(profile_id,),
|
|
)
|
|
cals = [r2d(r) for r in cur.fetchall()]
|
|
cals = sorted(
|
|
[c for c in cals if as_date(c.get("date")) is not None],
|
|
key=lambda x: as_date(x["date"]),
|
|
)
|
|
|
|
# Alle Keys sind datetime.date — vermeidet TypeError bei Vergleichen (str vs date)
|
|
all_dates = sorted(set(nutr.keys()) | set(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):
|
|
cd = as_date(cals[mi].get("date"))
|
|
if cd is None:
|
|
mi += 1
|
|
continue
|
|
if cd > d:
|
|
break
|
|
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
|