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.
241 lines
8.3 KiB
Python
241 lines
8.3 KiB
Python
"""
|
||
Layer 2b: Gesamtansicht «Verlauf» — komponiert nur Bundles aus body-, nutrition-, fitness-, recovery_viz.
|
||
|
||
Issue #53: keine parallele Business-Logik; ein Router-Endpoint liefert diese Zusammenfassung.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
from typing import Any, Dict, List, Optional
|
||
|
||
from data_layer.body_viz import get_body_history_viz_bundle
|
||
from data_layer.correlations import calculate_lag_correlation, calculate_top_drivers
|
||
from data_layer.fitness_viz import get_fitness_dashboard_viz_bundle
|
||
from data_layer.nutrition_viz import get_nutrition_history_viz_bundle
|
||
from data_layer.recovery_viz import get_recovery_dashboard_viz_bundle
|
||
|
||
|
||
def _take_kpis(tiles: Any, max_n: int = 4) -> List[Dict[str, Any]]:
|
||
if not isinstance(tiles, list):
|
||
return []
|
||
out: List[Dict[str, Any]] = []
|
||
for t in tiles[:max_n]:
|
||
if not isinstance(t, dict):
|
||
continue
|
||
out.append(
|
||
{
|
||
"key": t.get("key"),
|
||
"category": t.get("category"),
|
||
"icon": t.get("icon"),
|
||
"value": t.get("value"),
|
||
"sublabel": t.get("sublabel"),
|
||
"status": t.get("status"),
|
||
"verdict": t.get("verdict"),
|
||
}
|
||
)
|
||
return out
|
||
|
||
|
||
def _short_body_interpretation_tiles(tiles: Any, max_n: int = 3) -> List[Dict[str, Any]]:
|
||
"""Körper-Interpretationskacheln (keine KPI-Kacheln)."""
|
||
if not isinstance(tiles, list):
|
||
return []
|
||
out: List[Dict[str, Any]] = []
|
||
for t in tiles[:max_n]:
|
||
if not isinstance(t, dict):
|
||
continue
|
||
det = str(t.get("detail") or "")
|
||
if len(det) > 140:
|
||
det = det[:137] + "…"
|
||
out.append(
|
||
{
|
||
"title": t.get("title") or t.get("category") or "Hinweis",
|
||
"detail": det,
|
||
"status": t.get("status"),
|
||
}
|
||
)
|
||
return out
|
||
|
||
|
||
def _take_insights(items: Any, max_n: int = 2) -> List[Dict[str, Any]]:
|
||
if not isinstance(items, list):
|
||
return []
|
||
out: List[Dict[str, Any]] = []
|
||
for it in items[:max_n]:
|
||
if not isinstance(it, dict):
|
||
continue
|
||
out.append(
|
||
{
|
||
"title": it.get("title") or it.get("title_de"),
|
||
"body": it.get("body") or it.get("detail") or it.get("message"),
|
||
"tone": it.get("tone") or it.get("status"),
|
||
}
|
||
)
|
||
return out
|
||
|
||
|
||
def get_history_overview_viz_bundle(profile_id: str, days: int) -> Dict[str, Any]:
|
||
"""
|
||
Kompakte Übersicht für den ersten Reiter «Gesamtansicht»: KPI-Kurzformen + Lag-Korrelationen (C1–C4).
|
||
"""
|
||
eff = max(7, min(int(days), 9999))
|
||
body = get_body_history_viz_bundle(profile_id, eff)
|
||
nutr = get_nutrition_history_viz_bundle(profile_id, eff)
|
||
fit = get_fitness_dashboard_viz_bundle(profile_id, eff)
|
||
rec = get_recovery_dashboard_viz_bundle(profile_id, eff)
|
||
|
||
c1 = calculate_lag_correlation(profile_id, "energy_balance", "weight", 14)
|
||
c2 = calculate_lag_correlation(profile_id, "protein", "lbm", 14)
|
||
c3_hrv = calculate_lag_correlation(profile_id, "load", "hrv", 14)
|
||
c3_rhr = calculate_lag_correlation(profile_id, "load", "rhr", 14)
|
||
c3 = None
|
||
if c3_hrv and c3_rhr:
|
||
c3 = (
|
||
c3_hrv
|
||
if abs(float(c3_hrv.get("correlation") or 0)) >= abs(float(c3_rhr.get("correlation") or 0))
|
||
else c3_rhr
|
||
)
|
||
if c3 is c3_hrv:
|
||
c3 = dict(c3)
|
||
c3["metric"] = "HRV"
|
||
else:
|
||
c3 = dict(c3_rhr)
|
||
c3["metric"] = "RHR"
|
||
elif c3_hrv:
|
||
c3 = dict(c3_hrv)
|
||
c3["metric"] = "HRV"
|
||
elif c3_rhr:
|
||
c3 = dict(c3_rhr)
|
||
c3["metric"] = "RHR"
|
||
|
||
drivers = calculate_top_drivers(profile_id)
|
||
|
||
b_sum = body.get("summary") if isinstance(body.get("summary"), dict) else {}
|
||
last_w = b_sum.get("weight_kg")
|
||
|
||
fs = fit.get("summary") if isinstance(fit.get("summary"), dict) else {}
|
||
if fit.get("has_activity_entries"):
|
||
ac = int(fs.get("activity_count") or 0)
|
||
fitness_line = f"{ac} Trainingseinheiten im gewählten Fenster"
|
||
else:
|
||
fitness_line = fit.get("message") or "Keine Trainingsdaten"
|
||
|
||
drv_list = drivers if isinstance(drivers, list) else []
|
||
|
||
return {
|
||
"days_requested": days,
|
||
"effective_window_days": eff,
|
||
"confidence": _overview_confidence(body, nutr, fit, rec),
|
||
"sections": [
|
||
{
|
||
"id": "body",
|
||
"title": "Körper",
|
||
"tab_id": "body",
|
||
"summary_line": (
|
||
f"Letztes Gewicht: {last_w} kg"
|
||
if last_w is not None
|
||
else "Keine Gewichtsdaten im Fenster"
|
||
),
|
||
"interpretation_short": _short_body_interpretation_tiles(body.get("interpretation_tiles"), 3),
|
||
},
|
||
{
|
||
"id": "nutrition",
|
||
"title": "Ernährung",
|
||
"tab_id": "nutrition",
|
||
"summary_line": (
|
||
f"Ø {round(float((nutr.get('summary') or {}).get('kcal_avg') or 0))} kcal/Tag"
|
||
if nutr.get("has_nutrition_entries")
|
||
else (nutr.get("message") or "Keine Ernährungsdaten")
|
||
),
|
||
"kpi_short": _take_kpis(nutr.get("kpi_tiles"), 4),
|
||
"heuristic_short": (nutr.get("nutrition_correlation_heuristics") or [])[:2],
|
||
},
|
||
{
|
||
"id": "fitness",
|
||
"title": "Fitness",
|
||
"tab_id": "activity",
|
||
"summary_line": fitness_line,
|
||
"kpi_short": _take_kpis(fit.get("kpi_tiles"), 4),
|
||
"insights_short": _take_insights(fit.get("progress_insights"), 2),
|
||
},
|
||
{
|
||
"id": "recovery",
|
||
"title": "Erholung",
|
||
"tab_id": "activity",
|
||
"summary_line": "Schlaf & Vitalwerte"
|
||
if rec.get("has_recovery_data")
|
||
else (rec.get("message") or "Keine Erholungsdaten"),
|
||
"kpi_short": _take_kpis(rec.get("kpi_tiles"), 4),
|
||
"insights_short": _take_insights(rec.get("progress_insights"), 2),
|
||
},
|
||
],
|
||
"lag_correlations": {
|
||
"weight_energy": _compact_lag("C1 Energiebilanz ↔ Gewicht", c1),
|
||
"protein_lbm": _compact_lag("C2 Protein ↔ Magermasse", c2),
|
||
"load_vitals": _compact_lag(
|
||
f"C3 Last ↔ {(c3 or {}).get('metric') or 'Vital'}",
|
||
c3,
|
||
extra_keys=("metric",),
|
||
),
|
||
"recovery_performance": {
|
||
"label": "C4 Top-Treiber (Einflussfaktoren)",
|
||
"drivers": drv_list[:8],
|
||
},
|
||
},
|
||
"meta": {
|
||
"layer_1": "composed_metrics",
|
||
"layer_2b": "history_overview_viz",
|
||
"issue": "53-history-overview",
|
||
"sources": {
|
||
"body": "body_viz",
|
||
"nutrition": "nutrition_viz",
|
||
"fitness": "fitness_viz",
|
||
"recovery": "recovery_viz",
|
||
"lag": "correlations.calculate_lag_correlation",
|
||
"drivers": "correlations.calculate_top_drivers",
|
||
},
|
||
},
|
||
}
|
||
|
||
|
||
def _overview_confidence(b: Dict, n: Dict, f: Dict, r: Dict) -> str:
|
||
scores = []
|
||
for x in (b, n, f, r):
|
||
c = x.get("confidence")
|
||
if c == "high":
|
||
scores.append(3)
|
||
elif c == "medium":
|
||
scores.append(2)
|
||
elif c == "low":
|
||
scores.append(1)
|
||
else:
|
||
scores.append(0)
|
||
s = sum(scores) / max(len(scores), 1)
|
||
if s >= 2.5:
|
||
return "high"
|
||
if s >= 1.5:
|
||
return "medium"
|
||
return "low"
|
||
|
||
|
||
def _compact_lag(
|
||
label: str,
|
||
payload: Optional[Dict[str, Any]],
|
||
extra_keys: tuple = (),
|
||
) -> Dict[str, Any]:
|
||
if not payload:
|
||
return {"label": label, "available": False}
|
||
out: Dict[str, Any] = {
|
||
"label": label,
|
||
"available": payload.get("correlation") is not None,
|
||
"correlation": payload.get("correlation"),
|
||
"best_lag_days": payload.get("best_lag_days", payload.get("best_lag")),
|
||
"confidence": payload.get("confidence"),
|
||
"interpretation": payload.get("interpretation", ""),
|
||
"data_points": payload.get("data_points"),
|
||
}
|
||
for k in extra_keys:
|
||
if k in payload:
|
||
out[k] = payload[k]
|
||
return out
|