Files
workout/features/person_overview.py

268 lines
10 KiB
Python

from datetime import date, timedelta
from utils import calculate_estimated_1rm, get_exercise_graph_model
class PersonOverview:
def __init__(self, db_connection_method):
self.execute = db_connection_method
def get_earliest_and_latest_workout_dates(self, person_id):
sql_query = """
SELECT
MIN(w.start_date) AS earliest_date,
MAX(w.start_date) AS latest_date
FROM workout w
INNER JOIN topset t ON w.workout_id = t.workout_id
WHERE w.person_id = %s;
"""
result = self.execute(sql_query, [person_id])
if not result or not result[0]:
return None, None
return result[0]['earliest_date'], result[0]['latest_date']
def list_of_performed_exercise_ids(self, person_id, min_date, max_date):
sql_query = """
SELECT
ARRAY_AGG(DISTINCT e.exercise_id) AS exercise_ids
FROM workout w
LEFT JOIN topset t ON w.workout_id = t.workout_id
LEFT JOIN exercise e ON t.exercise_id = e.exercise_id
WHERE w.start_date BETWEEN %s AND %s
AND w.person_id = %s
"""
result = self.execute(sql_query, [min_date, max_date, person_id])
if not result or not result[0]:
return []
return result[0]['exercise_ids']
def get_exercises_with_selection(self, person_id, start_date, end_date, selected_exercise_ids):
# SQL query to fetch all exercises performed by the person in the given time range
sql_query = """
SELECT DISTINCT
e.exercise_id,
e.name AS exercise_name
FROM
workout w
JOIN
topset t ON w.workout_id = t.workout_id
JOIN
exercise e ON t.exercise_id = e.exercise_id
WHERE
w.person_id = %s
AND w.start_date BETWEEN %s AND %s
ORDER BY
e.name ASC;
"""
# Execute the query with parameters
result = self.execute(sql_query, [person_id, start_date, end_date])
if not result:
return [] # No exercises found in the given time range
# Add the "selected" property to each exercise
exercises = []
for row in result:
exercises.append({
"id": row["exercise_id"],
"name": row["exercise_name"],
"selected": row["exercise_id"] in selected_exercise_ids
})
return exercises
def get(self, person_id, start_date, end_date, selected_exercise_ids):
# Build placeholders for exercise IDs
placeholders = ", ".join(["%s"] * len(selected_exercise_ids))
# Dynamically inject placeholders into the query
sql_query = f"""
SELECT
p.person_id,
p.name AS person_name,
w.workout_id,
w.start_date,
w.note AS workout_note,
e.exercise_id,
e.name AS exercise_name,
t.topset_id,
t.repetitions,
t.weight
FROM
person p
JOIN
workout w ON p.person_id = w.person_id
JOIN
topset t ON w.workout_id = t.workout_id
JOIN
exercise e ON t.exercise_id = e.exercise_id
WHERE
p.person_id = %s
AND w.start_date BETWEEN %s AND %s
AND e.exercise_id IN ({placeholders})
ORDER BY
w.start_date DESC, e.exercise_id ASC, t.topset_id ASC;
"""
# Add parameters for the query
params = [person_id, start_date, end_date] + selected_exercise_ids
result = self.execute(sql_query, params)
if not result:
return {"person_id": person_id, "person_name": None, "workouts": [], "selected_exercises": [], "exercise_progress_graphs": []}
# Extract person info from the first row
person_info = {"person_id": result[0]["person_id"], "person_name": result[0]["person_name"]}
# Extract and sort all unique exercises by name
exercises = []
unique_exercise_ids = set()
for row in result:
if row["exercise_id"] not in unique_exercise_ids:
unique_exercise_ids.add(row["exercise_id"])
exercises.append({"id": row["exercise_id"], "name": row["exercise_name"]})
# Sort the exercises by name
exercises = sorted(exercises, key=lambda ex: ex["name"])
# Initialize the table structure
workouts = []
workout_map = {} # Map to track workouts
# Initialize the exercise sets dictionary
exercise_sets = {exercise["id"]: {"exercise_id": exercise["id"], "name": exercise["name"], "sets": []} for exercise in exercises}
workout_start_dates = []
set_count = 0
exercise_count = len(unique_exercise_ids)
for row in result:
workout_id = row["workout_id"]
# Initialize the workout if not already present
if workout_id not in workout_map:
workout_map[workout_id] = {
"id": workout_id,
"start_date": row["start_date"],
"note": row["workout_note"],
"exercises": {exercise["id"]: [] for exercise in exercises} # Keyed by exercise_id
}
workout_start_dates.append(row["start_date"])
# Add topset to the corresponding exercise
if row["exercise_id"] and row["topset_id"]:
# Add to workout exercises
workout_map[workout_id]["exercises"][row["exercise_id"]].append({
"repetitions": row["repetitions"],
"weight": row["weight"]
})
# Add to the exercise sets dictionary with workout start date
exercise_sets[row["exercise_id"]]["sets"].append({
"repetitions": row["repetitions"],
"weight": row["weight"],
"estimated_1rm": calculate_estimated_1rm(row["weight"], row["repetitions"]),
"workout_start_date": row["start_date"],
"exercise_name": row["exercise_name"]
})
set_count += 1
# Transform into a list of rows
for workout_id, workout in workout_map.items():
workouts.append(workout)
exercise_progress_graphs = self.generate_exercise_progress_graphs(person_info["person_id"], exercise_sets)
workout_count = len(workout_start_dates)
stats = [{"Text": "Total Workouts", "Value": workout_count},
{"Text": "Total Sets", "Value": set_count},
{"Text": "Total Exercises", "Value": exercise_count}]
if workout_count > 0:
first_workout_date = min(workout_start_dates)
last_workout_date = max(workout_start_dates)
stats.append({"Text": "Days Since First Workout", "Value": (
date.today() - first_workout_date).days})
if workout_count >= 2:
stats.append({"Text": "Days Since Last Workout",
"Value": (
date.today() - last_workout_date).days})
average_number_sets_per_workout = round(
set_count / workout_count, 1)
stats.append({"Text": "Average sets per workout",
"Value": average_number_sets_per_workout})
training_duration = last_workout_date - first_workout_date
if training_duration > timedelta(days=0):
average_workouts_per_week = round(
workout_count / (training_duration.days / 7), 1)
stats.append({"Text": "Average Workouts Per Week",
"Value": average_workouts_per_week})
return {
**person_info,
"workouts": workouts,
"selected_exercises": exercises,
"exercise_progress_graphs": exercise_progress_graphs,
"stats": stats
}
def generate_exercise_progress_graphs(self, person_id, exercise_sets):
exercise_progress_graphs = []
for exercise_id, exercise_data in exercise_sets.items():
# Sort the sets by start date in descending order
sorted_exercise_sets = sorted(exercise_data["sets"], key=lambda t: t["workout_start_date"], reverse=True)
# Extract the required data
estimated_1rm = [t["estimated_1rm"] for t in sorted_exercise_sets]
repetitions = [t["repetitions"] for t in sorted_exercise_sets]
weight = [t["weight"] for t in sorted_exercise_sets]
start_dates = [t["workout_start_date"] for t in sorted_exercise_sets]
messages = [
f'{t["repetitions"]} x {t["weight"]}kg ({t["estimated_1rm"]}kg E1RM) on {t["workout_start_date"].strftime("%d %b %y")}'
for t in sorted_exercise_sets
]
epoch = "All"
exercise_name = sorted_exercise_sets[0]["exercise_name"]
# Check for valid data before generating the graph
if exercise_name and estimated_1rm and repetitions and weight and start_dates and messages:
exercise_progress = get_exercise_graph_model(
title=exercise_name,
estimated_1rm=estimated_1rm,
repetitions=repetitions,
weight=weight,
start_dates=start_dates,
messages=messages,
epoch=epoch,
person_id=person_id,
exercise_id=exercise_id,
)
# Append the generated graph model to the list
exercise_progress_graphs.append({
"exercise_id": exercise_id,
"exercise_name": exercise_name,
"progress_graph": exercise_progress
})
return exercise_progress_graphs