1466 lines
58 KiB
Python
1466 lines
58 KiB
Python
"""
|
|
Placeholder Resolver for AI Prompts
|
|
|
|
Provides a registry of placeholder functions that resolve to actual user data.
|
|
Used for prompt templates and preview functionality.
|
|
|
|
Phase 0c: Refactored to use data_layer for structured data.
|
|
This module now focuses on FORMATTING for AI consumption.
|
|
"""
|
|
import re
|
|
from datetime import datetime, timedelta
|
|
from typing import Dict, List, Optional, Callable
|
|
from db import get_db, get_cursor, r2d
|
|
|
|
# Phase 0c: Import data layer
|
|
from data_layer.body_metrics import (
|
|
get_latest_weight_data,
|
|
get_weight_trend_data,
|
|
get_body_composition_data,
|
|
get_circumference_summary_data
|
|
)
|
|
from data_layer.nutrition_metrics import (
|
|
get_nutrition_average_data,
|
|
get_nutrition_days_data,
|
|
get_protein_targets_data
|
|
)
|
|
from data_layer.activity_metrics import (
|
|
get_activity_summary_data,
|
|
get_activity_detail_data,
|
|
get_training_type_distribution_data
|
|
)
|
|
from data_layer.recovery_metrics import (
|
|
get_sleep_duration_data,
|
|
get_sleep_quality_data,
|
|
get_rest_days_data
|
|
)
|
|
from data_layer.health_metrics import (
|
|
get_resting_heart_rate_data,
|
|
get_heart_rate_variability_data,
|
|
get_vo2_max_data
|
|
)
|
|
|
|
|
|
# ── Helper Functions ──────────────────────────────────────────────────────────
|
|
|
|
def get_profile_data(profile_id: str) -> Dict:
|
|
"""Load profile data for a user."""
|
|
with get_db() as conn:
|
|
cur = get_cursor(conn)
|
|
cur.execute("SELECT * FROM profiles WHERE id=%s", (profile_id,))
|
|
return r2d(cur.fetchone()) if cur.rowcount > 0 else {}
|
|
|
|
|
|
def get_latest_weight(profile_id: str) -> Optional[str]:
|
|
"""
|
|
Get latest weight entry.
|
|
|
|
Phase 0c: Refactored to use data_layer.body_metrics.get_latest_weight_data()
|
|
This function now only FORMATS the data for AI consumption.
|
|
"""
|
|
data = get_latest_weight_data(profile_id)
|
|
|
|
if data['confidence'] == 'insufficient':
|
|
return "nicht verfügbar"
|
|
|
|
return f"{data['weight']:.1f} kg"
|
|
|
|
|
|
def get_weight_trend(profile_id: str, days: int = 28) -> str:
|
|
"""
|
|
Calculate weight trend description.
|
|
|
|
Phase 0c: Refactored to use data_layer.body_metrics.get_weight_trend_data()
|
|
This function now only FORMATS the data for AI consumption.
|
|
"""
|
|
data = get_weight_trend_data(profile_id, days)
|
|
|
|
if data['confidence'] == 'insufficient':
|
|
return "nicht genug Daten"
|
|
|
|
direction = data['direction']
|
|
delta = data['delta']
|
|
|
|
if direction == "stable":
|
|
return "stabil"
|
|
elif direction == "increasing":
|
|
return f"steigend (+{delta:.1f} kg in {days} Tagen)"
|
|
else: # decreasing
|
|
return f"sinkend ({delta:.1f} kg in {days} Tagen)"
|
|
|
|
|
|
def get_latest_bf(profile_id: str) -> Optional[str]:
|
|
"""
|
|
Get latest body fat percentage from caliper.
|
|
|
|
Phase 0c: Refactored to use data_layer.body_metrics.get_body_composition_data()
|
|
This function now only FORMATS the data for AI consumption.
|
|
"""
|
|
data = get_body_composition_data(profile_id)
|
|
|
|
if data['confidence'] == 'insufficient':
|
|
return "nicht verfügbar"
|
|
|
|
return f"{data['body_fat_pct']:.1f}%"
|
|
|
|
|
|
def get_nutrition_avg(profile_id: str, field: str, days: int = 30) -> str:
|
|
"""
|
|
Calculate average nutrition value.
|
|
|
|
Phase 0c: Refactored to use data_layer.nutrition_metrics.get_nutrition_average_data()
|
|
This function now only FORMATS the data for AI consumption.
|
|
"""
|
|
data = get_nutrition_average_data(profile_id, days)
|
|
|
|
if data['confidence'] == 'insufficient':
|
|
return "nicht verfügbar"
|
|
|
|
# Map field names to data keys
|
|
field_map = {
|
|
'protein': 'protein_avg',
|
|
'fat': 'fat_avg',
|
|
'carb': 'carbs_avg',
|
|
'kcal': 'kcal_avg'
|
|
}
|
|
data_key = field_map.get(field, f'{field}_avg')
|
|
value = data.get(data_key, 0)
|
|
|
|
if field == 'kcal':
|
|
return f"{int(value)} kcal/Tag (Ø {days} Tage)"
|
|
else:
|
|
return f"{int(value)}g/Tag (Ø {days} Tage)"
|
|
|
|
|
|
def get_caliper_summary(profile_id: str) -> str:
|
|
"""
|
|
Get latest caliper measurements summary.
|
|
|
|
Phase 0c: Refactored to use data_layer.body_metrics.get_body_composition_data()
|
|
This function now only FORMATS the data for AI consumption.
|
|
"""
|
|
data = get_body_composition_data(profile_id)
|
|
|
|
if data['confidence'] == 'insufficient':
|
|
return "keine Caliper-Messungen"
|
|
|
|
method = data.get('method', 'unbekannt')
|
|
return f"{data['body_fat_pct']:.1f}% ({method} am {data['date']})"
|
|
|
|
|
|
def get_circ_summary(profile_id: str) -> str:
|
|
"""
|
|
Get latest circumference measurements summary with age annotations.
|
|
|
|
Phase 0c: Refactored to use data_layer.body_metrics.get_circumference_summary_data()
|
|
This function now only FORMATS the data for AI consumption.
|
|
"""
|
|
data = get_circumference_summary_data(profile_id)
|
|
|
|
if data['confidence'] == 'insufficient':
|
|
return "keine Umfangsmessungen"
|
|
|
|
parts = []
|
|
for measurement in data['measurements']:
|
|
age_days = measurement['age_days']
|
|
|
|
# Format age annotation
|
|
if age_days == 0:
|
|
age_str = "heute"
|
|
elif age_days == 1:
|
|
age_str = "gestern"
|
|
elif age_days <= 7:
|
|
age_str = f"vor {age_days} Tagen"
|
|
elif age_days <= 30:
|
|
weeks = age_days // 7
|
|
age_str = f"vor {weeks} Woche{'n' if weeks > 1 else ''}"
|
|
else:
|
|
months = age_days // 30
|
|
age_str = f"vor {months} Monat{'en' if months > 1 else ''}"
|
|
|
|
parts.append(f"{measurement['point']} {measurement['value']:.1f}cm ({age_str})")
|
|
|
|
return ', '.join(parts) if parts else "keine Umfangsmessungen"
|
|
|
|
|
|
def get_goal_weight(profile_id: str) -> str:
|
|
"""Get goal weight from profile."""
|
|
profile = get_profile_data(profile_id)
|
|
goal = profile.get('goal_weight')
|
|
return f"{goal:.1f}" if goal else "nicht gesetzt"
|
|
|
|
|
|
def get_goal_bf_pct(profile_id: str) -> str:
|
|
"""Get goal body fat percentage from profile."""
|
|
profile = get_profile_data(profile_id)
|
|
goal = profile.get('goal_bf_pct')
|
|
return f"{goal:.1f}" if goal else "nicht gesetzt"
|
|
|
|
|
|
def get_nutrition_days(profile_id: str, days: int = 30) -> str:
|
|
"""
|
|
Get number of days with nutrition data.
|
|
|
|
Phase 0c: Refactored to use data_layer.nutrition_metrics.get_nutrition_days_data()
|
|
This function now only FORMATS the data for AI consumption.
|
|
"""
|
|
data = get_nutrition_days_data(profile_id, days)
|
|
return str(data['days_with_data'])
|
|
|
|
|
|
def get_protein_ziel_low(profile_id: str) -> str:
|
|
"""
|
|
Calculate lower protein target based on current weight (1.6g/kg).
|
|
|
|
Phase 0c: Refactored to use data_layer.nutrition_metrics.get_protein_targets_data()
|
|
This function now only FORMATS the data for AI consumption.
|
|
"""
|
|
data = get_protein_targets_data(profile_id)
|
|
|
|
if data['confidence'] == 'insufficient':
|
|
return "nicht verfügbar"
|
|
|
|
return f"{int(data['protein_target_low'])}"
|
|
|
|
|
|
def get_protein_ziel_high(profile_id: str) -> str:
|
|
"""
|
|
Calculate upper protein target based on current weight (2.2g/kg).
|
|
|
|
Phase 0c: Refactored to use data_layer.nutrition_metrics.get_protein_targets_data()
|
|
This function now only FORMATS the data for AI consumption.
|
|
"""
|
|
data = get_protein_targets_data(profile_id)
|
|
|
|
if data['confidence'] == 'insufficient':
|
|
return "nicht verfügbar"
|
|
|
|
return f"{int(data['protein_target_high'])}"
|
|
|
|
|
|
def get_activity_summary(profile_id: str, days: int = 14) -> str:
|
|
"""
|
|
Get activity summary for recent period.
|
|
|
|
Phase 0c: Refactored to use data_layer.activity_metrics.get_activity_summary_data()
|
|
This function now only FORMATS the data for AI consumption.
|
|
"""
|
|
data = get_activity_summary_data(profile_id, days)
|
|
|
|
if data['confidence'] == 'insufficient':
|
|
return f"Keine Aktivitäten in den letzten {days} Tagen"
|
|
|
|
return f"{data['activity_count']} Einheiten in {days} Tagen (Ø {data['avg_duration_min']} min/Einheit, {data['total_kcal']} kcal gesamt)"
|
|
|
|
|
|
def calculate_age(dob) -> str:
|
|
"""Calculate age from date of birth (accepts date object or string)."""
|
|
if not dob:
|
|
return "unbekannt"
|
|
try:
|
|
# Handle both datetime.date objects and strings
|
|
if isinstance(dob, str):
|
|
birth = datetime.strptime(dob, '%Y-%m-%d').date()
|
|
else:
|
|
birth = dob # Already a date object from PostgreSQL
|
|
|
|
today = datetime.now().date()
|
|
age = today.year - birth.year - ((today.month, today.day) < (birth.month, birth.day))
|
|
return str(age)
|
|
except Exception as e:
|
|
return "unbekannt"
|
|
|
|
|
|
def get_activity_detail(profile_id: str, days: int = 14) -> str:
|
|
"""
|
|
Get detailed activity log for analysis.
|
|
|
|
Phase 0c: Refactored to use data_layer.activity_metrics.get_activity_detail_data()
|
|
This function now only FORMATS the data for AI consumption.
|
|
"""
|
|
data = get_activity_detail_data(profile_id, days)
|
|
|
|
if data['confidence'] == 'insufficient':
|
|
return f"Keine Aktivitäten in den letzten {days} Tagen"
|
|
|
|
# Format as readable list (max 20 entries to avoid token bloat)
|
|
lines = []
|
|
for activity in data['activities'][:20]:
|
|
hr_str = f" HF={activity['hr_avg']}" if activity['hr_avg'] else ""
|
|
lines.append(
|
|
f"{activity['date']}: {activity['activity_type']} "
|
|
f"({activity['duration_min']}min, {activity['kcal_active']}kcal{hr_str})"
|
|
)
|
|
|
|
return '\n'.join(lines)
|
|
|
|
|
|
def get_trainingstyp_verteilung(profile_id: str, days: int = 14) -> str:
|
|
"""
|
|
Get training type distribution.
|
|
|
|
Phase 0c: Refactored to use data_layer.activity_metrics.get_training_type_distribution_data()
|
|
This function now only FORMATS the data for AI consumption.
|
|
"""
|
|
data = get_training_type_distribution_data(profile_id, days)
|
|
|
|
if data['confidence'] == 'insufficient' or not data['distribution']:
|
|
return "Keine kategorisierten Trainings"
|
|
|
|
# Format top 3 categories with percentages
|
|
parts = [
|
|
f"{dist['category']}: {int(dist['percentage'])}%"
|
|
for dist in data['distribution'][:3]
|
|
]
|
|
return ", ".join(parts)
|
|
|
|
|
|
def get_sleep_avg_duration(profile_id: str, days: int = 7) -> str:
|
|
"""
|
|
Calculate average sleep duration in hours.
|
|
|
|
Phase 0c: Refactored to use data_layer.recovery_metrics.get_sleep_duration_data()
|
|
This function now only FORMATS the data for AI consumption.
|
|
"""
|
|
data = get_sleep_duration_data(profile_id, days)
|
|
|
|
if data['confidence'] == 'insufficient':
|
|
return "nicht verfügbar"
|
|
|
|
return f"{data['avg_duration_hours']:.1f}h"
|
|
|
|
|
|
def get_sleep_avg_quality(profile_id: str, days: int = 7) -> str:
|
|
"""
|
|
Calculate average sleep quality (Deep+REM %).
|
|
|
|
Phase 0c: Refactored to use data_layer.recovery_metrics.get_sleep_quality_data()
|
|
This function now only FORMATS the data for AI consumption.
|
|
"""
|
|
data = get_sleep_quality_data(profile_id, days)
|
|
|
|
if data['confidence'] == 'insufficient':
|
|
return "nicht verfügbar"
|
|
|
|
return f"{data['quality_score']:.0f}% (Deep+REM)"
|
|
|
|
|
|
def get_rest_days_count(profile_id: str, days: int = 30) -> str:
|
|
"""
|
|
Count rest days in the given period.
|
|
|
|
Phase 0c: Refactored to use data_layer.recovery_metrics.get_rest_days_data()
|
|
This function now only FORMATS the data for AI consumption.
|
|
"""
|
|
data = get_rest_days_data(profile_id, days)
|
|
return f"{data['total_rest_days']} Ruhetage"
|
|
|
|
|
|
def get_vitals_avg_hr(profile_id: str, days: int = 7) -> str:
|
|
"""
|
|
Calculate average resting heart rate.
|
|
|
|
Phase 0c: Refactored to use data_layer.health_metrics.get_resting_heart_rate_data()
|
|
This function now only FORMATS the data for AI consumption.
|
|
"""
|
|
data = get_resting_heart_rate_data(profile_id, days)
|
|
|
|
if data['confidence'] == 'insufficient':
|
|
return "nicht verfügbar"
|
|
|
|
return f"{data['avg_rhr']} bpm"
|
|
|
|
|
|
def get_vitals_avg_hrv(profile_id: str, days: int = 7) -> str:
|
|
"""
|
|
Calculate average heart rate variability.
|
|
|
|
Phase 0c: Refactored to use data_layer.health_metrics.get_heart_rate_variability_data()
|
|
This function now only FORMATS the data for AI consumption.
|
|
"""
|
|
data = get_heart_rate_variability_data(profile_id, days)
|
|
|
|
if data['confidence'] == 'insufficient':
|
|
return "nicht verfügbar"
|
|
|
|
return f"{data['avg_hrv']} ms"
|
|
|
|
|
|
def get_vitals_vo2_max(profile_id: str) -> str:
|
|
"""
|
|
Get latest VO2 Max value.
|
|
|
|
Phase 0c: Refactored to use data_layer.health_metrics.get_vo2_max_data()
|
|
This function now only FORMATS the data for AI consumption.
|
|
"""
|
|
data = get_vo2_max_data(profile_id)
|
|
|
|
if data['confidence'] == 'insufficient':
|
|
return "nicht verfügbar"
|
|
|
|
return f"{data['vo2_max']:.1f} ml/kg/min"
|
|
|
|
|
|
# ── Phase 0b Calculation Engine Integration ──────────────────────────────────
|
|
|
|
def _safe_int(func_name: str, profile_id: str) -> str:
|
|
"""
|
|
Safely call calculation function and return integer value or fallback.
|
|
|
|
Args:
|
|
func_name: Name of the calculation function (e.g., 'goal_progress_score')
|
|
profile_id: Profile ID
|
|
|
|
Returns:
|
|
String representation of integer value or 'nicht verfügbar'
|
|
"""
|
|
import traceback
|
|
try:
|
|
# Import calculations dynamically to avoid circular imports
|
|
from data_layer import body_metrics, nutrition_metrics, activity_metrics, recovery_metrics, scores
|
|
from data_layer import correlations as correlation_metrics
|
|
|
|
# Map function names to actual functions
|
|
func_map = {
|
|
'goal_progress_score': scores.calculate_goal_progress_score,
|
|
'body_progress_score': body_metrics.calculate_body_progress_score,
|
|
'nutrition_score': nutrition_metrics.calculate_nutrition_score,
|
|
'activity_score': activity_metrics.calculate_activity_score,
|
|
'recovery_score_v2': recovery_metrics.calculate_recovery_score_v2,
|
|
'data_quality_score': scores.calculate_data_quality_score,
|
|
'top_goal_progress_pct': lambda pid: scores.get_top_priority_goal(pid)['progress_pct'] if scores.get_top_priority_goal(pid) else None,
|
|
'top_focus_area_progress': lambda pid: scores.get_top_focus_area(pid)['progress'] if scores.get_top_focus_area(pid) else None,
|
|
'focus_cat_körper_progress': lambda pid: scores.calculate_category_progress(pid, 'körper'),
|
|
'focus_cat_ernährung_progress': lambda pid: scores.calculate_category_progress(pid, 'ernährung'),
|
|
'focus_cat_aktivität_progress': lambda pid: scores.calculate_category_progress(pid, 'aktivität'),
|
|
'focus_cat_recovery_progress': lambda pid: scores.calculate_category_progress(pid, 'recovery'),
|
|
'focus_cat_vitalwerte_progress': lambda pid: scores.calculate_category_progress(pid, 'vitalwerte'),
|
|
'focus_cat_mental_progress': lambda pid: scores.calculate_category_progress(pid, 'mental'),
|
|
'focus_cat_lebensstil_progress': lambda pid: scores.calculate_category_progress(pid, 'lebensstil'),
|
|
'training_minutes_week': activity_metrics.calculate_training_minutes_week,
|
|
'training_frequency_7d': activity_metrics.calculate_training_frequency_7d,
|
|
'quality_sessions_pct': activity_metrics.calculate_quality_sessions_pct,
|
|
'ability_balance_strength': activity_metrics.calculate_ability_balance_strength,
|
|
'ability_balance_endurance': activity_metrics.calculate_ability_balance_endurance,
|
|
'ability_balance_mental': activity_metrics.calculate_ability_balance_mental,
|
|
'ability_balance_coordination': activity_metrics.calculate_ability_balance_coordination,
|
|
'ability_balance_mobility': activity_metrics.calculate_ability_balance_mobility,
|
|
'proxy_internal_load_7d': activity_metrics.calculate_proxy_internal_load_7d,
|
|
'strain_score': activity_metrics.calculate_strain_score,
|
|
'rest_day_compliance': activity_metrics.calculate_rest_day_compliance,
|
|
'protein_adequacy_28d': nutrition_metrics.calculate_protein_adequacy_28d,
|
|
'macro_consistency_score': nutrition_metrics.calculate_macro_consistency_score,
|
|
'recent_load_balance_3d': recovery_metrics.calculate_recent_load_balance_3d,
|
|
'sleep_quality_7d': recovery_metrics.calculate_sleep_quality_7d,
|
|
}
|
|
|
|
func = func_map.get(func_name)
|
|
if not func:
|
|
return 'nicht verfügbar'
|
|
|
|
result = func(profile_id)
|
|
return str(int(result)) if result is not None else 'nicht verfügbar'
|
|
except Exception as e:
|
|
print(f"[ERROR] _safe_int({func_name}, {profile_id}): {type(e).__name__}: {e}")
|
|
traceback.print_exc()
|
|
return 'nicht verfügbar'
|
|
|
|
|
|
def _safe_float(func_name: str, profile_id: str, decimals: int = 1) -> str:
|
|
"""
|
|
Safely call calculation function and return float value or fallback.
|
|
|
|
Args:
|
|
func_name: Name of the calculation function
|
|
profile_id: Profile ID
|
|
decimals: Number of decimal places
|
|
|
|
Returns:
|
|
String representation of float value or 'nicht verfügbar'
|
|
"""
|
|
import traceback
|
|
try:
|
|
from data_layer import body_metrics, nutrition_metrics, activity_metrics, recovery_metrics, scores
|
|
|
|
func_map = {
|
|
'weight_7d_median': body_metrics.calculate_weight_7d_median,
|
|
'weight_28d_slope': body_metrics.calculate_weight_28d_slope,
|
|
'weight_90d_slope': body_metrics.calculate_weight_90d_slope,
|
|
'fm_28d_change': body_metrics.calculate_fm_28d_change,
|
|
'lbm_28d_change': body_metrics.calculate_lbm_28d_change,
|
|
'waist_28d_delta': body_metrics.calculate_waist_28d_delta,
|
|
'hip_28d_delta': body_metrics.calculate_hip_28d_delta,
|
|
'chest_28d_delta': body_metrics.calculate_chest_28d_delta,
|
|
'arm_28d_delta': body_metrics.calculate_arm_28d_delta,
|
|
'thigh_28d_delta': body_metrics.calculate_thigh_28d_delta,
|
|
'waist_hip_ratio': body_metrics.calculate_waist_hip_ratio,
|
|
'energy_balance_7d': nutrition_metrics.calculate_energy_balance_7d,
|
|
'protein_g_per_kg': nutrition_metrics.calculate_protein_g_per_kg,
|
|
'monotony_score': activity_metrics.calculate_monotony_score,
|
|
'vo2max_trend_28d': activity_metrics.calculate_vo2max_trend_28d,
|
|
'hrv_vs_baseline_pct': recovery_metrics.calculate_hrv_vs_baseline_pct,
|
|
'rhr_vs_baseline_pct': recovery_metrics.calculate_rhr_vs_baseline_pct,
|
|
'sleep_avg_duration_7d': recovery_metrics.calculate_sleep_avg_duration_7d,
|
|
'sleep_debt_hours': recovery_metrics.calculate_sleep_debt_hours,
|
|
'sleep_regularity_proxy': recovery_metrics.calculate_sleep_regularity_proxy,
|
|
'focus_cat_körper_weight': lambda pid: scores.calculate_category_weight(pid, 'körper'),
|
|
'focus_cat_ernährung_weight': lambda pid: scores.calculate_category_weight(pid, 'ernährung'),
|
|
'focus_cat_aktivität_weight': lambda pid: scores.calculate_category_weight(pid, 'aktivität'),
|
|
'focus_cat_recovery_weight': lambda pid: scores.calculate_category_weight(pid, 'recovery'),
|
|
'focus_cat_vitalwerte_weight': lambda pid: scores.calculate_category_weight(pid, 'vitalwerte'),
|
|
'focus_cat_mental_weight': lambda pid: scores.calculate_category_weight(pid, 'mental'),
|
|
'focus_cat_lebensstil_weight': lambda pid: scores.calculate_category_weight(pid, 'lebensstil'),
|
|
}
|
|
|
|
func = func_map.get(func_name)
|
|
if not func:
|
|
return 'nicht verfügbar'
|
|
|
|
result = func(profile_id)
|
|
return f"{result:.{decimals}f}" if result is not None else 'nicht verfügbar'
|
|
except Exception as e:
|
|
print(f"[ERROR] _safe_float({func_name}, {profile_id}): {type(e).__name__}: {e}")
|
|
traceback.print_exc()
|
|
return 'nicht verfügbar'
|
|
|
|
|
|
def _safe_str(func_name: str, profile_id: str) -> str:
|
|
"""
|
|
Safely call calculation function and return string value or fallback.
|
|
"""
|
|
import traceback
|
|
try:
|
|
from data_layer import body_metrics, nutrition_metrics, activity_metrics, scores
|
|
from data_layer import correlations as correlation_metrics
|
|
|
|
func_map = {
|
|
'top_goal_name': lambda pid: (scores.get_top_priority_goal(pid).get('name') or scores.get_top_priority_goal(pid).get('goal_type')) if scores.get_top_priority_goal(pid) else None,
|
|
'top_goal_status': lambda pid: scores.get_top_priority_goal(pid)['status'] if scores.get_top_priority_goal(pid) else None,
|
|
'top_focus_area_name': lambda pid: scores.get_top_focus_area(pid)['label'] if scores.get_top_focus_area(pid) else None,
|
|
'recomposition_quadrant': body_metrics.calculate_recomposition_quadrant,
|
|
'energy_deficit_surplus': nutrition_metrics.calculate_energy_deficit_surplus,
|
|
'protein_days_in_target': nutrition_metrics.calculate_protein_days_in_target,
|
|
'intake_volatility': nutrition_metrics.calculate_intake_volatility,
|
|
'active_goals_md': lambda pid: _format_goals_as_markdown(pid),
|
|
'focus_areas_weighted_md': lambda pid: _format_focus_areas_as_markdown(pid),
|
|
'top_3_focus_areas': lambda pid: _format_top_focus_areas(pid),
|
|
'top_3_goals_behind_schedule': lambda pid: _format_goals_behind(pid),
|
|
'top_3_goals_on_track': lambda pid: _format_goals_on_track(pid),
|
|
}
|
|
|
|
func = func_map.get(func_name)
|
|
if not func:
|
|
return 'nicht verfügbar'
|
|
|
|
result = func(profile_id)
|
|
return str(result) if result is not None else 'nicht verfügbar'
|
|
except Exception as e:
|
|
print(f"[ERROR] _safe_str({func_name}, {profile_id}): {type(e).__name__}: {e}")
|
|
traceback.print_exc()
|
|
return 'nicht verfügbar'
|
|
|
|
|
|
def _safe_json(func_name: str, profile_id: str) -> str:
|
|
"""
|
|
Safely call calculation function and return JSON string or fallback.
|
|
"""
|
|
import traceback
|
|
try:
|
|
import json
|
|
from data_layer import scores
|
|
from data_layer import correlations as correlation_metrics
|
|
|
|
func_map = {
|
|
'correlation_energy_weight_lag': lambda pid: correlation_metrics.calculate_lag_correlation(pid, 'energy', 'weight'),
|
|
'correlation_protein_lbm': lambda pid: correlation_metrics.calculate_lag_correlation(pid, 'protein', 'lbm'),
|
|
'correlation_load_hrv': lambda pid: correlation_metrics.calculate_lag_correlation(pid, 'training_load', 'hrv'),
|
|
'correlation_load_rhr': lambda pid: correlation_metrics.calculate_lag_correlation(pid, 'training_load', 'rhr'),
|
|
'correlation_sleep_recovery': correlation_metrics.calculate_correlation_sleep_recovery,
|
|
'plateau_detected': correlation_metrics.calculate_plateau_detected,
|
|
'top_drivers': correlation_metrics.calculate_top_drivers,
|
|
'active_goals_json': lambda pid: _get_active_goals_json(pid),
|
|
'focus_areas_weighted_json': lambda pid: _get_focus_areas_weighted_json(pid),
|
|
'focus_area_weights_json': lambda pid: json.dumps(scores.get_user_focus_weights(pid), ensure_ascii=False),
|
|
}
|
|
|
|
func = func_map.get(func_name)
|
|
if not func:
|
|
return '{}'
|
|
|
|
result = func(profile_id)
|
|
if result is None:
|
|
return '{}'
|
|
|
|
# If already string, return it; otherwise convert to JSON
|
|
if isinstance(result, str):
|
|
return result
|
|
else:
|
|
return json.dumps(result, ensure_ascii=False)
|
|
except Exception as e:
|
|
print(f"[ERROR] _safe_json({func_name}, {profile_id}): {type(e).__name__}: {e}")
|
|
traceback.print_exc()
|
|
return '{}'
|
|
|
|
|
|
def _get_active_goals_json(profile_id: str) -> str:
|
|
"""Get active goals as JSON string"""
|
|
import json
|
|
try:
|
|
from goal_utils import get_active_goals
|
|
goals = get_active_goals(profile_id)
|
|
return json.dumps(goals, default=str)
|
|
except Exception:
|
|
return '[]'
|
|
|
|
|
|
def _get_focus_areas_weighted_json(profile_id: str) -> str:
|
|
"""Get focus areas with weights as JSON string"""
|
|
import json
|
|
try:
|
|
from calculations.scores import get_user_focus_weights
|
|
from goal_utils import get_db, get_cursor
|
|
|
|
weights = get_user_focus_weights(profile_id)
|
|
|
|
# Get focus area details
|
|
with get_db() as conn:
|
|
cur = get_cursor(conn)
|
|
cur.execute("""
|
|
SELECT key, name_de, name_en, category
|
|
FROM focus_area_definitions
|
|
WHERE is_active = true
|
|
""")
|
|
definitions = {row['key']: row for row in cur.fetchall()}
|
|
|
|
# Build weighted list
|
|
result = []
|
|
for area_key, weight in weights.items():
|
|
if weight > 0 and area_key in definitions:
|
|
area = definitions[area_key]
|
|
result.append({
|
|
'key': area_key,
|
|
'name': area['name_de'],
|
|
'category': area['category'],
|
|
'weight': weight
|
|
})
|
|
|
|
# Sort by weight descending
|
|
result.sort(key=lambda x: x['weight'], reverse=True)
|
|
|
|
return json.dumps(result, default=str)
|
|
except Exception:
|
|
return '[]'
|
|
|
|
|
|
def _format_goals_as_markdown(profile_id: str) -> str:
|
|
"""Format goals as markdown table"""
|
|
try:
|
|
from goal_utils import get_active_goals
|
|
|
|
goals = get_active_goals(profile_id)
|
|
if not goals:
|
|
return 'Keine Ziele definiert'
|
|
|
|
# Build markdown table
|
|
lines = ['| Ziel | Aktuell | Ziel | Progress | Status |', '|------|---------|------|----------|--------|']
|
|
|
|
for goal in goals:
|
|
name = goal.get('name') or goal.get('goal_type', 'Unbekannt')
|
|
current = goal.get('current_value')
|
|
target = goal.get('target_value')
|
|
start = goal.get('start_value')
|
|
|
|
# Calculate progress if possible
|
|
progress_str = '-'
|
|
if None not in [current, target, start]:
|
|
try:
|
|
current_f = float(current)
|
|
target_f = float(target)
|
|
start_f = float(start)
|
|
|
|
if target_f == start_f:
|
|
progress_pct = 100 if current_f == target_f else 0
|
|
else:
|
|
progress_pct = ((current_f - start_f) / (target_f - start_f)) * 100
|
|
progress_pct = max(0, min(100, progress_pct))
|
|
|
|
progress_str = f"{int(progress_pct)}%"
|
|
except (ValueError, ZeroDivisionError):
|
|
progress_str = '-'
|
|
|
|
current_str = f"{current}" if current is not None else '-'
|
|
target_str = f"{target}" if target is not None else '-'
|
|
status = '🎯' if goal.get('is_primary') else '○'
|
|
|
|
lines.append(f"| {name} | {current_str} | {target_str} | {progress_str} | {status} |")
|
|
|
|
return '\n'.join(lines)
|
|
except Exception:
|
|
return 'Keine Ziele definiert'
|
|
|
|
|
|
def _format_focus_areas_as_markdown(profile_id: str) -> str:
|
|
"""Format focus areas as markdown"""
|
|
try:
|
|
import json
|
|
weighted_json = _get_focus_areas_weighted_json(profile_id)
|
|
areas = json.loads(weighted_json)
|
|
|
|
if not areas:
|
|
return 'Keine Focus Areas aktiv'
|
|
|
|
# Build markdown list
|
|
lines = []
|
|
for area in areas:
|
|
name = area.get('name', 'Unbekannt')
|
|
weight = area.get('weight', 0)
|
|
lines.append(f"- **{name}**: {weight}%")
|
|
|
|
return '\n'.join(lines)
|
|
except Exception:
|
|
return 'Keine Focus Areas aktiv'
|
|
|
|
|
|
def _format_top_focus_areas(profile_id: str, n: int = 3) -> str:
|
|
"""Format top N focus areas as text"""
|
|
try:
|
|
import json
|
|
weighted_json = _get_focus_areas_weighted_json(profile_id)
|
|
areas = json.loads(weighted_json)
|
|
|
|
if not areas:
|
|
return 'Keine Focus Areas definiert'
|
|
|
|
# Sort by weight descending and take top N
|
|
sorted_areas = sorted(areas, key=lambda x: x.get('weight', 0), reverse=True)[:n]
|
|
|
|
lines = []
|
|
for i, area in enumerate(sorted_areas, 1):
|
|
name = area.get('name', 'Unbekannt')
|
|
weight = area.get('weight', 0)
|
|
lines.append(f"{i}. {name} ({weight}%)")
|
|
|
|
return ', '.join(lines)
|
|
except Exception:
|
|
return 'nicht verfügbar'
|
|
|
|
|
|
def _format_goals_behind(profile_id: str, n: int = 3) -> str:
|
|
"""
|
|
Format top N goals behind schedule (based on time deviation).
|
|
|
|
Compares actual progress vs. expected progress based on elapsed time.
|
|
Negative deviation = behind schedule.
|
|
"""
|
|
try:
|
|
from goal_utils import get_active_goals
|
|
from datetime import date
|
|
|
|
goals = get_active_goals(profile_id)
|
|
|
|
if not goals:
|
|
return 'Keine Ziele definiert'
|
|
|
|
today = date.today()
|
|
goals_with_deviation = []
|
|
|
|
|
|
for g in goals:
|
|
goal_name = g.get('name') or g.get('goal_type', 'Unknown')
|
|
current = g.get('current_value')
|
|
target = g.get('target_value')
|
|
start = g.get('start_value')
|
|
start_date = g.get('start_date')
|
|
target_date = g.get('target_date')
|
|
|
|
|
|
# Skip if missing required values
|
|
if None in [current, target, start]:
|
|
continue
|
|
|
|
# Skip if no target_date (can't calculate time-based progress)
|
|
if not target_date:
|
|
continue
|
|
|
|
try:
|
|
current = float(current)
|
|
target = float(target)
|
|
start = float(start)
|
|
|
|
# Calculate actual progress percentage
|
|
if target == start:
|
|
actual_progress_pct = 100 if current == target else 0
|
|
else:
|
|
actual_progress_pct = ((current - start) / (target - start)) * 100
|
|
actual_progress_pct = max(0, min(100, actual_progress_pct))
|
|
|
|
# Calculate expected progress based on time
|
|
if start_date:
|
|
# Use start_date if available
|
|
start_dt = start_date if isinstance(start_date, date) else date.fromisoformat(str(start_date))
|
|
else:
|
|
# Fallback: assume start date = created_at date
|
|
created_at = g.get('created_at')
|
|
if created_at:
|
|
start_dt = date.fromisoformat(str(created_at).split('T')[0])
|
|
else:
|
|
continue # Can't calculate without start date
|
|
|
|
target_dt = target_date if isinstance(target_date, date) else date.fromisoformat(str(target_date))
|
|
|
|
# Calculate time progress
|
|
total_days = (target_dt - start_dt).days
|
|
elapsed_days = (today - start_dt).days
|
|
|
|
if total_days <= 0:
|
|
continue # Invalid date range
|
|
|
|
expected_progress_pct = (elapsed_days / total_days) * 100
|
|
expected_progress_pct = max(0, min(100, expected_progress_pct))
|
|
|
|
# Calculate deviation (negative = behind schedule)
|
|
deviation = actual_progress_pct - expected_progress_pct
|
|
|
|
g['_actual_progress'] = int(actual_progress_pct)
|
|
g['_expected_progress'] = int(expected_progress_pct)
|
|
g['_deviation'] = int(deviation)
|
|
goals_with_deviation.append(g)
|
|
|
|
except (ValueError, ZeroDivisionError, TypeError) as e:
|
|
continue
|
|
|
|
# Also process goals WITHOUT target_date (simple progress)
|
|
goals_without_date = []
|
|
|
|
for g in goals:
|
|
if g.get('target_date'):
|
|
continue # Already processed above
|
|
|
|
goal_name = g.get('name') or g.get('goal_type', 'Unknown')
|
|
current = g.get('current_value')
|
|
target = g.get('target_value')
|
|
start = g.get('start_value')
|
|
|
|
if None in [current, target, start]:
|
|
continue
|
|
|
|
try:
|
|
current = float(current)
|
|
target = float(target)
|
|
start = float(start)
|
|
|
|
if target == start:
|
|
progress_pct = 100 if current == target else 0
|
|
else:
|
|
progress_pct = ((current - start) / (target - start)) * 100
|
|
progress_pct = max(0, min(100, progress_pct))
|
|
|
|
g['_simple_progress'] = int(progress_pct)
|
|
goals_without_date.append(g)
|
|
except (ValueError, ZeroDivisionError, TypeError) as e:
|
|
continue
|
|
|
|
# Combine: Goals with negative deviation + Goals without date with low progress
|
|
behind_with_date = [g for g in goals_with_deviation if g['_deviation'] < 0]
|
|
behind_without_date = [g for g in goals_without_date if g['_simple_progress'] < 50]
|
|
|
|
|
|
# Create combined list with sort keys
|
|
combined = []
|
|
for g in behind_with_date:
|
|
combined.append({
|
|
'goal': g,
|
|
'sort_key': g['_deviation'], # Negative deviation (worst first)
|
|
'has_date': True
|
|
})
|
|
for g in behind_without_date:
|
|
# Map progress to deviation-like scale: 0% = -100, 50% = -50
|
|
combined.append({
|
|
'goal': g,
|
|
'sort_key': g['_simple_progress'] - 100, # Convert to negative scale
|
|
'has_date': False
|
|
})
|
|
|
|
if not combined:
|
|
return 'Alle Ziele im Zeitplan oder erreicht'
|
|
|
|
# Sort by sort_key (most negative first)
|
|
sorted_combined = sorted(combined, key=lambda x: x['sort_key'])[:n]
|
|
|
|
lines = []
|
|
for item in sorted_combined:
|
|
g = item['goal']
|
|
name = g.get('name') or g.get('goal_type', 'Unbekannt')
|
|
|
|
if item['has_date']:
|
|
actual = g['_actual_progress']
|
|
expected = g['_expected_progress']
|
|
deviation = g['_deviation']
|
|
lines.append(f"{name} ({actual}% statt {expected}%, {deviation}%)")
|
|
else:
|
|
progress = g['_simple_progress']
|
|
lines.append(f"{name} ({progress}% erreicht)")
|
|
|
|
return ', '.join(lines)
|
|
except Exception as e:
|
|
print(f"[ERROR] _format_goals_behind: {e}")
|
|
import traceback
|
|
traceback.print_exc()
|
|
return 'nicht verfügbar'
|
|
|
|
|
|
def _format_goals_on_track(profile_id: str, n: int = 3) -> str:
|
|
"""
|
|
Format top N goals ahead of schedule (based on time deviation).
|
|
|
|
Compares actual progress vs. expected progress based on elapsed time.
|
|
Positive deviation = ahead of schedule / on track.
|
|
"""
|
|
try:
|
|
from goal_utils import get_active_goals
|
|
from datetime import date
|
|
|
|
goals = get_active_goals(profile_id)
|
|
|
|
if not goals:
|
|
return 'Keine Ziele definiert'
|
|
|
|
today = date.today()
|
|
goals_with_deviation = []
|
|
|
|
|
|
for g in goals:
|
|
goal_name = g.get('name') or g.get('goal_type', 'Unknown')
|
|
current = g.get('current_value')
|
|
target = g.get('target_value')
|
|
start = g.get('start_value')
|
|
start_date = g.get('start_date')
|
|
target_date = g.get('target_date')
|
|
|
|
|
|
# Skip if missing required values
|
|
if None in [current, target, start]:
|
|
continue
|
|
|
|
# Skip if no target_date
|
|
if not target_date:
|
|
continue
|
|
|
|
try:
|
|
current = float(current)
|
|
target = float(target)
|
|
start = float(start)
|
|
|
|
# Calculate actual progress percentage
|
|
if target == start:
|
|
actual_progress_pct = 100 if current == target else 0
|
|
else:
|
|
actual_progress_pct = ((current - start) / (target - start)) * 100
|
|
actual_progress_pct = max(0, min(100, actual_progress_pct))
|
|
|
|
# Calculate expected progress based on time
|
|
if start_date:
|
|
start_dt = start_date if isinstance(start_date, date) else date.fromisoformat(str(start_date))
|
|
else:
|
|
created_at = g.get('created_at')
|
|
if created_at:
|
|
start_dt = date.fromisoformat(str(created_at).split('T')[0])
|
|
else:
|
|
continue
|
|
|
|
target_dt = target_date if isinstance(target_date, date) else date.fromisoformat(str(target_date))
|
|
|
|
# Calculate time progress
|
|
total_days = (target_dt - start_dt).days
|
|
elapsed_days = (today - start_dt).days
|
|
|
|
if total_days <= 0:
|
|
continue
|
|
|
|
expected_progress_pct = (elapsed_days / total_days) * 100
|
|
expected_progress_pct = max(0, min(100, expected_progress_pct))
|
|
|
|
# Calculate deviation (positive = ahead of schedule)
|
|
deviation = actual_progress_pct - expected_progress_pct
|
|
|
|
g['_actual_progress'] = int(actual_progress_pct)
|
|
g['_expected_progress'] = int(expected_progress_pct)
|
|
g['_deviation'] = int(deviation)
|
|
goals_with_deviation.append(g)
|
|
|
|
except (ValueError, ZeroDivisionError, TypeError) as e:
|
|
continue
|
|
|
|
# Also process goals WITHOUT target_date (simple progress)
|
|
goals_without_date = []
|
|
|
|
for g in goals:
|
|
if g.get('target_date'):
|
|
continue # Already processed above
|
|
|
|
goal_name = g.get('name') or g.get('goal_type', 'Unknown')
|
|
current = g.get('current_value')
|
|
target = g.get('target_value')
|
|
start = g.get('start_value')
|
|
|
|
if None in [current, target, start]:
|
|
continue
|
|
|
|
try:
|
|
current = float(current)
|
|
target = float(target)
|
|
start = float(start)
|
|
|
|
if target == start:
|
|
progress_pct = 100 if current == target else 0
|
|
else:
|
|
progress_pct = ((current - start) / (target - start)) * 100
|
|
progress_pct = max(0, min(100, progress_pct))
|
|
|
|
g['_simple_progress'] = int(progress_pct)
|
|
goals_without_date.append(g)
|
|
except (ValueError, ZeroDivisionError, TypeError) as e:
|
|
continue
|
|
|
|
# Combine: Goals with positive deviation + Goals without date with high progress
|
|
ahead_with_date = [g for g in goals_with_deviation if g['_deviation'] >= 0]
|
|
ahead_without_date = [g for g in goals_without_date if g['_simple_progress'] >= 50]
|
|
|
|
|
|
# Create combined list with sort keys
|
|
combined = []
|
|
for g in ahead_with_date:
|
|
combined.append({
|
|
'goal': g,
|
|
'sort_key': g['_deviation'], # Positive deviation (best first)
|
|
'has_date': True
|
|
})
|
|
for g in ahead_without_date:
|
|
# Map progress to deviation-like scale: 50% = 0, 100% = +50
|
|
combined.append({
|
|
'goal': g,
|
|
'sort_key': g['_simple_progress'] - 50, # Convert to positive scale
|
|
'has_date': False
|
|
})
|
|
|
|
if not combined:
|
|
return 'Keine Ziele im Zeitplan'
|
|
|
|
# Sort by sort_key descending (most positive first)
|
|
sorted_combined = sorted(combined, key=lambda x: x['sort_key'], reverse=True)[:n]
|
|
|
|
lines = []
|
|
for item in sorted_combined:
|
|
g = item['goal']
|
|
name = g.get('name') or g.get('goal_type', 'Unbekannt')
|
|
|
|
if item['has_date']:
|
|
actual = g['_actual_progress']
|
|
deviation = g['_deviation']
|
|
lines.append(f"{name} ({actual}%, +{deviation}% voraus)")
|
|
else:
|
|
progress = g['_simple_progress']
|
|
lines.append(f"{name} ({progress}% erreicht)")
|
|
|
|
return ', '.join(lines)
|
|
except Exception as e:
|
|
print(f"[ERROR] _format_goals_on_track: {e}")
|
|
import traceback
|
|
traceback.print_exc()
|
|
return 'nicht verfügbar'
|
|
|
|
|
|
# ── Placeholder Registry ──────────────────────────────────────────────────────
|
|
|
|
PLACEHOLDER_MAP: Dict[str, Callable[[str], str]] = {
|
|
# Profil
|
|
'{{name}}': lambda pid: get_profile_data(pid).get('name', 'Nutzer'),
|
|
'{{age}}': lambda pid: calculate_age(get_profile_data(pid).get('dob')),
|
|
'{{height}}': lambda pid: str(get_profile_data(pid).get('height', 'unbekannt')),
|
|
'{{geschlecht}}': lambda pid: 'männlich' if get_profile_data(pid).get('sex') == 'm' else 'weiblich',
|
|
|
|
# Körper
|
|
'{{weight_aktuell}}': get_latest_weight,
|
|
'{{weight_trend}}': get_weight_trend,
|
|
'{{kf_aktuell}}': get_latest_bf,
|
|
'{{bmi}}': lambda pid: calculate_bmi(pid),
|
|
'{{caliper_summary}}': get_caliper_summary,
|
|
'{{circ_summary}}': get_circ_summary,
|
|
'{{goal_weight}}': get_goal_weight,
|
|
'{{goal_bf_pct}}': get_goal_bf_pct,
|
|
|
|
# Ernährung
|
|
'{{kcal_avg}}': lambda pid: get_nutrition_avg(pid, 'kcal', 30),
|
|
'{{protein_avg}}': lambda pid: get_nutrition_avg(pid, 'protein', 30),
|
|
'{{carb_avg}}': lambda pid: get_nutrition_avg(pid, 'carb', 30),
|
|
'{{fat_avg}}': lambda pid: get_nutrition_avg(pid, 'fat', 30),
|
|
'{{nutrition_days}}': lambda pid: get_nutrition_days(pid, 30),
|
|
'{{protein_ziel_low}}': get_protein_ziel_low,
|
|
'{{protein_ziel_high}}': get_protein_ziel_high,
|
|
|
|
# Training
|
|
'{{activity_summary}}': get_activity_summary,
|
|
'{{activity_detail}}': get_activity_detail,
|
|
'{{trainingstyp_verteilung}}': get_trainingstyp_verteilung,
|
|
|
|
# Schlaf & Erholung
|
|
'{{sleep_avg_duration}}': lambda pid: get_sleep_avg_duration(pid, 7),
|
|
'{{sleep_avg_quality}}': lambda pid: get_sleep_avg_quality(pid, 7),
|
|
'{{rest_days_count}}': lambda pid: get_rest_days_count(pid, 30),
|
|
|
|
# Vitalwerte
|
|
'{{vitals_avg_hr}}': lambda pid: get_vitals_avg_hr(pid, 7),
|
|
'{{vitals_avg_hrv}}': lambda pid: get_vitals_avg_hrv(pid, 7),
|
|
'{{vitals_vo2_max}}': get_vitals_vo2_max,
|
|
|
|
# Zeitraum
|
|
'{{datum_heute}}': lambda pid: datetime.now().strftime('%d.%m.%Y'),
|
|
'{{zeitraum_7d}}': lambda pid: 'letzte 7 Tage',
|
|
'{{zeitraum_30d}}': lambda pid: 'letzte 30 Tage',
|
|
'{{zeitraum_90d}}': lambda pid: 'letzte 90 Tage',
|
|
|
|
# ========================================================================
|
|
# PHASE 0b: Goal-Aware Placeholders (Dynamic Focus Areas v2.0)
|
|
# ========================================================================
|
|
|
|
# --- Meta Scores (Ebene 1: Aggregierte Scores) ---
|
|
'{{goal_progress_score}}': lambda pid: _safe_int('goal_progress_score', pid),
|
|
'{{body_progress_score}}': lambda pid: _safe_int('body_progress_score', pid),
|
|
'{{nutrition_score}}': lambda pid: _safe_int('nutrition_score', pid),
|
|
'{{activity_score}}': lambda pid: _safe_int('activity_score', pid),
|
|
'{{recovery_score}}': lambda pid: _safe_int('recovery_score_v2', pid),
|
|
'{{data_quality_score}}': lambda pid: _safe_int('data_quality_score', pid),
|
|
|
|
# --- Top-Weighted Goals/Focus Areas (Ebene 2: statt Primary) ---
|
|
'{{top_goal_name}}': lambda pid: _safe_str('top_goal_name', pid),
|
|
'{{top_goal_progress_pct}}': lambda pid: _safe_str('top_goal_progress_pct', pid),
|
|
'{{top_goal_status}}': lambda pid: _safe_str('top_goal_status', pid),
|
|
'{{top_focus_area_name}}': lambda pid: _safe_str('top_focus_area_name', pid),
|
|
'{{top_focus_area_progress}}': lambda pid: _safe_int('top_focus_area_progress', pid),
|
|
|
|
# --- Category Scores (Ebene 3: 7 Kategorien) ---
|
|
'{{focus_cat_körper_progress}}': lambda pid: _safe_int('focus_cat_körper_progress', pid),
|
|
'{{focus_cat_körper_weight}}': lambda pid: _safe_float('focus_cat_körper_weight', pid),
|
|
'{{focus_cat_ernährung_progress}}': lambda pid: _safe_int('focus_cat_ernährung_progress', pid),
|
|
'{{focus_cat_ernährung_weight}}': lambda pid: _safe_float('focus_cat_ernährung_weight', pid),
|
|
'{{focus_cat_aktivität_progress}}': lambda pid: _safe_int('focus_cat_aktivität_progress', pid),
|
|
'{{focus_cat_aktivität_weight}}': lambda pid: _safe_float('focus_cat_aktivität_weight', pid),
|
|
'{{focus_cat_recovery_progress}}': lambda pid: _safe_int('focus_cat_recovery_progress', pid),
|
|
'{{focus_cat_recovery_weight}}': lambda pid: _safe_float('focus_cat_recovery_weight', pid),
|
|
'{{focus_cat_vitalwerte_progress}}': lambda pid: _safe_int('focus_cat_vitalwerte_progress', pid),
|
|
'{{focus_cat_vitalwerte_weight}}': lambda pid: _safe_float('focus_cat_vitalwerte_weight', pid),
|
|
'{{focus_cat_mental_progress}}': lambda pid: _safe_int('focus_cat_mental_progress', pid),
|
|
'{{focus_cat_mental_weight}}': lambda pid: _safe_float('focus_cat_mental_weight', pid),
|
|
'{{focus_cat_lebensstil_progress}}': lambda pid: _safe_int('focus_cat_lebensstil_progress', pid),
|
|
'{{focus_cat_lebensstil_weight}}': lambda pid: _safe_float('focus_cat_lebensstil_weight', pid),
|
|
|
|
# --- Body Metrics (Ebene 4: Einzelmetriken K1-K5) ---
|
|
'{{weight_7d_median}}': lambda pid: _safe_float('weight_7d_median', pid),
|
|
'{{weight_28d_slope}}': lambda pid: _safe_float('weight_28d_slope', pid, decimals=4),
|
|
'{{weight_90d_slope}}': lambda pid: _safe_float('weight_90d_slope', pid, decimals=4),
|
|
'{{fm_28d_change}}': lambda pid: _safe_float('fm_28d_change', pid),
|
|
'{{lbm_28d_change}}': lambda pid: _safe_float('lbm_28d_change', pid),
|
|
'{{waist_28d_delta}}': lambda pid: _safe_float('waist_28d_delta', pid),
|
|
'{{hip_28d_delta}}': lambda pid: _safe_float('hip_28d_delta', pid),
|
|
'{{chest_28d_delta}}': lambda pid: _safe_float('chest_28d_delta', pid),
|
|
'{{arm_28d_delta}}': lambda pid: _safe_float('arm_28d_delta', pid),
|
|
'{{thigh_28d_delta}}': lambda pid: _safe_float('thigh_28d_delta', pid),
|
|
'{{waist_hip_ratio}}': lambda pid: _safe_float('waist_hip_ratio', pid, decimals=3),
|
|
'{{recomposition_quadrant}}': lambda pid: _safe_str('recomposition_quadrant', pid),
|
|
|
|
# --- Nutrition Metrics (E1-E5) ---
|
|
'{{energy_balance_7d}}': lambda pid: _safe_float('energy_balance_7d', pid, decimals=0),
|
|
'{{energy_deficit_surplus}}': lambda pid: _safe_str('energy_deficit_surplus', pid),
|
|
'{{protein_g_per_kg}}': lambda pid: _safe_float('protein_g_per_kg', pid),
|
|
'{{protein_days_in_target}}': lambda pid: _safe_str('protein_days_in_target', pid),
|
|
'{{protein_adequacy_28d}}': lambda pid: _safe_int('protein_adequacy_28d', pid),
|
|
'{{macro_consistency_score}}': lambda pid: _safe_int('macro_consistency_score', pid),
|
|
'{{intake_volatility}}': lambda pid: _safe_str('intake_volatility', pid),
|
|
|
|
# --- Activity Metrics (A1-A8) ---
|
|
'{{training_minutes_week}}': lambda pid: _safe_int('training_minutes_week', pid),
|
|
'{{training_frequency_7d}}': lambda pid: _safe_int('training_frequency_7d', pid),
|
|
'{{quality_sessions_pct}}': lambda pid: _safe_int('quality_sessions_pct', pid),
|
|
'{{ability_balance_strength}}': lambda pid: _safe_int('ability_balance_strength', pid),
|
|
'{{ability_balance_endurance}}': lambda pid: _safe_int('ability_balance_endurance', pid),
|
|
'{{ability_balance_mental}}': lambda pid: _safe_int('ability_balance_mental', pid),
|
|
'{{ability_balance_coordination}}': lambda pid: _safe_int('ability_balance_coordination', pid),
|
|
'{{ability_balance_mobility}}': lambda pid: _safe_int('ability_balance_mobility', pid),
|
|
'{{proxy_internal_load_7d}}': lambda pid: _safe_int('proxy_internal_load_7d', pid),
|
|
'{{monotony_score}}': lambda pid: _safe_float('monotony_score', pid),
|
|
'{{strain_score}}': lambda pid: _safe_int('strain_score', pid),
|
|
'{{rest_day_compliance}}': lambda pid: _safe_int('rest_day_compliance', pid),
|
|
'{{vo2max_trend_28d}}': lambda pid: _safe_float('vo2max_trend_28d', pid),
|
|
|
|
# --- Recovery Metrics (Recovery Score v2) ---
|
|
'{{hrv_vs_baseline_pct}}': lambda pid: _safe_float('hrv_vs_baseline_pct', pid),
|
|
'{{rhr_vs_baseline_pct}}': lambda pid: _safe_float('rhr_vs_baseline_pct', pid),
|
|
'{{sleep_avg_duration_7d}}': lambda pid: _safe_float('sleep_avg_duration_7d', pid),
|
|
'{{sleep_debt_hours}}': lambda pid: _safe_float('sleep_debt_hours', pid),
|
|
'{{sleep_regularity_proxy}}': lambda pid: _safe_float('sleep_regularity_proxy', pid),
|
|
'{{recent_load_balance_3d}}': lambda pid: _safe_int('recent_load_balance_3d', pid),
|
|
'{{sleep_quality_7d}}': lambda pid: _safe_int('sleep_quality_7d', pid),
|
|
|
|
# --- Correlation Metrics (C1-C7) ---
|
|
'{{correlation_energy_weight_lag}}': lambda pid: _safe_json('correlation_energy_weight_lag', pid),
|
|
'{{correlation_protein_lbm}}': lambda pid: _safe_json('correlation_protein_lbm', pid),
|
|
'{{correlation_load_hrv}}': lambda pid: _safe_json('correlation_load_hrv', pid),
|
|
'{{correlation_load_rhr}}': lambda pid: _safe_json('correlation_load_rhr', pid),
|
|
'{{correlation_sleep_recovery}}': lambda pid: _safe_json('correlation_sleep_recovery', pid),
|
|
'{{plateau_detected}}': lambda pid: _safe_json('plateau_detected', pid),
|
|
'{{top_drivers}}': lambda pid: _safe_json('top_drivers', pid),
|
|
|
|
# --- JSON/Markdown Structured Data (Ebene 5) ---
|
|
'{{active_goals_json}}': lambda pid: _safe_json('active_goals_json', pid),
|
|
'{{active_goals_md}}': lambda pid: _safe_str('active_goals_md', pid),
|
|
'{{focus_areas_weighted_json}}': lambda pid: _safe_json('focus_areas_weighted_json', pid),
|
|
'{{focus_areas_weighted_md}}': lambda pid: _safe_str('focus_areas_weighted_md', pid),
|
|
'{{focus_area_weights_json}}': lambda pid: _safe_json('focus_area_weights_json', pid),
|
|
'{{top_3_focus_areas}}': lambda pid: _safe_str('top_3_focus_areas', pid),
|
|
'{{top_3_goals_behind_schedule}}': lambda pid: _safe_str('top_3_goals_behind_schedule', pid),
|
|
'{{top_3_goals_on_track}}': lambda pid: _safe_str('top_3_goals_on_track', pid),
|
|
}
|
|
|
|
|
|
def calculate_bmi(profile_id: str) -> str:
|
|
"""Calculate BMI from latest weight and profile height."""
|
|
profile = get_profile_data(profile_id)
|
|
if not profile.get('height'):
|
|
return "nicht verfügbar"
|
|
|
|
with get_db() as conn:
|
|
cur = get_cursor(conn)
|
|
cur.execute(
|
|
"SELECT weight FROM weight_log WHERE profile_id=%s ORDER BY date DESC LIMIT 1",
|
|
(profile_id,)
|
|
)
|
|
row = cur.fetchone()
|
|
if not row:
|
|
return "nicht verfügbar"
|
|
|
|
height_m = profile['height'] / 100
|
|
bmi = row['weight'] / (height_m ** 2)
|
|
return f"{bmi:.1f}"
|
|
|
|
|
|
# ── Public API ────────────────────────────────────────────────────────────────
|
|
|
|
def resolve_placeholders(template: str, profile_id: str) -> str:
|
|
"""
|
|
Replace all {{placeholders}} in template with actual user data.
|
|
|
|
Args:
|
|
template: Prompt template with placeholders
|
|
profile_id: User profile ID
|
|
|
|
Returns:
|
|
Resolved template with placeholders replaced by values
|
|
"""
|
|
result = template
|
|
|
|
for placeholder, resolver in PLACEHOLDER_MAP.items():
|
|
if placeholder in result:
|
|
try:
|
|
value = resolver(profile_id)
|
|
result = result.replace(placeholder, str(value))
|
|
except Exception as e:
|
|
# On error, replace with error message
|
|
result = result.replace(placeholder, f"[Fehler: {placeholder}]")
|
|
|
|
return result
|
|
|
|
|
|
def get_unknown_placeholders(template: str) -> List[str]:
|
|
"""
|
|
Find all placeholders in template that are not in PLACEHOLDER_MAP.
|
|
|
|
Args:
|
|
template: Prompt template
|
|
|
|
Returns:
|
|
List of unknown placeholder names (without {{}})
|
|
"""
|
|
# Find all {{...}} patterns
|
|
found = re.findall(r'\{\{(\w+)\}\}', template)
|
|
|
|
# Filter to only unknown ones
|
|
known_names = {p.strip('{}') for p in PLACEHOLDER_MAP.keys()}
|
|
unknown = [p for p in found if p not in known_names]
|
|
|
|
return list(set(unknown)) # Remove duplicates
|
|
|
|
|
|
def get_available_placeholders(categories: Optional[List[str]] = None) -> Dict[str, List[str]]:
|
|
"""
|
|
Get available placeholders, optionally filtered by categories.
|
|
|
|
Args:
|
|
categories: Optional list of categories to filter (körper, ernährung, training, etc.)
|
|
|
|
Returns:
|
|
Dict mapping category to list of placeholders
|
|
"""
|
|
placeholder_categories = {
|
|
'profil': [
|
|
'{{name}}', '{{age}}', '{{height}}', '{{geschlecht}}'
|
|
],
|
|
'körper': [
|
|
'{{weight_aktuell}}', '{{weight_trend}}', '{{kf_aktuell}}', '{{bmi}}'
|
|
],
|
|
'ernährung': [
|
|
'{{kcal_avg}}', '{{protein_avg}}', '{{carb_avg}}', '{{fat_avg}}'
|
|
],
|
|
'training': [
|
|
'{{activity_summary}}', '{{trainingstyp_verteilung}}'
|
|
],
|
|
'zeitraum': [
|
|
'{{datum_heute}}', '{{zeitraum_7d}}', '{{zeitraum_30d}}', '{{zeitraum_90d}}'
|
|
]
|
|
}
|
|
|
|
if not categories:
|
|
return placeholder_categories
|
|
|
|
# Filter to requested categories
|
|
return {k: v for k, v in placeholder_categories.items() if k in categories}
|
|
|
|
|
|
def get_placeholder_example_values(profile_id: str) -> Dict[str, str]:
|
|
"""
|
|
Get example values for all placeholders using real user data.
|
|
|
|
Args:
|
|
profile_id: User profile ID
|
|
|
|
Returns:
|
|
Dict mapping placeholder to example value
|
|
"""
|
|
examples = {}
|
|
|
|
for placeholder, resolver in PLACEHOLDER_MAP.items():
|
|
try:
|
|
examples[placeholder] = resolver(profile_id)
|
|
except Exception as e:
|
|
examples[placeholder] = f"[Fehler: {str(e)}]"
|
|
|
|
return examples
|
|
|
|
|
|
def get_placeholder_catalog(profile_id: str) -> Dict[str, List[Dict[str, str]]]:
|
|
"""
|
|
Get grouped placeholder catalog with descriptions and example values.
|
|
|
|
Args:
|
|
profile_id: User profile ID
|
|
|
|
Returns:
|
|
Dict mapping category to list of {key, description, example}
|
|
"""
|
|
# Placeholder definitions with descriptions
|
|
placeholders = {
|
|
'Profil': [
|
|
('name', 'Name des Nutzers'),
|
|
('age', 'Alter in Jahren'),
|
|
('height', 'Körpergröße in cm'),
|
|
('geschlecht', 'Geschlecht'),
|
|
],
|
|
'Körper': [
|
|
('weight_aktuell', 'Aktuelles Gewicht in kg'),
|
|
('weight_trend', 'Gewichtstrend (7d/30d)'),
|
|
('kf_aktuell', 'Aktueller Körperfettanteil in %'),
|
|
('bmi', 'Body Mass Index'),
|
|
('weight_7d_median', 'Gewicht 7d Median (kg)'),
|
|
('weight_28d_slope', 'Gewichtstrend 28d (kg/Tag)'),
|
|
('fm_28d_change', 'Fettmasse Änderung 28d (kg)'),
|
|
('lbm_28d_change', 'Magermasse Änderung 28d (kg)'),
|
|
('waist_28d_delta', 'Taillenumfang Änderung 28d (cm)'),
|
|
('waist_hip_ratio', 'Taille/Hüfte-Verhältnis'),
|
|
('recomposition_quadrant', 'Rekomposition-Status'),
|
|
],
|
|
'Ernährung': [
|
|
('kcal_avg', 'Durchschn. Kalorien (30d)'),
|
|
('protein_avg', 'Durchschn. Protein in g (30d)'),
|
|
('carb_avg', 'Durchschn. Kohlenhydrate in g (30d)'),
|
|
('fat_avg', 'Durchschn. Fett in g (30d)'),
|
|
('energy_balance_7d', 'Energiebilanz 7d (kcal/Tag)'),
|
|
('protein_g_per_kg', 'Protein g/kg Körpergewicht'),
|
|
('protein_adequacy_28d', 'Protein Adequacy Score (0-100)'),
|
|
('macro_consistency_score', 'Makro-Konsistenz Score (0-100)'),
|
|
],
|
|
'Training': [
|
|
('activity_summary', 'Aktivitäts-Zusammenfassung (7d)'),
|
|
('trainingstyp_verteilung', 'Verteilung nach Trainingstypen'),
|
|
('training_minutes_week', 'Trainingsminuten pro Woche'),
|
|
('training_frequency_7d', 'Trainingshäufigkeit 7d'),
|
|
('quality_sessions_pct', 'Qualitätssessions (%)'),
|
|
('ability_balance_strength', 'Ability Balance - Kraft (0-100)'),
|
|
('ability_balance_endurance', 'Ability Balance - Ausdauer (0-100)'),
|
|
('proxy_internal_load_7d', 'Proxy Load 7d'),
|
|
('rest_day_compliance', 'Ruhetags-Compliance (%)'),
|
|
],
|
|
'Schlaf & Erholung': [
|
|
('sleep_avg_duration', 'Durchschn. Schlafdauer (7d)'),
|
|
('sleep_avg_quality', 'Durchschn. Schlafqualität (7d)'),
|
|
('rest_days_count', 'Anzahl Ruhetage (30d)'),
|
|
('sleep_avg_duration_7d', 'Schlaf 7d (Stunden)'),
|
|
('sleep_debt_hours', 'Schlafschuld (Stunden)'),
|
|
('sleep_regularity_proxy', 'Schlaf-Regelmäßigkeit (Min Abweichung)'),
|
|
('sleep_quality_7d', 'Schlafqualität 7d (0-100)'),
|
|
],
|
|
'Vitalwerte': [
|
|
('vitals_avg_hr', 'Durchschn. Ruhepuls (7d)'),
|
|
('vitals_avg_hrv', 'Durchschn. HRV (7d)'),
|
|
('vitals_vo2_max', 'Aktueller VO2 Max'),
|
|
('hrv_vs_baseline_pct', 'HRV vs. Baseline (%)'),
|
|
('rhr_vs_baseline_pct', 'RHR vs. Baseline (%)'),
|
|
('vo2max_trend_28d', 'VO2max Trend 28d'),
|
|
],
|
|
'Scores (Phase 0b)': [
|
|
('goal_progress_score', 'Goal Progress Score (0-100)'),
|
|
('body_progress_score', 'Body Progress Score (0-100)'),
|
|
('nutrition_score', 'Nutrition Score (0-100)'),
|
|
('activity_score', 'Activity Score (0-100)'),
|
|
('recovery_score', 'Recovery Score (0-100)'),
|
|
('data_quality_score', 'Data Quality Score (0-100)'),
|
|
],
|
|
'Focus Areas': [
|
|
('top_focus_area_name', 'Top Focus Area Name'),
|
|
('top_focus_area_progress', 'Top Focus Area Progress (%)'),
|
|
('focus_cat_körper_progress', 'Kategorie Körper - Progress (%)'),
|
|
('focus_cat_körper_weight', 'Kategorie Körper - Gewichtung (%)'),
|
|
('focus_cat_ernährung_progress', 'Kategorie Ernährung - Progress (%)'),
|
|
('focus_cat_ernährung_weight', 'Kategorie Ernährung - Gewichtung (%)'),
|
|
('focus_cat_aktivität_progress', 'Kategorie Aktivität - Progress (%)'),
|
|
('focus_cat_aktivität_weight', 'Kategorie Aktivität - Gewichtung (%)'),
|
|
],
|
|
'Zeitraum': [
|
|
('datum_heute', 'Heutiges Datum'),
|
|
('zeitraum_7d', '7-Tage-Zeitraum'),
|
|
('zeitraum_30d', '30-Tage-Zeitraum'),
|
|
],
|
|
}
|
|
|
|
catalog = {}
|
|
|
|
for category, items in placeholders.items():
|
|
catalog[category] = []
|
|
for key, description in items:
|
|
placeholder = f'{{{{{key}}}}}'
|
|
# Get example value if resolver exists
|
|
resolver = PLACEHOLDER_MAP.get(placeholder)
|
|
if resolver:
|
|
try:
|
|
example = resolver(profile_id)
|
|
except Exception:
|
|
example = '[Nicht verfügbar]'
|
|
else:
|
|
example = '[Nicht implementiert]'
|
|
|
|
catalog[category].append({
|
|
'key': key,
|
|
'description': description,
|
|
'example': str(example)
|
|
})
|
|
|
|
return catalog
|