Lars
8e67175ed2
OLD: Showed 3 goals with lowest progress %
NEW: Calculates expected progress based on elapsed time vs. total time
Shows goals with largest negative deviation (behind schedule)
Example Weight Goal:
- Total time: 98 days (22.02 - 31.05)
- Elapsed: 34 days (35%)
- Actual progress: 41%
- Deviation: +7% (AHEAD, not behind)
Also updated on_track to show goals with positive deviation (ahead of schedule).
Note: Linear progress is a simplification. Real-world progress curves vary
by goal type (weight loss, muscle gain, VO2max, etc). Future: AI-based
projection models for more realistic expectations.
1417 lines
58 KiB
Python
1417 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.
|
|
"""
|
|
import re
|
|
from datetime import datetime, timedelta
|
|
from typing import Dict, List, Optional, Callable
|
|
from db import get_db, get_cursor, r2d
|
|
|
|
|
|
# ── 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."""
|
|
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()
|
|
return f"{row['weight']:.1f} kg" if row else "nicht verfügbar"
|
|
|
|
|
|
def get_weight_trend(profile_id: str, days: int = 28) -> str:
|
|
"""Calculate weight trend description."""
|
|
with get_db() as conn:
|
|
cur = get_cursor(conn)
|
|
cutoff = (datetime.now() - timedelta(days=days)).strftime('%Y-%m-%d')
|
|
cur.execute(
|
|
"""SELECT weight, date FROM weight_log
|
|
WHERE profile_id=%s AND date >= %s
|
|
ORDER BY date""",
|
|
(profile_id, cutoff)
|
|
)
|
|
rows = [r2d(r) for r in cur.fetchall()]
|
|
|
|
if len(rows) < 2:
|
|
return "nicht genug Daten"
|
|
|
|
first = rows[0]['weight']
|
|
last = rows[-1]['weight']
|
|
delta = last - first
|
|
|
|
if abs(delta) < 0.3:
|
|
return "stabil"
|
|
elif delta > 0:
|
|
return f"steigend (+{delta:.1f} kg in {days} Tagen)"
|
|
else:
|
|
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."""
|
|
with get_db() as conn:
|
|
cur = get_cursor(conn)
|
|
cur.execute(
|
|
"""SELECT body_fat_pct FROM caliper_log
|
|
WHERE profile_id=%s AND body_fat_pct IS NOT NULL
|
|
ORDER BY date DESC LIMIT 1""",
|
|
(profile_id,)
|
|
)
|
|
row = cur.fetchone()
|
|
return f"{row['body_fat_pct']:.1f}%" if row else "nicht verfügbar"
|
|
|
|
|
|
def get_nutrition_avg(profile_id: str, field: str, days: int = 30) -> str:
|
|
"""Calculate average nutrition value."""
|
|
with get_db() as conn:
|
|
cur = get_cursor(conn)
|
|
cutoff = (datetime.now() - timedelta(days=days)).strftime('%Y-%m-%d')
|
|
|
|
# Map field names to actual column names
|
|
field_map = {
|
|
'protein': 'protein_g',
|
|
'fat': 'fat_g',
|
|
'carb': 'carbs_g',
|
|
'kcal': 'kcal'
|
|
}
|
|
db_field = field_map.get(field, field)
|
|
|
|
cur.execute(
|
|
f"""SELECT AVG({db_field}) as avg FROM nutrition_log
|
|
WHERE profile_id=%s AND date >= %s AND {db_field} IS NOT NULL""",
|
|
(profile_id, cutoff)
|
|
)
|
|
row = cur.fetchone()
|
|
if row and row['avg']:
|
|
if field == 'kcal':
|
|
return f"{int(row['avg'])} kcal/Tag (Ø {days} Tage)"
|
|
else:
|
|
return f"{int(row['avg'])}g/Tag (Ø {days} Tage)"
|
|
return "nicht verfügbar"
|
|
|
|
|
|
def get_caliper_summary(profile_id: str) -> str:
|
|
"""Get latest caliper measurements summary."""
|
|
with get_db() as conn:
|
|
cur = get_cursor(conn)
|
|
cur.execute(
|
|
"""SELECT body_fat_pct, sf_method, date FROM caliper_log
|
|
WHERE profile_id=%s AND body_fat_pct IS NOT NULL
|
|
ORDER BY date DESC LIMIT 1""",
|
|
(profile_id,)
|
|
)
|
|
row = r2d(cur.fetchone()) if cur.rowcount > 0 else None
|
|
|
|
if not row:
|
|
return "keine Caliper-Messungen"
|
|
|
|
method = row.get('sf_method', 'unbekannt')
|
|
return f"{row['body_fat_pct']:.1f}% ({method} am {row['date']})"
|
|
|
|
|
|
def get_circ_summary(profile_id: str) -> str:
|
|
"""Get latest circumference measurements summary with age annotations.
|
|
|
|
For each measurement point, fetches the most recent value (even if from different dates).
|
|
Annotates each value with measurement age for AI context.
|
|
"""
|
|
with get_db() as conn:
|
|
cur = get_cursor(conn)
|
|
|
|
# Define all circumference points with their labels
|
|
fields = [
|
|
('c_neck', 'Nacken'),
|
|
('c_chest', 'Brust'),
|
|
('c_waist', 'Taille'),
|
|
('c_belly', 'Bauch'),
|
|
('c_hip', 'Hüfte'),
|
|
('c_thigh', 'Oberschenkel'),
|
|
('c_calf', 'Wade'),
|
|
('c_arm', 'Arm')
|
|
]
|
|
|
|
parts = []
|
|
today = datetime.now().date()
|
|
|
|
# Get latest value for each field individually
|
|
for field_name, label in fields:
|
|
cur.execute(
|
|
f"""SELECT {field_name}, date,
|
|
CURRENT_DATE - date AS age_days
|
|
FROM circumference_log
|
|
WHERE profile_id=%s AND {field_name} IS NOT NULL
|
|
ORDER BY date DESC LIMIT 1""",
|
|
(profile_id,)
|
|
)
|
|
row = r2d(cur.fetchone()) if cur.rowcount > 0 else None
|
|
|
|
if row:
|
|
value = row[field_name]
|
|
age_days = row['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"{label} {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."""
|
|
with get_db() as conn:
|
|
cur = get_cursor(conn)
|
|
cutoff = (datetime.now() - timedelta(days=days)).strftime('%Y-%m-%d')
|
|
cur.execute(
|
|
"""SELECT COUNT(DISTINCT date) as days FROM nutrition_log
|
|
WHERE profile_id=%s AND date >= %s""",
|
|
(profile_id, cutoff)
|
|
)
|
|
row = cur.fetchone()
|
|
return str(row['days']) if row else "0"
|
|
|
|
|
|
def get_protein_ziel_low(profile_id: str) -> str:
|
|
"""Calculate lower protein target based on current weight (1.6g/kg)."""
|
|
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 row:
|
|
return f"{int(float(row['weight']) * 1.6)}"
|
|
return "nicht verfügbar"
|
|
|
|
|
|
def get_protein_ziel_high(profile_id: str) -> str:
|
|
"""Calculate upper protein target based on current weight (2.2g/kg)."""
|
|
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 row:
|
|
return f"{int(float(row['weight']) * 2.2)}"
|
|
return "nicht verfügbar"
|
|
|
|
|
|
def get_activity_summary(profile_id: str, days: int = 14) -> str:
|
|
"""Get activity summary for recent period."""
|
|
with get_db() as conn:
|
|
cur = get_cursor(conn)
|
|
cutoff = (datetime.now() - timedelta(days=days)).strftime('%Y-%m-%d')
|
|
cur.execute(
|
|
"""SELECT COUNT(*) as count,
|
|
SUM(duration_min) as total_min,
|
|
SUM(kcal_active) as total_kcal
|
|
FROM activity_log
|
|
WHERE profile_id=%s AND date >= %s""",
|
|
(profile_id, cutoff)
|
|
)
|
|
row = r2d(cur.fetchone())
|
|
|
|
if row['count'] == 0:
|
|
return f"Keine Aktivitäten in den letzten {days} Tagen"
|
|
|
|
avg_min = int(row['total_min'] / row['count']) if row['total_min'] else 0
|
|
return f"{row['count']} Einheiten in {days} Tagen (Ø {avg_min} min/Einheit, {int(row['total_kcal'] or 0)} 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."""
|
|
with get_db() as conn:
|
|
cur = get_cursor(conn)
|
|
cutoff = (datetime.now() - timedelta(days=days)).strftime('%Y-%m-%d')
|
|
cur.execute(
|
|
"""SELECT date, activity_type, duration_min, kcal_active, hr_avg
|
|
FROM activity_log
|
|
WHERE profile_id=%s AND date >= %s
|
|
ORDER BY date DESC
|
|
LIMIT 50""",
|
|
(profile_id, cutoff)
|
|
)
|
|
rows = [r2d(r) for r in cur.fetchall()]
|
|
|
|
if not rows:
|
|
return f"Keine Aktivitäten in den letzten {days} Tagen"
|
|
|
|
# Format as readable list
|
|
lines = []
|
|
for r in rows:
|
|
hr_str = f" HF={r['hr_avg']}" if r.get('hr_avg') else ""
|
|
lines.append(
|
|
f"{r['date']}: {r['activity_type']} ({r['duration_min']}min, {r.get('kcal_active', 0)}kcal{hr_str})"
|
|
)
|
|
|
|
return '\n'.join(lines[:20]) # Max 20 entries to avoid token bloat
|
|
|
|
|
|
def get_trainingstyp_verteilung(profile_id: str, days: int = 14) -> str:
|
|
"""Get training type distribution."""
|
|
with get_db() as conn:
|
|
cur = get_cursor(conn)
|
|
cutoff = (datetime.now() - timedelta(days=days)).strftime('%Y-%m-%d')
|
|
cur.execute(
|
|
"""SELECT training_category, COUNT(*) as count
|
|
FROM activity_log
|
|
WHERE profile_id=%s AND date >= %s AND training_category IS NOT NULL
|
|
GROUP BY training_category
|
|
ORDER BY count DESC""",
|
|
(profile_id, cutoff)
|
|
)
|
|
rows = [r2d(r) for r in cur.fetchall()]
|
|
|
|
if not rows:
|
|
return "Keine kategorisierten Trainings"
|
|
|
|
total = sum(r['count'] for r in rows)
|
|
parts = [f"{r['training_category']}: {int(r['count']/total*100)}%" for r in rows[:3]]
|
|
return ", ".join(parts)
|
|
|
|
|
|
def get_sleep_avg_duration(profile_id: str, days: int = 7) -> str:
|
|
"""Calculate average sleep duration in hours."""
|
|
with get_db() as conn:
|
|
cur = get_cursor(conn)
|
|
cutoff = (datetime.now() - timedelta(days=days)).strftime('%Y-%m-%d')
|
|
cur.execute(
|
|
"""SELECT sleep_segments FROM sleep_log
|
|
WHERE profile_id=%s AND date >= %s
|
|
ORDER BY date DESC""",
|
|
(profile_id, cutoff)
|
|
)
|
|
rows = cur.fetchall()
|
|
|
|
if not rows:
|
|
return "nicht verfügbar"
|
|
|
|
total_minutes = 0
|
|
for row in rows:
|
|
segments = row['sleep_segments']
|
|
if segments:
|
|
# Sum duration_min from all segments
|
|
for seg in segments:
|
|
total_minutes += seg.get('duration_min', 0)
|
|
|
|
if total_minutes == 0:
|
|
return "nicht verfügbar"
|
|
|
|
avg_hours = total_minutes / len(rows) / 60
|
|
return f"{avg_hours:.1f}h"
|
|
|
|
|
|
def get_sleep_avg_quality(profile_id: str, days: int = 7) -> str:
|
|
"""Calculate average sleep quality (Deep+REM %)."""
|
|
with get_db() as conn:
|
|
cur = get_cursor(conn)
|
|
cutoff = (datetime.now() - timedelta(days=days)).strftime('%Y-%m-%d')
|
|
cur.execute(
|
|
"""SELECT sleep_segments FROM sleep_log
|
|
WHERE profile_id=%s AND date >= %s
|
|
ORDER BY date DESC""",
|
|
(profile_id, cutoff)
|
|
)
|
|
rows = cur.fetchall()
|
|
|
|
if not rows:
|
|
return "nicht verfügbar"
|
|
|
|
total_quality = 0
|
|
count = 0
|
|
for row in rows:
|
|
segments = row['sleep_segments']
|
|
if segments:
|
|
# Note: segments use 'phase' key (not 'stage'), stored lowercase (deep, rem, light, awake)
|
|
deep_rem_min = sum(s.get('duration_min', 0) for s in segments if s.get('phase') in ['deep', 'rem'])
|
|
total_min = sum(s.get('duration_min', 0) for s in segments)
|
|
if total_min > 0:
|
|
quality_pct = (deep_rem_min / total_min) * 100
|
|
total_quality += quality_pct
|
|
count += 1
|
|
|
|
if count == 0:
|
|
return "nicht verfügbar"
|
|
|
|
avg_quality = total_quality / count
|
|
return f"{avg_quality:.0f}% (Deep+REM)"
|
|
|
|
|
|
def get_rest_days_count(profile_id: str, days: int = 30) -> str:
|
|
"""Count rest days in the given period."""
|
|
with get_db() as conn:
|
|
cur = get_cursor(conn)
|
|
cutoff = (datetime.now() - timedelta(days=days)).strftime('%Y-%m-%d')
|
|
cur.execute(
|
|
"""SELECT COUNT(DISTINCT date) as count FROM rest_days
|
|
WHERE profile_id=%s AND date >= %s""",
|
|
(profile_id, cutoff)
|
|
)
|
|
row = cur.fetchone()
|
|
count = row['count'] if row else 0
|
|
return f"{count} Ruhetage"
|
|
|
|
|
|
def get_vitals_avg_hr(profile_id: str, days: int = 7) -> str:
|
|
"""Calculate average resting heart rate."""
|
|
with get_db() as conn:
|
|
cur = get_cursor(conn)
|
|
cutoff = (datetime.now() - timedelta(days=days)).strftime('%Y-%m-%d')
|
|
cur.execute(
|
|
"""SELECT AVG(resting_hr) as avg FROM vitals_baseline
|
|
WHERE profile_id=%s AND date >= %s AND resting_hr IS NOT NULL""",
|
|
(profile_id, cutoff)
|
|
)
|
|
row = cur.fetchone()
|
|
|
|
if row and row['avg']:
|
|
return f"{int(row['avg'])} bpm"
|
|
return "nicht verfügbar"
|
|
|
|
|
|
def get_vitals_avg_hrv(profile_id: str, days: int = 7) -> str:
|
|
"""Calculate average heart rate variability."""
|
|
with get_db() as conn:
|
|
cur = get_cursor(conn)
|
|
cutoff = (datetime.now() - timedelta(days=days)).strftime('%Y-%m-%d')
|
|
cur.execute(
|
|
"""SELECT AVG(hrv) as avg FROM vitals_baseline
|
|
WHERE profile_id=%s AND date >= %s AND hrv IS NOT NULL""",
|
|
(profile_id, cutoff)
|
|
)
|
|
row = cur.fetchone()
|
|
|
|
if row and row['avg']:
|
|
return f"{int(row['avg'])} ms"
|
|
return "nicht verfügbar"
|
|
|
|
|
|
def get_vitals_vo2_max(profile_id: str) -> str:
|
|
"""Get latest VO2 Max value."""
|
|
with get_db() as conn:
|
|
cur = get_cursor(conn)
|
|
cur.execute(
|
|
"""SELECT vo2_max FROM vitals_baseline
|
|
WHERE profile_id=%s AND vo2_max IS NOT NULL
|
|
ORDER BY date DESC LIMIT 1""",
|
|
(profile_id,)
|
|
)
|
|
row = cur.fetchone()
|
|
|
|
if row and row['vo2_max']:
|
|
return f"{row['vo2_max']:.1f} ml/kg/min"
|
|
return "nicht verfügbar"
|
|
|
|
|
|
# ── 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 calculations import scores, body_metrics, nutrition_metrics, activity_metrics, recovery_metrics, 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 calculations 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 calculations import body_metrics, nutrition_metrics, activity_metrics, scores, 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 calculations import scores, 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:
|
|
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):
|
|
continue
|
|
|
|
if not goals_with_deviation:
|
|
return 'Keine Ziele mit Zeitvorgabe'
|
|
|
|
# Sort by deviation ascending (most negative first = most behind)
|
|
# Only include goals that are actually behind (deviation < 0)
|
|
behind_goals = [g for g in goals_with_deviation if g['_deviation'] < 0]
|
|
|
|
if not behind_goals:
|
|
return 'Alle Ziele im Zeitplan'
|
|
|
|
sorted_goals = sorted(behind_goals, key=lambda x: x['_deviation'])[:n]
|
|
|
|
lines = []
|
|
for goal in sorted_goals:
|
|
name = goal.get('name') or goal.get('goal_type', 'Unbekannt')
|
|
actual = goal['_actual_progress']
|
|
expected = goal['_expected_progress']
|
|
deviation = goal['_deviation']
|
|
lines.append(f"{name} ({actual}% statt {expected}%, {deviation}%)")
|
|
|
|
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:
|
|
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):
|
|
continue
|
|
|
|
if not goals_with_deviation:
|
|
return 'Keine Ziele mit Zeitvorgabe'
|
|
|
|
# Sort by deviation descending (most positive first = most ahead)
|
|
# Only include goals that are ahead or on track (deviation >= 0)
|
|
ahead_goals = [g for g in goals_with_deviation if g['_deviation'] >= 0]
|
|
|
|
if not ahead_goals:
|
|
return 'Keine Ziele im Zeitplan'
|
|
|
|
sorted_goals = sorted(ahead_goals, key=lambda x: x['_deviation'], reverse=True)[:n]
|
|
|
|
lines = []
|
|
for goal in sorted_goals:
|
|
name = goal.get('name') or goal.get('goal_type', 'Unbekannt')
|
|
actual = goal['_actual_progress']
|
|
expected = goal['_expected_progress']
|
|
deviation = goal['_deviation']
|
|
lines.append(f"{name} ({actual}%, +{deviation}% voraus)")
|
|
|
|
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
|