feat: enhance dashboard with PRs, adherence, activity, progression chart, and muscle heatmap
All checks were successful
Build and Push Docker Image / build-and-push (push) Successful in 13s
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>
This commit is contained in:
@@ -9,6 +9,8 @@ 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
|
||||
@@ -179,3 +181,197 @@ class AnalyticsService:
|
||||
)
|
||||
|
||||
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}
|
||||
|
||||
Reference in New Issue
Block a user