"""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