""" 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