233 lines
8.6 KiB
Python
233 lines
8.6 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}
|
|
|
|
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
|
|
}
|
|
|
|
# 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"]
|
|
})
|
|
|
|
# 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)
|
|
|
|
return {
|
|
**person_info,
|
|
"workouts": workouts,
|
|
"selected_exercises": exercises,
|
|
"exercise_progress_graphs": exercise_progress_graphs
|
|
}
|
|
|
|
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
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|