Phillip Tarrant
df8d5c65fb
All checks were successful
Build and Push Docker Image / build-and-push (push) Successful in 13s
Add 3 new stat cards (Last Workout, Personal Records, Adherence Rate), recent activity table, progression timeline chart, and muscle group recency heatmap to the dashboard. Remove Total Volume card. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
378 lines
12 KiB
Python
378 lines
12 KiB
Python
"""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}
|