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.
240 lines
8.4 KiB
Python
240 lines
8.4 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
|
||
from data_layer.utils import safe_float
|
||
|
||
|
||
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:
|
||
a1 = abs(safe_float(c3_hrv.get("correlation"), 0.0))
|
||
a2 = abs(safe_float(c3_rhr.get("correlation"), 0.0))
|
||
c3 = c3_hrv if a1 >= a2 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
|