Lars
3ab5dae130
- Introduced the `viz_bundle` block type to the report profile schema, allowing for the inclusion of bundled visualizations in PDF reports. - Updated the `build_structured_report_pdf` function to handle `VizBundleBlock` and append its content to the report. - Enhanced the report catalog API to include details for the new `viz_bundle` block type. - Added configuration editors for various visualization bundles in the frontend settings page. - Updated tests to validate the new `viz_bundle` functionality and ensure proper handling of report profiles. - Bumped application version to reflect these enhancements.
387 lines
18 KiB
Python
387 lines
18 KiB
Python
"""
|
||
Layer-2b Verlauf-Bundles → PDF-Abschnitte (KPIs + eingebettete Chart-Payloads).
|
||
|
||
Gleiche Datenquellen und Config-Validierung wie Dashboard-Widgets (dashboard_widget_config).
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import io
|
||
import logging
|
||
from typing import Any
|
||
|
||
from reportlab.lib.units import mm
|
||
from reportlab.platypus import Image as RLImage
|
||
from reportlab.platypus import Paragraph, Spacer
|
||
from xml.sax.saxutils import escape
|
||
|
||
from dashboard_widget_config import validate_widget_entry_config
|
||
from data_layer.body_viz import get_body_history_viz_bundle
|
||
from data_layer.fitness_viz import get_fitness_dashboard_viz_bundle
|
||
from data_layer.history_overview_viz import get_history_overview_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
|
||
from report_chart_plotting import chart_payload_to_png
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
BUNDLE_HEADINGS: dict[str, str] = {
|
||
"body_history_viz": "Körper — Kennwerte & Verlauf",
|
||
"nutrition_history_viz": "Ernährung — Kennwerte & Charts",
|
||
"fitness_history_viz": "Fitness / Training",
|
||
"recovery_history_viz": "Erholung & Vitalwerte",
|
||
"history_overview_viz": "Gesamtübersicht & Korrelationen",
|
||
}
|
||
|
||
|
||
def _add_chart_to_story(story: list, styles: dict, payload: dict[str, Any], caption: str | None = None) -> None:
|
||
meta = payload.get("metadata") or {}
|
||
if meta.get("confidence") == "insufficient":
|
||
msg = escape(meta.get("message") or "Keine Daten")
|
||
story.append(Paragraph(f"<i>{msg}</i>", styles["Normal"]))
|
||
story.append(Spacer(1, 2 * mm))
|
||
return
|
||
if caption:
|
||
story.append(Paragraph(f"<b>{escape(caption)}</b>", styles["Normal"]))
|
||
try:
|
||
png = chart_payload_to_png(payload)
|
||
story.append(RLImage(io.BytesIO(png), width=170 * mm, height=85 * mm))
|
||
except Exception as e:
|
||
logger.warning("bundle chart png: %s", e)
|
||
story.append(Paragraph("Diagramm konnte nicht gerendert werden.", styles["Normal"]))
|
||
story.append(Spacer(1, 4 * mm))
|
||
|
||
|
||
def _append_interpretation_tiles(story: list, styles: dict, tiles: list[dict[str, Any]]) -> None:
|
||
if not tiles:
|
||
return
|
||
story.append(Paragraph("<b>Einschätzungen</b>", styles["Heading4"]))
|
||
for t in tiles:
|
||
cat = escape(str(t.get("category") or t.get("title") or "—"))
|
||
title = t.get("title")
|
||
detail = t.get("detail")
|
||
val = t.get("value")
|
||
parts = [f"<b>{cat}</b>"]
|
||
if title and str(title) != str(cat):
|
||
parts.append(escape(str(title)))
|
||
if val is not None and val != "":
|
||
parts.append(f"({escape(str(val))})")
|
||
story.append(Paragraph(" — ".join(parts), styles["Normal"]))
|
||
if detail:
|
||
story.append(Paragraph(escape(str(detail)[:500]), styles["BodyText"]))
|
||
story.append(Spacer(1, 3 * mm))
|
||
|
||
|
||
def _append_kpi_tiles_fitness_nutreco(story: list, styles: dict, tiles: list[dict[str, Any]], compact: bool) -> None:
|
||
if not tiles:
|
||
return
|
||
use = tiles[:4] if compact else tiles
|
||
story.append(Paragraph("<b>KPI-Kacheln</b>", styles["Heading4"]))
|
||
for t in use:
|
||
cat = escape(str(t.get("category") or t.get("title") or "—"))
|
||
val = escape(str(t.get("value") or "—"))
|
||
sub = t.get("sublabel") or t.get("body")
|
||
line = f"• <b>{cat}</b>: {val}"
|
||
if sub:
|
||
line += f" — {escape(str(sub)[:180])}"
|
||
story.append(Paragraph(line, styles["Normal"]))
|
||
story.append(Spacer(1, 3 * mm))
|
||
|
||
|
||
def _append_insights_lines(story: list, styles: dict, insights: list[dict[str, Any]], label: str) -> None:
|
||
if not insights:
|
||
return
|
||
story.append(Paragraph(f"<b>{escape(label)}</b>", styles["Heading4"]))
|
||
for item in insights:
|
||
title = item.get("title") or item.get("heading")
|
||
body = item.get("body") or item.get("text")
|
||
if title:
|
||
story.append(Paragraph(escape(str(title)), styles["Normal"]))
|
||
if body:
|
||
story.append(Paragraph(escape(str(body)[:600]), styles["BodyText"]))
|
||
story.append(Spacer(1, 2 * mm))
|
||
|
||
|
||
def _weight_series_payload(bundle_weight: dict[str, Any]) -> dict[str, Any] | None:
|
||
series = bundle_weight.get("series") or []
|
||
if len(series) < 2:
|
||
return None
|
||
labels = [str(p.get("date") or "") for p in series]
|
||
datasets: list[dict[str, Any]] = [
|
||
{
|
||
"label": "Gewicht (kg)",
|
||
"data": [safe_float(p.get("weight")) for p in series],
|
||
"borderColor": "#1D9E75",
|
||
}
|
||
]
|
||
if any(p.get("avg7") is not None for p in series):
|
||
datasets.append(
|
||
{
|
||
"label": "Ø 7T",
|
||
"data": [safe_float(p.get("avg7")) for p in series],
|
||
"borderColor": "#378ADD",
|
||
}
|
||
)
|
||
return {"chart_type": "line", "data": {"labels": labels, "datasets": datasets}, "metadata": {"confidence": "high"}}
|
||
|
||
|
||
def _line_payload_from_points(
|
||
points: list[dict[str, Any]],
|
||
x_key: str,
|
||
y_key: str,
|
||
label: str,
|
||
) -> dict[str, Any] | None:
|
||
if len(points) < 2:
|
||
return None
|
||
labels = [str(p.get(x_key) or "") for p in points]
|
||
ys = [safe_float(p.get(y_key)) for p in points]
|
||
return {
|
||
"chart_type": "line",
|
||
"data": {
|
||
"labels": labels,
|
||
"datasets": [{"label": label, "data": ys, "borderColor": "#1D9E75"}],
|
||
},
|
||
"metadata": {"confidence": "high"},
|
||
}
|
||
|
||
|
||
def _append_body_bundle(story: list, styles: dict, profile_id: str, cfg: dict[str, Any]) -> None:
|
||
days = int(cfg.get("chart_days") or 30)
|
||
bundle = get_body_history_viz_bundle(profile_id, days)
|
||
story.append(Paragraph(escape(BUNDLE_HEADINGS["body_history_viz"]), styles["Heading2"]))
|
||
if bundle.get("confidence") == "insufficient":
|
||
story.append(Paragraph(escape(bundle.get("message") or "Keine Körperdaten"), styles["Normal"]))
|
||
story.append(Spacer(1, 4 * mm))
|
||
return
|
||
summ = bundle.get("summary") or {}
|
||
if summ:
|
||
w = summ.get("weight_kg")
|
||
bf = summ.get("body_fat_pct")
|
||
parts = []
|
||
if w is not None:
|
||
parts.append(f"Gewicht: {w} kg")
|
||
if bf is not None:
|
||
parts.append(f"KF%: {bf}")
|
||
if parts:
|
||
story.append(Paragraph(escape(" · ".join(parts)), styles["Normal"]))
|
||
story.append(Spacer(1, 2 * mm))
|
||
if cfg.get("show_kpis", True):
|
||
_append_interpretation_tiles(story, styles, bundle.get("interpretation_tiles") or [])
|
||
w = bundle.get("weight") or {}
|
||
if cfg.get("show_weight_chart", True):
|
||
pl = _weight_series_payload(w)
|
||
if pl:
|
||
_add_chart_to_story(story, styles, pl, "Gewicht")
|
||
cal = bundle.get("caliper") or {}
|
||
if cfg.get("show_body_fat_chart", False):
|
||
ser = cal.get("series") or []
|
||
pts = [{"date": p.get("date"), "y": p.get("body_fat_pct")} for p in ser if p.get("body_fat_pct") is not None]
|
||
pl = _line_payload_from_points(pts, "date", "y", "KF %")
|
||
if pl:
|
||
_add_chart_to_story(story, styles, pl, "Körperfett (Caliper)")
|
||
circ = bundle.get("circumference") or {}
|
||
if cfg.get("show_proportion_chart", False):
|
||
prop = circ.get("proportion_series") or []
|
||
pts = [{"date": p.get("date"), "y": p.get("v_taper_cm")} for p in prop if p.get("v_taper_cm") is not None]
|
||
pl = _line_payload_from_points(pts, "date", "y", "V-Taper (cm)")
|
||
if pl:
|
||
_add_chart_to_story(story, styles, pl, "Proportion (Brust–Taille)")
|
||
if cfg.get("show_circumference_index_chart", False):
|
||
idx = circ.get("index_series") or []
|
||
if len(idx) >= 2:
|
||
labels = [str(p.get("date") or "") for p in idx]
|
||
ds: list[dict[str, Any]] = []
|
||
for key, lab, col in (
|
||
("waist_idx", "Taille-Index", "#D85A30"),
|
||
("chest_idx", "Brust-Index", "#1D9E75"),
|
||
("belly_idx", "Bauch-Index", "#378ADD"),
|
||
):
|
||
ys = [safe_float(p.get(key)) for p in idx]
|
||
if any(v is not None for v in ys):
|
||
ds.append({"label": lab, "data": ys, "borderColor": col})
|
||
if ds:
|
||
pl = {"chart_type": "line", "data": {"labels": labels, "datasets": ds}, "metadata": {"confidence": "high"}}
|
||
_add_chart_to_story(story, styles, pl, "Umfang-Indizes")
|
||
story.append(Spacer(1, 2 * mm))
|
||
|
||
|
||
def _append_nutrition_bundle(story: list, styles: dict, profile_id: str, cfg: dict[str, Any]) -> None:
|
||
days = int(cfg.get("chart_days") or 30)
|
||
bundle = get_nutrition_history_viz_bundle(profile_id, days)
|
||
story.append(Paragraph(escape(BUNDLE_HEADINGS["nutrition_history_viz"]), styles["Heading2"]))
|
||
if not bundle.get("has_nutrition_entries"):
|
||
story.append(Paragraph(escape(bundle.get("message") or "Keine Ernährungsdaten"), styles["Normal"]))
|
||
story.append(Spacer(1, 4 * mm))
|
||
return
|
||
compact = cfg.get("kpi_detail") == "compact"
|
||
if cfg.get("show_kpis", True):
|
||
_append_kpi_tiles_fitness_nutreco(story, styles, bundle.get("kpi_tiles") or [], compact)
|
||
if cfg.get("show_heuristics", False):
|
||
h = bundle.get("nutrition_correlation_heuristics") or []
|
||
for item in h:
|
||
t = item.get("text") or item.get("title")
|
||
if t:
|
||
story.append(Paragraph(f"• {escape(str(t))}", styles["Normal"]))
|
||
story.append(Spacer(1, 2 * mm))
|
||
charts = bundle.get("chart_payloads") or {}
|
||
if cfg.get("show_calorie_balance_chart", False) or cfg.get("show_energy_protein_charts", False):
|
||
pl = charts.get("energy_balance")
|
||
if pl:
|
||
_add_chart_to_story(story, styles, pl, "Energiebilanz")
|
||
if cfg.get("show_energy_protein_charts", False) or cfg.get("show_protein_lean_chart", False):
|
||
pl = charts.get("protein_adequacy")
|
||
if pl:
|
||
_add_chart_to_story(story, styles, pl, "Protein-Adäquanz")
|
||
pl2 = charts.get("nutrition_adherence")
|
||
if pl2:
|
||
_add_chart_to_story(story, styles, pl2, "Ernährungs-Adherence")
|
||
if cfg.get("show_macro_distribution_pair", False) or cfg.get("show_macro_daily_bars", False):
|
||
wm = bundle.get("weekly_macro_chart")
|
||
if isinstance(wm, dict) and wm.get("chart_type"):
|
||
_add_chart_to_story(story, styles, wm, "Makros (wöchentlich)")
|
||
kw = bundle.get("kcal_vs_weight") or {}
|
||
if cfg.get("show_kcal_vs_weight", False) and kw.get("points"):
|
||
pts = kw["points"]
|
||
if pts:
|
||
pl = _line_payload_from_points(pts, "date", "kcal", "kcal")
|
||
if pl:
|
||
_add_chart_to_story(story, styles, pl, "Kalorien vs. Zeit")
|
||
story.append(Spacer(1, 2 * mm))
|
||
|
||
|
||
def _append_fitness_bundle(story: list, styles: dict, profile_id: str, cfg: dict[str, Any]) -> None:
|
||
days = int(cfg.get("chart_days") or 30)
|
||
bundle = get_fitness_dashboard_viz_bundle(profile_id, days)
|
||
story.append(Paragraph(escape(BUNDLE_HEADINGS["fitness_history_viz"]), styles["Heading2"]))
|
||
if not bundle.get("has_activity_entries"):
|
||
story.append(Paragraph(escape(bundle.get("message") or "Keine Aktivitätsdaten"), styles["Normal"]))
|
||
story.append(Spacer(1, 4 * mm))
|
||
return
|
||
compact = cfg.get("kpi_detail") == "compact"
|
||
if cfg.get("show_kpis", True):
|
||
_append_kpi_tiles_fitness_nutreco(story, styles, bundle.get("kpi_tiles") or [], compact)
|
||
if cfg.get("show_progress_insights", False):
|
||
_append_insights_lines(story, styles, bundle.get("progress_insights") or [], "Einschätzungen")
|
||
charts = bundle.get("charts") or {}
|
||
if cfg.get("show_chart_training_volume", True) and charts.get("training_volume"):
|
||
_add_chart_to_story(story, styles, charts["training_volume"], "Trainingsvolumen")
|
||
if cfg.get("show_chart_training_type_distribution", True) and charts.get("training_type_distribution"):
|
||
_add_chart_to_story(story, styles, charts["training_type_distribution"], "Trainingsarten")
|
||
if cfg.get("show_chart_quality_sessions", False) and charts.get("quality_sessions"):
|
||
_add_chart_to_story(story, styles, charts["quality_sessions"], "Qualitätssessions")
|
||
if cfg.get("show_chart_load_monitoring", False) and charts.get("load_monitoring"):
|
||
_add_chart_to_story(story, styles, charts["load_monitoring"], "Last / ACWR")
|
||
story.append(Spacer(1, 2 * mm))
|
||
|
||
|
||
def _append_recovery_bundle(story: list, styles: dict, profile_id: str, cfg: dict[str, Any]) -> None:
|
||
days = int(cfg.get("chart_days") or 30)
|
||
bundle = get_recovery_dashboard_viz_bundle(profile_id, days)
|
||
story.append(Paragraph(escape(BUNDLE_HEADINGS["recovery_history_viz"]), styles["Heading2"]))
|
||
if not bundle.get("has_recovery_data"):
|
||
story.append(Paragraph(escape(bundle.get("message") or "Keine Erholungsdaten"), styles["Normal"]))
|
||
story.append(Spacer(1, 4 * mm))
|
||
return
|
||
compact = cfg.get("kpi_detail") == "compact"
|
||
if cfg.get("show_kpis", True):
|
||
_append_kpi_tiles_fitness_nutreco(story, styles, bundle.get("kpi_tiles") or [], compact)
|
||
if cfg.get("show_progress_insights", False):
|
||
_append_insights_lines(story, styles, bundle.get("progress_insights") or [], "Einschätzungen")
|
||
charts = bundle.get("charts") or {}
|
||
if cfg.get("show_chart_recovery_score", True) and charts.get("recovery_score"):
|
||
_add_chart_to_story(story, styles, charts["recovery_score"], "Recovery-Score")
|
||
if cfg.get("show_chart_hrv_rhr", True) and charts.get("hrv_rhr"):
|
||
_add_chart_to_story(story, styles, charts["hrv_rhr"], "HRV / RHR")
|
||
if cfg.get("show_chart_sleep_quality", True) and charts.get("sleep_duration_quality"):
|
||
_add_chart_to_story(story, styles, charts["sleep_duration_quality"], "Schlaf Dauer & Qualität")
|
||
if cfg.get("show_chart_sleep_debt", False) and charts.get("sleep_debt"):
|
||
_add_chart_to_story(story, styles, charts["sleep_debt"], "Schlafschuld")
|
||
if cfg.get("show_vitals_extra_trends", False):
|
||
if charts.get("vital_signs_matrix"):
|
||
_add_chart_to_story(story, styles, charts["vital_signs_matrix"], "Vital-Matrix")
|
||
if charts.get("vitals_history"):
|
||
_add_chart_to_story(story, styles, charts["vitals_history"], "Vital-Trends")
|
||
story.append(Spacer(1, 2 * mm))
|
||
|
||
|
||
def _append_history_overview_bundle(story: list, styles: dict, profile_id: str, cfg: dict[str, Any]) -> None:
|
||
days = int(cfg.get("chart_days") or 30)
|
||
bundle = get_history_overview_viz_bundle(profile_id, days)
|
||
story.append(Paragraph(escape(BUNDLE_HEADINGS["history_overview_viz"]), styles["Heading2"]))
|
||
sect_keys = {
|
||
"body": cfg.get("show_section_body", True),
|
||
"nutrition": cfg.get("show_section_nutrition", True),
|
||
"fitness": cfg.get("show_section_fitness", True),
|
||
"recovery": cfg.get("show_section_recovery", True),
|
||
}
|
||
for sec in bundle.get("sections") or []:
|
||
sid = sec.get("id")
|
||
if not sect_keys.get(str(sid), True):
|
||
continue
|
||
title = escape(str(sec.get("title") or sid))
|
||
line = escape(str(sec.get("summary_line") or ""))
|
||
story.append(Paragraph(f"<b>{title}</b>: {line}", styles["Normal"]))
|
||
for it in sec.get("interpretation_short") or []:
|
||
t = it.get("title") if isinstance(it, dict) else None
|
||
if t:
|
||
story.append(Paragraph(f"• {escape(str(t))}", styles["BodyText"]))
|
||
for k in sec.get("kpi_short") or []:
|
||
if isinstance(k, dict):
|
||
cat = k.get("category") or k.get("title")
|
||
val = k.get("value")
|
||
if cat:
|
||
story.append(Paragraph(f"• {escape(str(cat))}: {escape(str(val or ''))}", styles["BodyText"]))
|
||
story.append(Spacer(1, 2 * mm))
|
||
if cfg.get("show_correlation_c1_c3", True) or cfg.get("show_drivers_c4", True):
|
||
lag = bundle.get("lag_correlations") or {}
|
||
we = lag.get("weight_energy") or {}
|
||
if we.get("available") and (we.get("interpretation") or we.get("label")):
|
||
lab = escape(str(we.get("label") or "C1"))
|
||
interp = escape(str(we.get("interpretation") or "").strip())
|
||
if interp:
|
||
story.append(Paragraph(f"{lab}: {interp}", styles["Normal"]))
|
||
charts = bundle.get("chart_payloads") or {}
|
||
if cfg.get("show_correlation_c1_c3", True):
|
||
for key, cap in (
|
||
("c1_weight_energy", "Korrelation Gewicht / Energie"),
|
||
("c2_protein_lbm", "Protein / Magermasse"),
|
||
("c3_load_vitals", "Last / Vitalwerte"),
|
||
):
|
||
pl = charts.get(key)
|
||
if pl:
|
||
_add_chart_to_story(story, styles, pl, cap)
|
||
if cfg.get("show_drivers_c4", True):
|
||
pl = charts.get("c4_recovery_performance")
|
||
if pl:
|
||
_add_chart_to_story(story, styles, pl, "Top-Treiber")
|
||
drv = (bundle.get("lag_correlations") or {}).get("recovery_performance") or {}
|
||
for d in (drv.get("drivers") or [])[:12]:
|
||
if isinstance(d, dict):
|
||
lab = d.get("label") or d.get("factor")
|
||
val = d.get("impact") or d.get("score")
|
||
if lab:
|
||
story.append(Paragraph(f"• {escape(str(lab))}: {escape(str(val or ''))}", styles["Normal"]))
|
||
story.append(Spacer(1, 2 * mm))
|
||
|
||
|
||
def append_viz_bundle_to_story(
|
||
story: list,
|
||
styles: dict,
|
||
profile_id: str,
|
||
bundle_id: str,
|
||
raw_config: dict[str, Any],
|
||
) -> None:
|
||
cfg = validate_widget_entry_config(bundle_id, raw_config)
|
||
if bundle_id == "body_history_viz":
|
||
_append_body_bundle(story, styles, profile_id, cfg)
|
||
elif bundle_id == "nutrition_history_viz":
|
||
_append_nutrition_bundle(story, styles, profile_id, cfg)
|
||
elif bundle_id == "fitness_history_viz":
|
||
_append_fitness_bundle(story, styles, profile_id, cfg)
|
||
elif bundle_id == "recovery_history_viz":
|
||
_append_recovery_bundle(story, styles, profile_id, cfg)
|
||
elif bundle_id == "history_overview_viz":
|
||
_append_history_overview_bundle(story, styles, profile_id, cfg)
|
||
else:
|
||
story.append(Paragraph(escape(f"Unbekanntes Bundle: {bundle_id}"), styles["Normal"]))
|