Lars
8fc7d9c1c4
- Updated the `get_body_history_viz_bundle` function to retrieve the two most recent circumference measurements for improved data accuracy. - Refactored the handling of previous measurement data to ensure comprehensive interpolation for body metrics. - Modified frontend labels in the `buildBodyKpiTiles` function to provide clearer descriptions in German, enhancing user understanding of body metrics.
469 lines
17 KiB
Python
469 lines
17 KiB
Python
"""
|
|
Layer 2b: Structured body history / Verlauf «Körper» bundle.
|
|
|
|
Single source for Verlauf-UI: series + Kennzahlen + Interpretation tiles.
|
|
All queries use the same tables as Layer 1 / Layer 2a body placeholders.
|
|
|
|
See: placeholder_registrations/body_metrics.py, body_extras.py
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from datetime import date, datetime, timedelta
|
|
from typing import Any, Dict, List, Optional, Tuple
|
|
|
|
from db import get_db, get_cursor, r2d
|
|
from data_layer.body_interpretation import get_body_interpretation_tiles
|
|
from data_layer.utils import safe_float
|
|
|
|
|
|
def _cutoff_sql(days: int) -> Optional[str]:
|
|
if days >= 9999:
|
|
return None
|
|
return (datetime.now() - timedelta(days=days)).strftime("%Y-%m-%d")
|
|
|
|
|
|
def _rolling_avg(rows: List[Dict[str, Any]], key: str, window: int) -> List[Dict[str, Any]]:
|
|
out: List[Dict[str, Any]] = []
|
|
for i, d in enumerate(rows):
|
|
sl = rows[max(0, i - window + 1) : i + 1]
|
|
vals: List[float] = []
|
|
for x in sl:
|
|
v = safe_float(x.get(key))
|
|
if v is not None:
|
|
vals.append(v)
|
|
if not vals:
|
|
out.append({**d, f"{key}_avg": None})
|
|
continue
|
|
avg = round(sum(vals) / len(vals), 1)
|
|
out.append({**d, f"{key}_avg": avg})
|
|
return out
|
|
|
|
|
|
def _iso(d: Any) -> Optional[str]:
|
|
if d is None:
|
|
return None
|
|
if hasattr(d, "isoformat"):
|
|
return d.isoformat()
|
|
return str(d)[:10]
|
|
|
|
|
|
def get_body_history_viz_bundle(profile_id: str, days: int) -> Dict[str, Any]:
|
|
"""
|
|
Returns chart-ready series and interpretation tiles for the body history tab.
|
|
|
|
Args:
|
|
profile_id: profiles.id
|
|
days: analysis window (use >= 9999 for full history)
|
|
|
|
Tables: weight_log, caliper_log, circumference_log, profiles
|
|
"""
|
|
cutoff = _cutoff_sql(days)
|
|
|
|
with get_db() as conn:
|
|
cur = get_cursor(conn)
|
|
cur.execute(
|
|
"""
|
|
SELECT id, sex, height, dob, goal_weight, goal_bf_pct
|
|
FROM profiles WHERE id = %s
|
|
""",
|
|
(profile_id,),
|
|
)
|
|
pr = r2d(cur.fetchone())
|
|
if not pr:
|
|
return {
|
|
"confidence": "insufficient",
|
|
"message": "Profil nicht gefunden",
|
|
"profile": {},
|
|
"weight": {},
|
|
"caliper": {},
|
|
"circumference": {},
|
|
"interpretation_tiles": [],
|
|
"meta": {},
|
|
}
|
|
|
|
profile_ui = {
|
|
"sex": pr.get("sex") or "m",
|
|
"height": safe_float(pr.get("height")) or 178.0,
|
|
"goal_weight_kg": safe_float(pr.get("goal_weight")),
|
|
"goal_bf_pct": safe_float(pr.get("goal_bf_pct")),
|
|
}
|
|
|
|
# ── Weight (same window as Verlauf-Filter) ────────────────────────────
|
|
if cutoff:
|
|
cur.execute(
|
|
"""
|
|
SELECT date, weight FROM weight_log
|
|
WHERE profile_id = %s AND date >= %s
|
|
ORDER BY date ASC
|
|
""",
|
|
(profile_id, cutoff),
|
|
)
|
|
else:
|
|
cur.execute(
|
|
"""
|
|
SELECT date, weight FROM weight_log
|
|
WHERE profile_id = %s
|
|
ORDER BY date ASC
|
|
""",
|
|
(profile_id,),
|
|
)
|
|
wrows = [r2d(r) for r in cur.fetchall()]
|
|
w_points = [
|
|
{"date": r["date"], "weight": safe_float(r["weight"])}
|
|
for r in wrows
|
|
if r.get("weight") is not None
|
|
]
|
|
w_with_avg7 = _rolling_avg([dict(x) for x in w_points], "weight", 7)
|
|
w_with_avg14 = _rolling_avg([dict(x) for x in w_points], "weight", 14)
|
|
weight_series: List[Dict[str, Any]] = []
|
|
for i, base in enumerate(w_points):
|
|
weight_series.append(
|
|
{
|
|
"date": _iso(base["date"]),
|
|
"weight": base["weight"],
|
|
"avg7": w_with_avg7[i].get("weight_avg") if i < len(w_with_avg7) else None,
|
|
"avg14": w_with_avg14[i].get("weight_avg") if i < len(w_with_avg14) else None,
|
|
}
|
|
)
|
|
|
|
ws = [p["weight"] for p in w_points if p.get("weight") is not None]
|
|
overall_avg = round(sum(ws) / len(ws), 1) if len(ws) else None
|
|
min_w = min(ws) if ws else None
|
|
max_w = max(ws) if ws else None
|
|
|
|
today = datetime.now().date()
|
|
trend_periods: List[Dict[str, Any]] = []
|
|
for span in (7, 30, 90):
|
|
cut = today - timedelta(days=span)
|
|
per = [p for p in w_points if p["date"] >= cut]
|
|
if len(per) >= 2:
|
|
diff = round(float(per[-1]["weight"]) - float(per[0]["weight"]), 1)
|
|
trend_periods.append({"label": f"{span}T", "diff_kg": diff, "count": len(per)})
|
|
|
|
# ── Caliper series ───────────────────────────────────────────────────
|
|
if cutoff:
|
|
cur.execute(
|
|
"""
|
|
SELECT date, body_fat_pct, lean_mass, fat_mass
|
|
FROM caliper_log
|
|
WHERE profile_id = %s
|
|
AND body_fat_pct IS NOT NULL
|
|
AND date >= %s
|
|
ORDER BY date ASC
|
|
""",
|
|
(profile_id, cutoff),
|
|
)
|
|
else:
|
|
cur.execute(
|
|
"""
|
|
SELECT date, body_fat_pct, lean_mass, fat_mass
|
|
FROM caliper_log
|
|
WHERE profile_id = %s AND body_fat_pct IS NOT NULL
|
|
ORDER BY date ASC
|
|
""",
|
|
(profile_id,),
|
|
)
|
|
cal_rows = [r2d(r) for r in cur.fetchall()]
|
|
caliper_series = [
|
|
{
|
|
"date": _iso(r["date"]),
|
|
"body_fat_pct": safe_float(r.get("body_fat_pct")),
|
|
"lean_mass": safe_float(r.get("lean_mass")),
|
|
}
|
|
for r in cal_rows
|
|
]
|
|
|
|
# Latest / prev caliper in window (for interpretation)
|
|
if cutoff:
|
|
cur.execute(
|
|
"""
|
|
SELECT date, body_fat_pct, lean_mass
|
|
FROM caliper_log
|
|
WHERE profile_id = %s AND date >= %s
|
|
ORDER BY date DESC
|
|
LIMIT 2
|
|
""",
|
|
(profile_id, cutoff),
|
|
)
|
|
else:
|
|
cur.execute(
|
|
"""
|
|
SELECT date, body_fat_pct, lean_mass
|
|
FROM caliper_log
|
|
WHERE profile_id = %s
|
|
ORDER BY date DESC
|
|
LIMIT 2
|
|
""",
|
|
(profile_id,),
|
|
)
|
|
cal_latest_rows = [r2d(r) for r in cur.fetchall()]
|
|
latest_cal = cal_latest_rows[0] if cal_latest_rows else None
|
|
prev_cal = cal_latest_rows[1] if len(cal_latest_rows) > 1 else None
|
|
|
|
# ── Circumference rows ───────────────────────────────────────────────
|
|
if cutoff:
|
|
cur.execute(
|
|
"""
|
|
SELECT date, c_chest, c_waist, c_hip, c_belly
|
|
FROM circumference_log
|
|
WHERE profile_id = %s AND date >= %s
|
|
ORDER BY date ASC
|
|
""",
|
|
(profile_id, cutoff),
|
|
)
|
|
else:
|
|
cur.execute(
|
|
"""
|
|
SELECT date, c_chest, c_waist, c_hip, c_belly
|
|
FROM circumference_log
|
|
WHERE profile_id = %s
|
|
ORDER BY date ASC
|
|
""",
|
|
(profile_id,),
|
|
)
|
|
cir_rows = [r2d(r) for r in cur.fetchall()]
|
|
|
|
if cutoff:
|
|
cur.execute(
|
|
"""
|
|
SELECT date, c_chest, c_waist, c_hip, c_belly
|
|
FROM circumference_log
|
|
WHERE profile_id = %s AND date >= %s
|
|
ORDER BY date DESC
|
|
LIMIT 2
|
|
""",
|
|
(profile_id, cutoff),
|
|
)
|
|
else:
|
|
cur.execute(
|
|
"""
|
|
SELECT date, c_chest, c_waist, c_hip, c_belly
|
|
FROM circumference_log
|
|
WHERE profile_id = %s
|
|
ORDER BY date DESC
|
|
LIMIT 2
|
|
""",
|
|
(profile_id,),
|
|
)
|
|
circ_latest_desc = [r2d(r) for r in cur.fetchall()]
|
|
latest_circ_row = circ_latest_desc[0] if circ_latest_desc else None
|
|
prev_circ_row = circ_latest_desc[1] if len(circ_latest_desc) > 1 else None
|
|
|
|
# Latest weight in window
|
|
latest_w = w_points[-1] if w_points else None
|
|
|
|
# ── Proportion & index (computed from L1 rows only) ─────────────────────
|
|
prop_base: List[Dict[str, Any]] = []
|
|
for r in cir_rows:
|
|
ch = safe_float(r.get("c_chest"))
|
|
wa = safe_float(r.get("c_waist"))
|
|
if ch is None or wa is None:
|
|
continue
|
|
belly = safe_float(r.get("c_belly"))
|
|
prop_base.append(
|
|
{
|
|
"date": _iso(r["date"]),
|
|
"v_taper_cm": round(ch - wa, 1),
|
|
"belly_cm": belly,
|
|
}
|
|
)
|
|
prop_chart = _rolling_avg([dict(x) for x in prop_base], "v_taper_cm", 3) if len(prop_base) >= 2 else []
|
|
for i, row in enumerate(prop_chart):
|
|
row["belly_cm"] = prop_base[i].get("belly_cm")
|
|
|
|
fb_first: Dict[str, Optional[float]] = {"chest": None, "waist": None, "belly": None}
|
|
for r in cir_rows:
|
|
if fb_first["chest"] is None and r.get("c_chest") is not None:
|
|
fb_first["chest"] = safe_float(r["c_chest"])
|
|
if fb_first["waist"] is None and r.get("c_waist") is not None:
|
|
fb_first["waist"] = safe_float(r["c_waist"])
|
|
if fb_first["belly"] is None and r.get("c_belly") is not None:
|
|
fb_first["belly"] = safe_float(r["c_belly"])
|
|
|
|
index_series: List[Dict[str, Any]] = []
|
|
for r in cir_rows:
|
|
idx_row: Dict[str, Any] = {"date": _iso(r["date"])}
|
|
cc = safe_float(r.get("c_chest"))
|
|
ww = safe_float(r.get("c_waist"))
|
|
bb = safe_float(r.get("c_belly"))
|
|
if cc is not None and fb_first["chest"]:
|
|
idx_row["chest_idx"] = round(cc / fb_first["chest"] * 100, 1)
|
|
else:
|
|
idx_row["chest_idx"] = None
|
|
if ww is not None and fb_first["waist"]:
|
|
idx_row["waist_idx"] = round(ww / fb_first["waist"] * 100, 1)
|
|
else:
|
|
idx_row["waist_idx"] = None
|
|
if bb is not None and fb_first["belly"]:
|
|
idx_row["belly_idx"] = round(bb / fb_first["belly"] * 100, 1)
|
|
else:
|
|
idx_row["belly_idx"] = None
|
|
index_series.append(idx_row)
|
|
|
|
idx_nonempty = sum(
|
|
1
|
|
for row in index_series
|
|
if row.get("chest_idx") is not None
|
|
or row.get("waist_idx") is not None
|
|
or row.get("belly_idx") is not None
|
|
)
|
|
|
|
fallback_circ = [
|
|
{
|
|
"date": _iso(r["date"]),
|
|
"waist": safe_float(r.get("c_waist")),
|
|
"hip": safe_float(r.get("c_hip")),
|
|
"belly": safe_float(r.get("c_belly")),
|
|
}
|
|
for r in cir_rows
|
|
if r.get("c_waist") or r.get("c_hip") or r.get("c_belly")
|
|
]
|
|
|
|
# ── Merge measurement for interpretation ────────────────────────────────
|
|
measurement: Dict[str, Any] = {}
|
|
if latest_cal:
|
|
measurement.update(
|
|
{
|
|
"date": latest_cal.get("date"),
|
|
"body_fat_pct": safe_float(latest_cal.get("body_fat_pct")),
|
|
"lean_mass": safe_float(latest_cal.get("lean_mass")),
|
|
}
|
|
)
|
|
if latest_circ_row:
|
|
measurement["c_waist"] = safe_float(latest_circ_row.get("c_waist"))
|
|
measurement["c_hip"] = safe_float(latest_circ_row.get("c_hip"))
|
|
measurement["c_belly"] = safe_float(latest_circ_row.get("c_belly"))
|
|
if latest_w:
|
|
measurement["weight"] = safe_float(latest_w.get("weight"))
|
|
# Referenzdatum für „aktuell“: neueste verfügbare Quelle (Caliper > Umfang > Gewicht)
|
|
if not measurement.get("date"):
|
|
if latest_circ_row and latest_circ_row.get("date"):
|
|
measurement["date"] = latest_circ_row.get("date")
|
|
elif latest_w and latest_w.get("date"):
|
|
measurement["date"] = latest_w.get("date")
|
|
|
|
# Vorperiode: vorherige Caliper-Zeile + vorherige Umfangsmessung + vorheriges Gewicht (w_points[-2])
|
|
prev_for_interp: Optional[Dict[str, Any]] = {}
|
|
if prev_cal:
|
|
prev_for_interp["date"] = prev_cal.get("date")
|
|
prev_for_interp["body_fat_pct"] = safe_float(prev_cal.get("body_fat_pct"))
|
|
prev_for_interp["lean_mass"] = safe_float(prev_cal.get("lean_mass"))
|
|
if prev_circ_row:
|
|
prev_for_interp["c_waist"] = safe_float(prev_circ_row.get("c_waist"))
|
|
prev_for_interp["c_hip"] = safe_float(prev_circ_row.get("c_hip"))
|
|
prev_for_interp["c_belly"] = safe_float(prev_circ_row.get("c_belly"))
|
|
if not prev_for_interp.get("date") and prev_circ_row.get("date"):
|
|
prev_for_interp["date"] = prev_circ_row.get("date")
|
|
if len(w_points) >= 2:
|
|
prev_for_interp["weight"] = safe_float(w_points[-2].get("weight"))
|
|
if not prev_for_interp.get("date") and w_points[-2].get("date"):
|
|
prev_for_interp["date"] = w_points[-2].get("date")
|
|
|
|
if not prev_for_interp:
|
|
prev_for_interp = None
|
|
else:
|
|
# Mindestens ein vergleichbares Feld zur aktuellen Messung
|
|
has_cmp = any(
|
|
prev_for_interp.get(k) is not None
|
|
for k in ("body_fat_pct", "lean_mass", "weight", "c_waist", "c_belly")
|
|
)
|
|
if not has_cmp:
|
|
prev_for_interp = None
|
|
|
|
tiles = get_body_interpretation_tiles(measurement, profile_ui, prev_for_interp)
|
|
|
|
last_dates: List[date] = []
|
|
if w_points:
|
|
last_dates.append(w_points[-1]["date"])
|
|
if latest_cal and latest_cal.get("date"):
|
|
d = latest_cal["date"]
|
|
if isinstance(d, str):
|
|
d = datetime.fromisoformat(d[:10]).date()
|
|
last_dates.append(d)
|
|
if latest_circ_row and latest_circ_row.get("date"):
|
|
d = latest_circ_row["date"]
|
|
if isinstance(d, str):
|
|
d = datetime.fromisoformat(d[:10]).date()
|
|
last_dates.append(d)
|
|
last_updated = max(last_dates).isoformat() if last_dates else None
|
|
|
|
bf_cat = None
|
|
if measurement.get("body_fat_pct") is not None:
|
|
# simple label bucket (aligned with frontend BF_CATEGORIES order)
|
|
bf = float(measurement["body_fat_pct"])
|
|
sex = profile_ui["sex"]
|
|
if sex == "f":
|
|
labels = ["Essenziell", "Athletisch", "Fit", "Durchschnitt", "Übergewicht"]
|
|
bounds = [14, 21, 25, 32, 1000]
|
|
else:
|
|
labels = ["Essenziell", "Athletisch", "Fit", "Durchschnitt", "Übergewicht"]
|
|
bounds = [6, 14, 18, 25, 1000]
|
|
for i, b in enumerate(bounds):
|
|
if bf <= b:
|
|
bf_cat = labels[i]
|
|
break
|
|
|
|
summary = {
|
|
"weight_kg": measurement.get("weight"),
|
|
"body_fat_pct": measurement.get("body_fat_pct"),
|
|
"lean_mass_kg": measurement.get("lean_mass"),
|
|
"whr": (
|
|
round(measurement["c_waist"] / measurement["c_hip"], 2)
|
|
if measurement.get("c_waist") and measurement.get("c_hip")
|
|
else None
|
|
),
|
|
"whtr": (
|
|
round(measurement["c_waist"] / profile_ui["height"], 2)
|
|
if measurement.get("c_waist") and profile_ui.get("height")
|
|
else None
|
|
),
|
|
"ffmi": None,
|
|
"bf_category_label": bf_cat,
|
|
}
|
|
if measurement.get("lean_mass") and profile_ui.get("height"):
|
|
hm = float(profile_ui["height"]) / 100.0
|
|
summary["ffmi"] = round(float(measurement["lean_mass"]) / (hm**2), 1)
|
|
|
|
return {
|
|
"confidence": "high" if w_points or caliper_series or cir_rows else "insufficient",
|
|
"days_requested": days,
|
|
"last_updated": last_updated,
|
|
"profile": profile_ui,
|
|
"summary": summary,
|
|
"weight": {
|
|
"series": weight_series,
|
|
"overall_avg_kg": overall_avg,
|
|
"min_kg": min_w,
|
|
"max_kg": max_w,
|
|
"trend_periods": trend_periods,
|
|
"data_points": len(w_points),
|
|
"related_placeholder_keys": [
|
|
"weight_aktuell",
|
|
"weight_trend",
|
|
"weight_7d_median",
|
|
"weight_28d_slope",
|
|
"weight_90d_slope",
|
|
],
|
|
},
|
|
"caliper": {
|
|
"series": caliper_series,
|
|
"data_points": len(caliper_series),
|
|
"related_placeholder_keys": ["caliper_summary", "fm_28d_change", "lbm_28d_change"],
|
|
},
|
|
"circumference": {
|
|
"proportion_series": prop_chart,
|
|
"index_series": index_series,
|
|
"index_usable": idx_nonempty >= 2 and any(v for v in fb_first.values()),
|
|
"fallback_multiline": fallback_circ,
|
|
"has_chest_waist": len(prop_base) >= 2,
|
|
"related_placeholder_keys": ["circ_summary", "waist_hip_ratio", "waist_28d_delta"],
|
|
},
|
|
"interpretation_tiles": tiles,
|
|
"meta": {
|
|
"layer_1": "data_layer.body_viz + data_layer.body_interpretation",
|
|
"layer_2b": "This bundle — sole numeric source for Verlauf Körper charts/tiles",
|
|
"layer_2a_alignment": "Tiles carry related_placeholder_keys; metrics from same tables as body_metrics placeholders",
|
|
},
|
|
}
|