Lars
97dbb0f80b
- Introduced the `history_overview_viz` widget to the dashboard, allowing users to visualize consolidated history data across various metrics. - Updated widget configuration to include `history_overview_viz` in the allowed widgets and added validation for its configuration. - Enhanced the widget catalog with details for the new `history_overview_viz` entry. - Implemented default values and validation logic for the widget's configuration, ensuring proper handling of user inputs. - Added tests to ensure proper validation of the `history_overview_viz` widget configuration. - Bumped application version to reflect the addition of the new widget.
252 lines
9.0 KiB
Python
252 lines
9.0 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.correlation_chart_payloads import (
|
||
build_lbm_protein_correlation_chart_payload,
|
||
build_load_vitals_correlation_chart_payload,
|
||
build_recovery_performance_chart_payload,
|
||
build_weight_energy_correlation_chart_payload,
|
||
)
|
||
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],
|
||
},
|
||
},
|
||
"chart_payloads": {
|
||
"c1_weight_energy": build_weight_energy_correlation_chart_payload(profile_id, 14),
|
||
"c2_protein_lbm": build_lbm_protein_correlation_chart_payload(profile_id, 14),
|
||
"c3_load_vitals": build_load_vitals_correlation_chart_payload(profile_id, 14),
|
||
"c4_recovery_performance": build_recovery_performance_chart_payload(profile_id),
|
||
},
|
||
"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
|