feat: add Phase 5 Progression & Analytics — smart suggestions, dashboard, schedule
Add auto-progression engine (ProgressionService) with rep increase, weight increase, deload, and felt-easy acceleration rules. Add AnalyticsService for user stats, exercise progress charts, and volume-by-day data. New dashboard and schedule routes with Chart.js visualizations. Progression badges shown inline on workout day view. Navigation updated with Dashboard and Schedule links. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
173
app/services/analytics_service.py
Normal file
173
app/services/analytics_service.py
Normal file
@@ -0,0 +1,173 @@
|
||||
"""Analytics service for progress dashboard and chart data.
|
||||
|
||||
Aggregates workout log data into stats, trends, and chart-ready formats.
|
||||
"""
|
||||
|
||||
import re
|
||||
from datetime import date, timedelta
|
||||
|
||||
import structlog
|
||||
from sqlmodel import Session, select
|
||||
|
||||
from app.models.workout_day import WorkoutDay
|
||||
from app.models.workout_log import WorkoutLog
|
||||
from app.models.workout_session import WorkoutSession
|
||||
|
||||
logger = structlog.get_logger(__name__)
|
||||
|
||||
|
||||
def _weight_to_float(weight_str: str) -> float:
|
||||
"""Convert weight string to float for volume calculations.
|
||||
|
||||
Args:
|
||||
weight_str: Weight like '30 lbs' or 'BW'.
|
||||
|
||||
Returns:
|
||||
Numeric weight, or 0.0 for bodyweight.
|
||||
"""
|
||||
if not weight_str or weight_str.upper() == "BW":
|
||||
return 0.0
|
||||
match = re.search(r"(\d+(?:\.\d+)?)", weight_str)
|
||||
return float(match.group(1)) if match else 0.0
|
||||
|
||||
|
||||
class AnalyticsService:
|
||||
"""Aggregates workout data for dashboards and charts.
|
||||
|
||||
Args:
|
||||
session: An active SQLModel Session.
|
||||
"""
|
||||
|
||||
def __init__(self, session: Session) -> None:
|
||||
self._session = session
|
||||
|
||||
def get_user_stats(self, user_id: int) -> dict:
|
||||
"""Get summary statistics for a user.
|
||||
|
||||
Returns:
|
||||
Dict with keys: total_sessions, total_volume, total_sets,
|
||||
current_streak, last_workout_date.
|
||||
"""
|
||||
sessions = self._session.exec(
|
||||
select(WorkoutSession)
|
||||
.where(WorkoutSession.user_id == user_id)
|
||||
.order_by(WorkoutSession.date.desc())
|
||||
).all()
|
||||
total_sessions = len(sessions)
|
||||
|
||||
total_volume = 0.0
|
||||
total_sets = 0
|
||||
for ws in sessions:
|
||||
logs = self._session.exec(
|
||||
select(WorkoutLog).where(WorkoutLog.session_id == ws.id)
|
||||
).all()
|
||||
for log_entry in logs:
|
||||
total_sets += 1
|
||||
weight = _weight_to_float(log_entry.weight_used)
|
||||
total_volume += log_entry.reps_completed * weight
|
||||
|
||||
current_streak = 0
|
||||
if sessions:
|
||||
week_start = date.today() - timedelta(days=date.today().weekday())
|
||||
for week_offset in range(52):
|
||||
week_check = week_start - timedelta(weeks=week_offset)
|
||||
week_end = week_check + timedelta(days=6)
|
||||
has_session = any(
|
||||
week_check <= ws.date <= week_end for ws in sessions
|
||||
)
|
||||
if has_session:
|
||||
current_streak += 1
|
||||
else:
|
||||
break
|
||||
|
||||
last_workout = sessions[0].date if sessions else None
|
||||
|
||||
return {
|
||||
"total_sessions": total_sessions,
|
||||
"total_volume": round(total_volume),
|
||||
"total_sets": total_sets,
|
||||
"current_streak": current_streak,
|
||||
"last_workout_date": last_workout,
|
||||
}
|
||||
|
||||
def get_exercise_progress(
|
||||
self, user_id: int, exercise_id: int,
|
||||
) -> dict:
|
||||
"""Get chart-ready progress data for a specific exercise.
|
||||
|
||||
Returns:
|
||||
Dict with keys: dates, reps, weights, volumes.
|
||||
"""
|
||||
sessions = self._session.exec(
|
||||
select(WorkoutSession)
|
||||
.where(WorkoutSession.user_id == user_id)
|
||||
.order_by(WorkoutSession.date.asc())
|
||||
).all()
|
||||
|
||||
dates = []
|
||||
reps = []
|
||||
weights = []
|
||||
volumes = []
|
||||
|
||||
for ws in sessions:
|
||||
logs = self._session.exec(
|
||||
select(WorkoutLog).where(
|
||||
WorkoutLog.session_id == ws.id,
|
||||
WorkoutLog.exercise_id == exercise_id,
|
||||
)
|
||||
).all()
|
||||
|
||||
if not logs:
|
||||
continue
|
||||
|
||||
avg_reps = sum(
|
||||
log_entry.reps_completed for log_entry in logs
|
||||
) / len(logs)
|
||||
weight = _weight_to_float(logs[0].weight_used)
|
||||
session_volume = sum(
|
||||
log_entry.reps_completed * _weight_to_float(log_entry.weight_used)
|
||||
for log_entry in logs
|
||||
)
|
||||
|
||||
dates.append(ws.date.isoformat())
|
||||
reps.append(round(avg_reps, 1))
|
||||
weights.append(weight)
|
||||
volumes.append(round(session_volume))
|
||||
|
||||
return {
|
||||
"dates": dates,
|
||||
"reps": reps,
|
||||
"weights": weights,
|
||||
"volumes": volumes,
|
||||
}
|
||||
|
||||
def get_volume_by_day(self, user_id: int) -> dict:
|
||||
"""Get total volume broken down by workout day.
|
||||
|
||||
Returns:
|
||||
Dict mapping workout day name to total volume.
|
||||
"""
|
||||
days = self._session.exec(select(WorkoutDay)).all()
|
||||
day_map = {d.id: d.name for d in days}
|
||||
|
||||
sessions = self._session.exec(
|
||||
select(WorkoutSession)
|
||||
.where(WorkoutSession.user_id == user_id)
|
||||
).all()
|
||||
|
||||
volume_by_day = {}
|
||||
for ws in sessions:
|
||||
day_name = day_map.get(ws.workout_day_id, "Unknown")
|
||||
logs = self._session.exec(
|
||||
select(WorkoutLog).where(WorkoutLog.session_id == ws.id)
|
||||
).all()
|
||||
|
||||
day_volume = sum(
|
||||
log_entry.reps_completed * _weight_to_float(log_entry.weight_used)
|
||||
for log_entry in logs
|
||||
)
|
||||
volume_by_day[day_name] = (
|
||||
volume_by_day.get(day_name, 0) + round(day_volume)
|
||||
)
|
||||
|
||||
return volume_by_day
|
||||
271
app/services/progression_service.py
Normal file
271
app/services/progression_service.py
Normal file
@@ -0,0 +1,271 @@
|
||||
"""Auto-progression engine for workout programming.
|
||||
|
||||
Analyzes workout log history and applies the progression model:
|
||||
- +1-2 reps/week until wk4 rep target
|
||||
- +5 lbs every 2 weeks once at rep target
|
||||
- Deload at week 5 (-20% weight, reset to wk1 reps)
|
||||
- Accelerated weight increase when all sets felt easy
|
||||
"""
|
||||
|
||||
import re
|
||||
from datetime import date
|
||||
from typing import Optional
|
||||
|
||||
import structlog
|
||||
from sqlmodel import Session, select
|
||||
|
||||
from app.models.progress_log import ProgressLog
|
||||
from app.models.user_exercise_program import UserExerciseProgram
|
||||
from app.models.workout_log import WorkoutLog
|
||||
from app.models.workout_session import WorkoutSession
|
||||
|
||||
logger = structlog.get_logger(__name__)
|
||||
|
||||
|
||||
def _parse_weight(weight_str: str) -> Optional[float]:
|
||||
"""Extract numeric weight from a string like '30 lbs' or 'BW'.
|
||||
|
||||
Args:
|
||||
weight_str: Weight as a string.
|
||||
|
||||
Returns:
|
||||
Numeric weight in lbs, or None for bodyweight.
|
||||
"""
|
||||
if not weight_str or weight_str.upper() == "BW":
|
||||
return None
|
||||
match = re.search(r"(\d+(?:\.\d+)?)", weight_str)
|
||||
return float(match.group(1)) if match else None
|
||||
|
||||
|
||||
def _format_weight(weight_lbs: Optional[float]) -> str:
|
||||
"""Format a numeric weight back to a display string.
|
||||
|
||||
Args:
|
||||
weight_lbs: Weight in lbs, or None for bodyweight.
|
||||
|
||||
Returns:
|
||||
Formatted string like '35 lbs' or 'BW'.
|
||||
"""
|
||||
if weight_lbs is None:
|
||||
return "BW"
|
||||
if weight_lbs == int(weight_lbs):
|
||||
return f"{int(weight_lbs)} lbs"
|
||||
return f"{weight_lbs:.1f} lbs"
|
||||
|
||||
|
||||
class ProgressionService:
|
||||
"""Implements the auto-progression engine.
|
||||
|
||||
Args:
|
||||
session: An active SQLModel Session.
|
||||
"""
|
||||
|
||||
def __init__(self, session: Session) -> None:
|
||||
self._session = session
|
||||
|
||||
def _get_program(
|
||||
self, user_id: int, exercise_id: int,
|
||||
) -> Optional[UserExerciseProgram]:
|
||||
"""Look up the user's program for a specific exercise."""
|
||||
statement = select(UserExerciseProgram).where(
|
||||
UserExerciseProgram.user_id == user_id,
|
||||
UserExerciseProgram.exercise_id == exercise_id,
|
||||
)
|
||||
return self._session.exec(statement).first()
|
||||
|
||||
def _get_recent_sessions(
|
||||
self, user_id: int, exercise_id: int, limit: int = 5,
|
||||
) -> list[dict]:
|
||||
"""Get recent session summaries for an exercise.
|
||||
|
||||
Returns a list of dicts with: date, avg_reps, weight, all_felt_easy.
|
||||
"""
|
||||
statement = (
|
||||
select(WorkoutSession)
|
||||
.where(WorkoutSession.user_id == user_id)
|
||||
.order_by(WorkoutSession.date.desc())
|
||||
.limit(limit * 2)
|
||||
)
|
||||
sessions = self._session.exec(statement).all()
|
||||
|
||||
results = []
|
||||
for ws in sessions:
|
||||
logs = self._session.exec(
|
||||
select(WorkoutLog).where(
|
||||
WorkoutLog.session_id == ws.id,
|
||||
WorkoutLog.exercise_id == exercise_id,
|
||||
)
|
||||
).all()
|
||||
|
||||
if not logs:
|
||||
continue
|
||||
|
||||
avg_reps = sum(log.reps_completed for log in logs) / len(logs)
|
||||
weight = logs[0].weight_used
|
||||
all_felt_easy = all(log.felt_easy for log in logs)
|
||||
|
||||
results.append({
|
||||
"date": ws.date,
|
||||
"avg_reps": avg_reps,
|
||||
"weight": weight,
|
||||
"all_felt_easy": all_felt_easy,
|
||||
"set_count": len(logs),
|
||||
})
|
||||
|
||||
if len(results) >= limit:
|
||||
break
|
||||
|
||||
return results
|
||||
|
||||
def get_suggestion(
|
||||
self, user_id: int, exercise_id: int,
|
||||
) -> dict:
|
||||
"""Generate a progression suggestion for the next workout.
|
||||
|
||||
Analyzes recent log history against the user's program targets
|
||||
and applies progression rules.
|
||||
|
||||
Returns:
|
||||
Dict with keys: suggested_reps, suggested_weight,
|
||||
progression_type, message.
|
||||
"""
|
||||
program = self._get_program(user_id, exercise_id)
|
||||
|
||||
if program is None:
|
||||
return {
|
||||
"suggested_reps": 0,
|
||||
"suggested_weight": "",
|
||||
"progression_type": "no_program",
|
||||
"message": "No program found for this exercise.",
|
||||
}
|
||||
|
||||
try:
|
||||
wk1_reps = int(program.wk1_reps)
|
||||
wk4_reps = int(program.wk4_reps)
|
||||
except (ValueError, TypeError):
|
||||
wk1_reps = 0
|
||||
wk4_reps = 0
|
||||
|
||||
wk1_weight = program.wk1_weight
|
||||
|
||||
recent = self._get_recent_sessions(user_id, exercise_id, limit=5)
|
||||
|
||||
if not recent:
|
||||
return {
|
||||
"suggested_reps": wk1_reps,
|
||||
"suggested_weight": wk1_weight,
|
||||
"progression_type": "baseline",
|
||||
"message": f"Start with {wk1_reps} reps @ {wk1_weight}.",
|
||||
}
|
||||
|
||||
latest = recent[0]
|
||||
current_reps = int(round(latest["avg_reps"]))
|
||||
current_weight = latest["weight"]
|
||||
current_weight_num = _parse_weight(current_weight)
|
||||
consecutive_sessions = len(recent)
|
||||
|
||||
# Rule: Deload at week 5 (4 consecutive sessions completed)
|
||||
if consecutive_sessions >= 4:
|
||||
if current_weight_num is not None:
|
||||
deload_weight = current_weight_num * 0.8
|
||||
return {
|
||||
"suggested_reps": wk1_reps,
|
||||
"suggested_weight": _format_weight(deload_weight),
|
||||
"progression_type": "deload",
|
||||
"message": (
|
||||
f"Deload week: {wk1_reps} reps @ "
|
||||
f"{_format_weight(deload_weight)} (-20%)."
|
||||
),
|
||||
}
|
||||
return {
|
||||
"suggested_reps": wk1_reps,
|
||||
"suggested_weight": current_weight,
|
||||
"progression_type": "deload",
|
||||
"message": f"Deload week: reset to {wk1_reps} reps.",
|
||||
}
|
||||
|
||||
# Rule: Weight increase if at rep target and felt easy
|
||||
if current_reps >= wk4_reps and latest["all_felt_easy"]:
|
||||
if current_weight_num is not None:
|
||||
new_weight = current_weight_num + 5
|
||||
return {
|
||||
"suggested_reps": wk1_reps,
|
||||
"suggested_weight": _format_weight(new_weight),
|
||||
"progression_type": "weight_increase",
|
||||
"message": (
|
||||
f"Weight up: {wk1_reps} reps @ "
|
||||
f"{_format_weight(new_weight)} (+5 lbs)."
|
||||
),
|
||||
}
|
||||
|
||||
# Rule: Weight increase after 2 weeks at rep target
|
||||
if (
|
||||
current_reps >= wk4_reps
|
||||
and len(recent) >= 2
|
||||
and int(round(recent[1]["avg_reps"])) >= wk4_reps
|
||||
):
|
||||
if current_weight_num is not None:
|
||||
new_weight = current_weight_num + 5
|
||||
return {
|
||||
"suggested_reps": wk1_reps,
|
||||
"suggested_weight": _format_weight(new_weight),
|
||||
"progression_type": "weight_increase",
|
||||
"message": (
|
||||
f"2 weeks at target: {wk1_reps} reps @ "
|
||||
f"{_format_weight(new_weight)} (+5 lbs)."
|
||||
),
|
||||
}
|
||||
|
||||
# Rule: Rep increase (+1-2 reps)
|
||||
if current_reps < wk4_reps:
|
||||
increment = 2 if latest["all_felt_easy"] else 1
|
||||
new_reps = min(current_reps + increment, wk4_reps)
|
||||
return {
|
||||
"suggested_reps": new_reps,
|
||||
"suggested_weight": current_weight,
|
||||
"progression_type": "reps_increase",
|
||||
"message": (
|
||||
f"Reps up: {new_reps} reps @ {current_weight} "
|
||||
f"(+{increment})."
|
||||
),
|
||||
}
|
||||
|
||||
# Hold: at target, waiting for biweekly weight increase
|
||||
return {
|
||||
"suggested_reps": current_reps,
|
||||
"suggested_weight": current_weight,
|
||||
"progression_type": "hold",
|
||||
"message": f"Hold at {current_reps} reps @ {current_weight}.",
|
||||
}
|
||||
|
||||
def record_progression(
|
||||
self,
|
||||
user_id: int,
|
||||
exercise_id: int,
|
||||
suggested_reps: int,
|
||||
suggested_weight: str,
|
||||
actual_reps: int,
|
||||
actual_weight: str,
|
||||
progression_type: str,
|
||||
) -> ProgressLog:
|
||||
"""Record a progression entry in the progress_log table."""
|
||||
progress_log = ProgressLog(
|
||||
user_id=user_id,
|
||||
exercise_id=exercise_id,
|
||||
date=date.today(),
|
||||
suggested_reps=suggested_reps,
|
||||
suggested_weight=suggested_weight,
|
||||
actual_reps=actual_reps,
|
||||
actual_weight=actual_weight,
|
||||
progression_applied=progression_type,
|
||||
)
|
||||
self._session.add(progress_log)
|
||||
self._session.commit()
|
||||
self._session.refresh(progress_log)
|
||||
logger.info(
|
||||
"progression_recorded",
|
||||
user_id=user_id,
|
||||
exercise_id=exercise_id,
|
||||
type=progression_type,
|
||||
)
|
||||
return progress_log
|
||||
Reference in New Issue
Block a user