Lars
337667fc07
- Strategic Layer: Goal modes (weight_loss, strength, endurance, recomposition, health) - Tactical Layer: Concrete goal targets with progress tracking - Training phases (manual + auto-detection framework) - Fitness tests (standardized performance tracking) Backend: - Migration 022: goal_mode in profiles, goals, training_phases, fitness_tests tables - New router: routers/goals.py with full CRUD for goals, phases, tests - API endpoints: /api/goals/* (mode, list, create, update, delete) Frontend: - GoalsPage: Goal mode selector + goal management UI - Dashboard: Goals preview card with link - API integration: goal mode, CRUD operations, progress calculation Basis for 120+ placeholders and goal-aware analyses (Phase 0b) Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
474 lines
16 KiB
Python
474 lines
16 KiB
Python
"""
|
|
Goals Router - Goal System (Strategic + Tactical)
|
|
|
|
Endpoints for managing:
|
|
- Strategic goal modes (weight_loss, strength, etc.)
|
|
- Tactical goal targets (concrete values with deadlines)
|
|
- Training phase detection
|
|
- Fitness tests
|
|
|
|
Part of v9e Goal System implementation.
|
|
"""
|
|
from fastapi import APIRouter, Depends, HTTPException
|
|
from pydantic import BaseModel
|
|
from typing import Optional, List
|
|
from datetime import date, datetime, timedelta
|
|
from decimal import Decimal
|
|
import json
|
|
|
|
from db import get_db, get_cursor, r2d
|
|
from auth import require_auth
|
|
|
|
router = APIRouter(prefix="/api/goals", tags=["goals"])
|
|
|
|
# ============================================================================
|
|
# Pydantic Models
|
|
# ============================================================================
|
|
|
|
class GoalModeUpdate(BaseModel):
|
|
"""Update strategic goal mode"""
|
|
goal_mode: str # weight_loss, strength, endurance, recomposition, health
|
|
|
|
class GoalCreate(BaseModel):
|
|
"""Create or update a concrete goal"""
|
|
goal_type: str # weight, body_fat, lean_mass, vo2max, strength, flexibility, bp, rhr
|
|
is_primary: bool = False
|
|
target_value: float
|
|
unit: str # kg, %, ml/kg/min, bpm, mmHg, cm, reps
|
|
target_date: Optional[date] = None
|
|
name: Optional[str] = None
|
|
description: Optional[str] = None
|
|
|
|
class GoalUpdate(BaseModel):
|
|
"""Update existing goal"""
|
|
target_value: Optional[float] = None
|
|
target_date: Optional[date] = None
|
|
status: Optional[str] = None # active, reached, abandoned, expired
|
|
name: Optional[str] = None
|
|
description: Optional[str] = None
|
|
|
|
class TrainingPhaseCreate(BaseModel):
|
|
"""Create training phase (manual or auto-detected)"""
|
|
phase_type: str # calorie_deficit, calorie_surplus, deload, maintenance, periodization
|
|
start_date: date
|
|
end_date: Optional[date] = None
|
|
notes: Optional[str] = None
|
|
|
|
class FitnessTestCreate(BaseModel):
|
|
"""Record fitness test result"""
|
|
test_type: str
|
|
result_value: float
|
|
result_unit: str
|
|
test_date: date
|
|
test_conditions: Optional[str] = None
|
|
|
|
# ============================================================================
|
|
# Strategic Layer: Goal Modes
|
|
# ============================================================================
|
|
|
|
@router.get("/mode")
|
|
def get_goal_mode(session: dict = Depends(require_auth)):
|
|
"""Get user's current strategic goal mode"""
|
|
pid = session['profile_id']
|
|
|
|
with get_db() as conn:
|
|
cur = get_cursor(conn)
|
|
cur.execute(
|
|
"SELECT goal_mode FROM profiles WHERE id = %s",
|
|
(pid,)
|
|
)
|
|
row = cur.fetchone()
|
|
if not row:
|
|
raise HTTPException(status_code=404, detail="Profil nicht gefunden")
|
|
|
|
return {
|
|
"goal_mode": row['goal_mode'] or 'health',
|
|
"description": _get_goal_mode_description(row['goal_mode'] or 'health')
|
|
}
|
|
|
|
@router.put("/mode")
|
|
def update_goal_mode(data: GoalModeUpdate, session: dict = Depends(require_auth)):
|
|
"""Update user's strategic goal mode"""
|
|
pid = session['profile_id']
|
|
|
|
# Validate goal mode
|
|
valid_modes = ['weight_loss', 'strength', 'endurance', 'recomposition', 'health']
|
|
if data.goal_mode not in valid_modes:
|
|
raise HTTPException(
|
|
status_code=400,
|
|
detail=f"Ungültiger Goal Mode. Erlaubt: {', '.join(valid_modes)}"
|
|
)
|
|
|
|
with get_db() as conn:
|
|
cur = get_cursor(conn)
|
|
cur.execute(
|
|
"UPDATE profiles SET goal_mode = %s WHERE id = %s",
|
|
(data.goal_mode, pid)
|
|
)
|
|
|
|
return {
|
|
"goal_mode": data.goal_mode,
|
|
"description": _get_goal_mode_description(data.goal_mode)
|
|
}
|
|
|
|
def _get_goal_mode_description(mode: str) -> str:
|
|
"""Get description for goal mode"""
|
|
descriptions = {
|
|
'weight_loss': 'Gewichtsreduktion (Kaloriendefizit, Fettabbau)',
|
|
'strength': 'Kraftaufbau (Muskelwachstum, progressive Belastung)',
|
|
'endurance': 'Ausdauer (VO2Max, aerobe Kapazität)',
|
|
'recomposition': 'Körperkomposition (gleichzeitig Fett ab- und Muskeln aufbauen)',
|
|
'health': 'Allgemeine Gesundheit (ausgewogen, präventiv)'
|
|
}
|
|
return descriptions.get(mode, 'Unbekannt')
|
|
|
|
# ============================================================================
|
|
# Tactical Layer: Concrete Goals
|
|
# ============================================================================
|
|
|
|
@router.get("/list")
|
|
def list_goals(session: dict = Depends(require_auth)):
|
|
"""List all goals for current user"""
|
|
pid = session['profile_id']
|
|
|
|
with get_db() as conn:
|
|
cur = get_cursor(conn)
|
|
cur.execute("""
|
|
SELECT id, goal_type, is_primary, status,
|
|
target_value, current_value, start_value, unit,
|
|
start_date, target_date, reached_date,
|
|
name, description,
|
|
progress_pct, projection_date, on_track,
|
|
created_at, updated_at
|
|
FROM goals
|
|
WHERE profile_id = %s
|
|
ORDER BY is_primary DESC, created_at DESC
|
|
""", (pid,))
|
|
|
|
goals = [r2d(row) for row in cur.fetchall()]
|
|
|
|
# Update current values for each goal
|
|
for goal in goals:
|
|
_update_goal_progress(conn, pid, goal)
|
|
|
|
return goals
|
|
|
|
@router.post("/create")
|
|
def create_goal(data: GoalCreate, session: dict = Depends(require_auth)):
|
|
"""Create new goal"""
|
|
pid = session['profile_id']
|
|
|
|
with get_db() as conn:
|
|
cur = get_cursor(conn)
|
|
|
|
# If this is set as primary, unset other primary goals
|
|
if data.is_primary:
|
|
cur.execute(
|
|
"UPDATE goals SET is_primary = false WHERE profile_id = %s",
|
|
(pid,)
|
|
)
|
|
|
|
# Get current value for this goal type
|
|
current_value = _get_current_value_for_goal_type(conn, pid, data.goal_type)
|
|
|
|
# Insert goal
|
|
cur.execute("""
|
|
INSERT INTO goals (
|
|
profile_id, goal_type, is_primary,
|
|
target_value, current_value, start_value, unit,
|
|
target_date, name, description
|
|
) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s)
|
|
RETURNING id
|
|
""", (
|
|
pid, data.goal_type, data.is_primary,
|
|
data.target_value, current_value, current_value, data.unit,
|
|
data.target_date, data.name, data.description
|
|
))
|
|
|
|
goal_id = cur.fetchone()['id']
|
|
|
|
return {"id": goal_id, "message": "Ziel erstellt"}
|
|
|
|
@router.put("/{goal_id}")
|
|
def update_goal(goal_id: str, data: GoalUpdate, session: dict = Depends(require_auth)):
|
|
"""Update existing goal"""
|
|
pid = session['profile_id']
|
|
|
|
with get_db() as conn:
|
|
cur = get_cursor(conn)
|
|
|
|
# Verify ownership
|
|
cur.execute(
|
|
"SELECT id FROM goals WHERE id = %s AND profile_id = %s",
|
|
(goal_id, pid)
|
|
)
|
|
if not cur.fetchone():
|
|
raise HTTPException(status_code=404, detail="Ziel nicht gefunden")
|
|
|
|
# Build update query dynamically
|
|
updates = []
|
|
params = []
|
|
|
|
if data.target_value is not None:
|
|
updates.append("target_value = %s")
|
|
params.append(data.target_value)
|
|
|
|
if data.target_date is not None:
|
|
updates.append("target_date = %s")
|
|
params.append(data.target_date)
|
|
|
|
if data.status is not None:
|
|
updates.append("status = %s")
|
|
params.append(data.status)
|
|
if data.status == 'reached':
|
|
updates.append("reached_date = CURRENT_DATE")
|
|
|
|
if data.name is not None:
|
|
updates.append("name = %s")
|
|
params.append(data.name)
|
|
|
|
if data.description is not None:
|
|
updates.append("description = %s")
|
|
params.append(data.description)
|
|
|
|
if not updates:
|
|
raise HTTPException(status_code=400, detail="Keine Änderungen angegeben")
|
|
|
|
updates.append("updated_at = NOW()")
|
|
params.extend([goal_id, pid])
|
|
|
|
cur.execute(
|
|
f"UPDATE goals SET {', '.join(updates)} WHERE id = %s AND profile_id = %s",
|
|
tuple(params)
|
|
)
|
|
|
|
return {"message": "Ziel aktualisiert"}
|
|
|
|
@router.delete("/{goal_id}")
|
|
def delete_goal(goal_id: str, session: dict = Depends(require_auth)):
|
|
"""Delete goal"""
|
|
pid = session['profile_id']
|
|
|
|
with get_db() as conn:
|
|
cur = get_cursor(conn)
|
|
cur.execute(
|
|
"DELETE FROM goals WHERE id = %s AND profile_id = %s",
|
|
(goal_id, pid)
|
|
)
|
|
|
|
if cur.rowcount == 0:
|
|
raise HTTPException(status_code=404, detail="Ziel nicht gefunden")
|
|
|
|
return {"message": "Ziel gelöscht"}
|
|
|
|
# ============================================================================
|
|
# Training Phases
|
|
# ============================================================================
|
|
|
|
@router.get("/phases")
|
|
def list_training_phases(session: dict = Depends(require_auth)):
|
|
"""List training phases"""
|
|
pid = session['profile_id']
|
|
|
|
with get_db() as conn:
|
|
cur = get_cursor(conn)
|
|
cur.execute("""
|
|
SELECT id, phase_type, detected_automatically, confidence_score,
|
|
status, start_date, end_date, duration_days,
|
|
detection_params, notes, created_at
|
|
FROM training_phases
|
|
WHERE profile_id = %s
|
|
ORDER BY start_date DESC
|
|
""", (pid,))
|
|
|
|
return [r2d(row) for row in cur.fetchall()]
|
|
|
|
@router.post("/phases")
|
|
def create_training_phase(data: TrainingPhaseCreate, session: dict = Depends(require_auth)):
|
|
"""Create training phase (manual)"""
|
|
pid = session['profile_id']
|
|
|
|
with get_db() as conn:
|
|
cur = get_cursor(conn)
|
|
|
|
duration = None
|
|
if data.end_date:
|
|
duration = (data.end_date - data.start_date).days
|
|
|
|
cur.execute("""
|
|
INSERT INTO training_phases (
|
|
profile_id, phase_type, detected_automatically,
|
|
status, start_date, end_date, duration_days, notes
|
|
) VALUES (%s, %s, %s, %s, %s, %s, %s, %s)
|
|
RETURNING id
|
|
""", (
|
|
pid, data.phase_type, False,
|
|
'active', data.start_date, data.end_date, duration, data.notes
|
|
))
|
|
|
|
phase_id = cur.fetchone()['id']
|
|
|
|
return {"id": phase_id, "message": "Trainingsphase erstellt"}
|
|
|
|
@router.put("/phases/{phase_id}/status")
|
|
def update_phase_status(
|
|
phase_id: str,
|
|
status: str,
|
|
session: dict = Depends(require_auth)
|
|
):
|
|
"""Update training phase status (accept/reject auto-detected phases)"""
|
|
pid = session['profile_id']
|
|
|
|
valid_statuses = ['suggested', 'accepted', 'active', 'completed', 'rejected']
|
|
if status not in valid_statuses:
|
|
raise HTTPException(
|
|
status_code=400,
|
|
detail=f"Ungültiger Status. Erlaubt: {', '.join(valid_statuses)}"
|
|
)
|
|
|
|
with get_db() as conn:
|
|
cur = get_cursor(conn)
|
|
cur.execute(
|
|
"UPDATE training_phases SET status = %s WHERE id = %s AND profile_id = %s",
|
|
(status, phase_id, pid)
|
|
)
|
|
|
|
if cur.rowcount == 0:
|
|
raise HTTPException(status_code=404, detail="Trainingsphase nicht gefunden")
|
|
|
|
return {"message": "Status aktualisiert"}
|
|
|
|
# ============================================================================
|
|
# Fitness Tests
|
|
# ============================================================================
|
|
|
|
@router.get("/tests")
|
|
def list_fitness_tests(session: dict = Depends(require_auth)):
|
|
"""List all fitness tests"""
|
|
pid = session['profile_id']
|
|
|
|
with get_db() as conn:
|
|
cur = get_cursor(conn)
|
|
cur.execute("""
|
|
SELECT id, test_type, result_value, result_unit,
|
|
test_date, test_conditions, norm_category, created_at
|
|
FROM fitness_tests
|
|
WHERE profile_id = %s
|
|
ORDER BY test_date DESC
|
|
""", (pid,))
|
|
|
|
return [r2d(row) for row in cur.fetchall()]
|
|
|
|
@router.post("/tests")
|
|
def create_fitness_test(data: FitnessTestCreate, session: dict = Depends(require_auth)):
|
|
"""Record fitness test result"""
|
|
pid = session['profile_id']
|
|
|
|
with get_db() as conn:
|
|
cur = get_cursor(conn)
|
|
|
|
# Calculate norm category (simplified for now)
|
|
norm_category = _calculate_norm_category(
|
|
data.test_type,
|
|
data.result_value,
|
|
data.result_unit
|
|
)
|
|
|
|
cur.execute("""
|
|
INSERT INTO fitness_tests (
|
|
profile_id, test_type, result_value, result_unit,
|
|
test_date, test_conditions, norm_category
|
|
) VALUES (%s, %s, %s, %s, %s, %s, %s)
|
|
RETURNING id
|
|
""", (
|
|
pid, data.test_type, data.result_value, data.result_unit,
|
|
data.test_date, data.test_conditions, norm_category
|
|
))
|
|
|
|
test_id = cur.fetchone()['id']
|
|
|
|
return {"id": test_id, "norm_category": norm_category}
|
|
|
|
# ============================================================================
|
|
# Helper Functions
|
|
# ============================================================================
|
|
|
|
def _get_current_value_for_goal_type(conn, profile_id: str, goal_type: str) -> Optional[float]:
|
|
"""Get current value for a goal type from latest data"""
|
|
cur = get_cursor(conn)
|
|
|
|
if goal_type == 'weight':
|
|
cur.execute("""
|
|
SELECT weight FROM weight_log
|
|
WHERE profile_id = %s
|
|
ORDER BY date DESC LIMIT 1
|
|
""", (profile_id,))
|
|
row = cur.fetchone()
|
|
return float(row['weight']) if row else None
|
|
|
|
elif goal_type == 'body_fat':
|
|
cur.execute("""
|
|
SELECT body_fat_pct FROM caliper_log
|
|
WHERE profile_id = %s
|
|
ORDER BY date DESC LIMIT 1
|
|
""", (profile_id,))
|
|
row = cur.fetchone()
|
|
return float(row['body_fat_pct']) if row else None
|
|
|
|
elif goal_type == 'vo2max':
|
|
cur.execute("""
|
|
SELECT vo2max FROM vitals_baseline
|
|
WHERE profile_id = %s AND vo2max IS NOT NULL
|
|
ORDER BY date DESC LIMIT 1
|
|
""", (profile_id,))
|
|
row = cur.fetchone()
|
|
return float(row['vo2max']) if row else None
|
|
|
|
elif goal_type == 'rhr':
|
|
cur.execute("""
|
|
SELECT resting_hr FROM vitals_baseline
|
|
WHERE profile_id = %s AND resting_hr IS NOT NULL
|
|
ORDER BY date DESC LIMIT 1
|
|
""", (profile_id,))
|
|
row = cur.fetchone()
|
|
return float(row['resting_hr']) if row else None
|
|
|
|
return None
|
|
|
|
def _update_goal_progress(conn, profile_id: str, goal: dict):
|
|
"""Update goal progress (modifies goal dict in-place)"""
|
|
# Get current value
|
|
current = _get_current_value_for_goal_type(conn, profile_id, goal['goal_type'])
|
|
|
|
if current is not None and goal['start_value'] is not None and goal['target_value'] is not None:
|
|
goal['current_value'] = current
|
|
|
|
# Calculate progress percentage
|
|
total_delta = float(goal['target_value']) - float(goal['start_value'])
|
|
current_delta = current - float(goal['start_value'])
|
|
|
|
if total_delta != 0:
|
|
progress_pct = (current_delta / total_delta) * 100
|
|
goal['progress_pct'] = round(progress_pct, 2)
|
|
|
|
# Simple linear projection
|
|
if goal['start_date'] and current_delta != 0:
|
|
days_elapsed = (date.today() - goal['start_date']).days
|
|
if days_elapsed > 0:
|
|
days_per_unit = days_elapsed / current_delta
|
|
remaining_units = float(goal['target_value']) - current
|
|
remaining_days = int(days_per_unit * remaining_units)
|
|
goal['projection_date'] = date.today() + timedelta(days=remaining_days)
|
|
|
|
# Check if on track
|
|
if goal['target_date'] and goal['projection_date']:
|
|
goal['on_track'] = goal['projection_date'] <= goal['target_date']
|
|
|
|
def _calculate_norm_category(test_type: str, value: float, unit: str) -> Optional[str]:
|
|
"""
|
|
Calculate norm category for fitness test
|
|
(Simplified - would need age/gender-specific norms)
|
|
"""
|
|
# Placeholder - should use proper norm tables
|
|
return None
|