"""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.exercise import Exercise from app.models.progress_log import ProgressLog 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. """ all_sessions = self._session.exec( select(WorkoutSession) .where(WorkoutSession.user_id == user_id) .order_by(WorkoutSession.date.desc()) ).all() # Only count sessions that still have log entries sessions = [] total_volume = 0.0 total_sets = 0 for ws in all_sessions: logs = self._session.exec( select(WorkoutLog).where(WorkoutLog.session_id == ws.id) ).all() if not logs: continue sessions.append(ws) for log_entry in logs: total_sets += 1 weight = _weight_to_float(log_entry.weight_used) total_volume += log_entry.reps_completed * weight total_sessions = len(sessions) 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() if not logs: continue 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 def get_personal_records(self, user_id: int) -> list[dict]: """Get per-exercise max weight records for a user. Returns: List of dicts with exercise_name, weight, weight_display, date. Sorted by weight descending. BW-only exercises excluded. """ sessions = self._session.exec( select(WorkoutSession) .where(WorkoutSession.user_id == user_id) ).all() # Map exercise_id -> {max_weight, weight_str, date, exercise_name} records: dict[int, dict] = {} for ws in sessions: logs = self._session.exec( select(WorkoutLog).where(WorkoutLog.session_id == ws.id) ).all() for log_entry in logs: weight = _weight_to_float(log_entry.weight_used) if weight == 0.0: continue existing = records.get(log_entry.exercise_id) if existing is None or weight > existing["weight"]: records[log_entry.exercise_id] = { "exercise_id": log_entry.exercise_id, "weight": weight, "weight_display": log_entry.weight_used, "date": ws.date, } # Resolve exercise names result = [] for exercise_id, rec in records.items(): exercise = self._session.get(Exercise, exercise_id) if exercise: rec["exercise_name"] = exercise.name result.append(rec) result.sort(key=lambda r: r["weight"], reverse=True) return result def get_adherence_rate(self, user_id: int, weeks: int = 8) -> dict: """Calculate workout adherence rate over the past N weeks. Returns: Dict with rate (0-100), completed, expected, weeks. """ cutoff = date.today() - timedelta(weeks=weeks) sessions = self._session.exec( select(WorkoutSession) .where( WorkoutSession.user_id == user_id, WorkoutSession.date >= cutoff, ) ).all() # Only count sessions with logs completed = 0 for ws in sessions: logs = self._session.exec( select(WorkoutLog).where(WorkoutLog.session_id == ws.id) ).all() if logs: completed += 1 expected = weeks * 4 rate = round((completed / expected) * 100) if expected > 0 else 0 return { "rate": min(rate, 100), "completed": completed, "expected": expected, "weeks": weeks, } def get_muscle_group_recency(self, user_id: int) -> list[dict]: """Get the most recent workout date for each muscle group. Returns: List of dicts with muscle_group, last_worked, days_ago. Sorted by days_ago descending (most stale first). """ exercises = self._session.exec(select(Exercise)).all() muscle_groups = {e.muscle_group for e in exercises if e.muscle_group} # Map exercise_id -> muscle_group for fast lookup ex_muscle = {e.id: e.muscle_group for e in exercises} sessions = self._session.exec( select(WorkoutSession) .where(WorkoutSession.user_id == user_id) .order_by(WorkoutSession.date.desc()) ).all() recency: dict[str, date] = {} for ws in sessions: logs = self._session.exec( select(WorkoutLog).where(WorkoutLog.session_id == ws.id) ).all() for log_entry in logs: mg = ex_muscle.get(log_entry.exercise_id, "") if mg and (mg not in recency or ws.date > recency[mg]): recency[mg] = ws.date today = date.today() result = [] for mg in sorted(muscle_groups): last_worked = recency.get(mg) days_ago = (today - last_worked).days if last_worked else None result.append({ "muscle_group": mg, "last_worked": last_worked, "days_ago": days_ago, }) # Sort: never-worked first, then most stale result.sort( key=lambda r: (r["days_ago"] is None, -(r["days_ago"] or 0)), reverse=True, ) return result def get_recent_activity(self, user_id: int, limit: int = 5) -> list[dict]: """Get the last N workout sessions with summary data. Returns: List of dicts with date, workout_day_name, total_volume, total_sets. """ 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) .order_by(WorkoutSession.date.desc()) ).all() result = [] for ws in sessions: logs = self._session.exec( select(WorkoutLog).where(WorkoutLog.session_id == ws.id) ).all() if not logs: continue total_volume = sum( log_entry.reps_completed * _weight_to_float(log_entry.weight_used) for log_entry in logs ) result.append({ "date": ws.date, "workout_day_name": day_map.get(ws.workout_day_id, "Unknown"), "total_volume": round(total_volume), "total_sets": len(logs), }) if len(result) >= limit: break return result def get_progression_timeline(self, user_id: int) -> dict: """Get progression history for Chart.js multi-line chart. Returns: Dict with 'exercises' key mapping exercise names to {dates, weights, events} lists. """ logs = self._session.exec( select(ProgressLog) .where(ProgressLog.user_id == user_id) .order_by(ProgressLog.date.asc()) ).all() exercises: dict[int, list] = {} for pl in logs: exercises.setdefault(pl.exercise_id, []).append(pl) result = {} for exercise_id, entries in exercises.items(): exercise = self._session.get(Exercise, exercise_id) if not exercise: continue name = exercise.name result[name] = { "dates": [e.date.isoformat() for e in entries], "weights": [ _weight_to_float(e.actual_weight or e.suggested_weight or "0") for e in entries ], "events": [e.progression_applied or "" for e in entries], } return {"exercises": result}