Add person graphs endpoint for workouts per week & PRs per week, consumed via dashboard, person overview and notes

app.py

@@ -2,12 +2,11 @@ from datetime import datetime, date, timedelta
 from dateutil.relativedelta import relativedelta
 import os
 from flask import Flask, abort, render_template, redirect, request, url_for
-from jinja2 import Environment, FileSystemLoader, select_autoescape
 import jinja_partials
 from jinja2_fragments import render_block
 from decorators import validate_person, validate_topset, validate_workout
 from db import DataBase
-from utils import count_prs_over_time, get_people_and_exercise_rep_maxes, convert_str_to_date, get_earliest_and_latest_workout_date, filter_workout_topsets, first_and_last_visible_days_in_month, get_weekly_pr_graph_model, get_workout_counts, generate_plot
+from utils import get_people_and_exercise_rep_maxes, convert_str_to_date, get_earliest_and_latest_workout_date, filter_workout_topsets, first_and_last_visible_days_in_month, generate_plot
 from flask_htmx import HTMX
 import minify_html
 from urllib.parse import quote
@@ -66,13 +65,9 @@ def dashboard():
     people_and_exercise_rep_maxes = get_people_and_exercise_rep_maxes(
         all_topsets, selected_person_ids, selected_exercise_ids, min_date, max_date)
 
-    weekly_counts = get_workout_counts(all_topsets, 'week')
-    weekly_pr_counts = count_prs_over_time(all_topsets, 'week')
-    dashboard_graphs = [get_weekly_pr_graph_model('Workouts per week', weekly_counts), get_weekly_pr_graph_model('PRs per week', weekly_pr_counts)]
-
     if htmx:
-        return render_block(app.jinja_env, 'dashboard.html', 'content', model=people_and_exercise_rep_maxes, people=people, exercises=exercises, min_date=min_date, max_date=max_date, selected_person_ids=selected_person_ids, selected_exercise_ids=selected_exercise_ids, tags=tags, dashboard_graphs=dashboard_graphs)
+        return render_block(app.jinja_env, 'dashboard.html', 'content', model=people_and_exercise_rep_maxes, people=people, exercises=exercises, min_date=min_date, max_date=max_date, selected_person_ids=selected_person_ids, selected_exercise_ids=selected_exercise_ids, tags=tags)
-    return render_template('dashboard.html', model=people_and_exercise_rep_maxes, people=people, exercises=exercises, min_date=min_date, max_date=max_date, selected_person_ids=selected_person_ids, selected_exercise_ids=selected_exercise_ids, tags=tags, dashboard_graphs=dashboard_graphs)
+    return render_template('dashboard.html', model=people_and_exercise_rep_maxes, people=people, exercises=exercises, min_date=min_date, max_date=max_date, selected_person_ids=selected_person_ids, selected_exercise_ids=selected_exercise_ids, tags=tags)
 
 
 @ app.route("/person/list", methods=['GET'])
@@ -457,6 +452,18 @@ def get_stats():
     stats = db.stats.fetch_stats(selected_people_ids, min_date, max_date, selected_exercise_ids)
     return render_template('partials/stats.html', stats=stats, refresh_url=request.full_path)
 
+@app.route("/graphs", methods=['GET'])
+def get_people_graphs():
+    selected_people_ids = request.args.getlist('person_id', type=int)
+    min_date = request.args.get('min_date', type=convert_str_to_date)
+    max_date = request.args.get('max_date', type=convert_str_to_date)
+    selected_exercise_ids = request.args.getlist('exercise_id', type=int)
+
+    graphs = db.people_graphs.get(selected_people_ids, min_date, max_date, selected_exercise_ids)
+
+    return render_template('partials/people_graphs.html', graphs=graphs, refresh_url=request.full_path)
+
+
 @ app.route("/person/<int:person_id>/workout/<int:workout_id>", methods=['GET'])
 def show_workout(person_id, workout_id):
     view_model = db.workout.get(person_id, workout_id)

db.py

@@ -1,6 +1,5 @@
 import os
 import psycopg2
-import numpy as np
 from psycopg2.extras import RealDictCursor
 from datetime import datetime
 from dateutil.relativedelta import relativedelta
@@ -9,11 +8,12 @@ from flask import g
 import pandas as pd
 from features.calendar import Calendar
 from features.exercises import Exercises
+from features.people_graphs import PeopleGraphs
 from features.person_overview import PersonOverview
 from features.stats import Stats
 from features.workout import Workout
 from features.sql_explorer import SQLExplorer
-from utils import count_prs_over_time, get_all_exercises_from_topsets, get_exercise_graph_model, get_topsets_for_person, get_weekly_pr_graph_model, get_workout_counts, get_workouts
+from utils import get_all_exercises_from_topsets, get_exercise_graph_model, get_topsets_for_person, get_workouts
 
 
 class DataBase():
@@ -24,6 +24,7 @@ class DataBase():
         self.exercises = Exercises(self.execute)
         self.sql_explorer = SQLExplorer(self.execute)
         self.person_overview = PersonOverview(self.execute)
+        self.people_graphs = PeopleGraphs(self.execute)
         db_url = urlparse(os.environ['DATABASE_URL'])
         # if db_url is null then throw error
         if not db_url:
@@ -207,17 +208,12 @@ class DataBase():
                 LEFT JOIN Exercise E ON T.exercise_id=E.exercise_id
             WHERE P.person_id=%s""", [person_id])
         
-        weekly_counts = get_workout_counts(topsets, 'week')
-        weekly_pr_counts = count_prs_over_time(topsets, 'week')
-        person_graphs = [get_weekly_pr_graph_model('Workouts per week', weekly_counts), get_weekly_pr_graph_model('PRs per week', weekly_pr_counts)]
-
         return {
             'PersonId': next((t['PersonId'] for t in topsets), -1),
             'PersonName': next((t['PersonName'] for t in topsets), 'Unknown'),
             'Exercises': get_all_exercises_from_topsets(topsets),
             'Workouts': get_workouts(topsets),
-            'ExerciseProgressGraphs': get_topsets_for_person(topsets),
+            'ExerciseProgressGraphs': get_topsets_for_person(topsets)
-            'PersonGraphs': person_graphs
         }
 
     def get_workout(self, person_id, workout_id):

features/people_graphs.py

@@ -0,0 +1,184 @@
+import pandas as pd
+from utils import get_distinct_colors
+
+
+class PeopleGraphs:
+    def __init__(self, db_connection_method):
+        self.execute = db_connection_method
+
+    def get(self, selected_people_ids=None, min_date=None, max_date=None, selected_exercise_ids=None):
+        # Base query
+        query = """
+        SELECT
+            P.person_id AS "PersonId",
+            P.name AS "PersonName",
+            W.workout_id AS "WorkoutId",
+            W.start_date AS "StartDate",
+            T.topset_id AS "TopSetId",
+            E.exercise_id AS "ExerciseId",
+            E.name AS "ExerciseName",
+            T.repetitions AS "Repetitions",
+            T.weight AS "Weight",
+            round((100 * T.Weight::numeric::integer)/(101.3-2.67123 * T.Repetitions),0)::numeric::integer AS "Estimated1RM"
+        FROM Person P
+            LEFT JOIN Workout W ON P.person_id = W.person_id
+            LEFT JOIN TopSet T ON W.workout_id = T.workout_id
+            LEFT JOIN Exercise E ON T.exercise_id = E.exercise_id
+        WHERE TRUE
+        """
+
+        # Parameters for the query
+        params = []
+
+        # Add optional filters
+        if selected_people_ids:
+            placeholders = ", ".join(["%s"] * len(selected_people_ids))
+            query += f" AND P.person_id IN ({placeholders})"
+            params.extend(selected_people_ids)
+        if min_date:
+            query += " AND W.start_date >= %s"
+            params.append(min_date)
+        if max_date:
+            query += " AND W.start_date <= %s"
+            params.append(max_date)
+        if selected_exercise_ids:
+            placeholders = ", ".join(["%s"] * len(selected_exercise_ids))
+            query += f" AND E.exercise_id IN ({placeholders})"
+            params.extend(selected_exercise_ids)
+
+            # Execute the query
+        topsets = self.execute(query, params)
+
+        # Generate graphs
+        weekly_counts = self.get_workout_counts(topsets, 'week')
+        weekly_pr_counts = self.count_prs_over_time(topsets, 'week')
+
+        graphs = [self.get_weekly_pr_graph_model('Workouts per week', weekly_counts), self.get_weekly_pr_graph_model('PRs per week', weekly_pr_counts)]
+        return graphs
+
+    def get_weekly_pr_graph_model(self, title, weekly_pr_data):
+        # Assuming weekly_pr_data is in the format {1: {"PersonName": "Alice", "PRCounts": {Timestamp('2022-01-01', freq='W-MON'): 0, ...}}, 2: {...}, ...}
+
+        # Find the overall date range for all users
+        all_dates = [date for user_data in weekly_pr_data.values() for date in user_data["PRCounts"].keys()]
+        min_date, max_date = min(all_dates), max(all_dates)
+        total_span = (max_date - min_date).days or 1
+        relative_positions = [(date - min_date).days / total_span for date in all_dates]
+
+        # Calculate viewBox dimensions
+        max_value = max(max(user_data["PRCounts"].values()) for user_data in weekly_pr_data.values()) or 1
+        min_value = 0
+        value_range = max_value - min_value
+        vb_width = 200
+        vb_height= 75
+
+        plots = []
+        colors = get_distinct_colors(len(weekly_pr_data.items()))
+        for count, (user_id, user_data) in enumerate(weekly_pr_data.items()):
+            pr_counts = user_data["PRCounts"]
+            person_name = user_data["PersonName"]
+
+            values = pr_counts.values()
+
+            values_scaled = [((value - min_value) / value_range) * vb_height for value in values]
+            plot_points = list(zip(values_scaled, relative_positions))
+            messages = [f'{value} for {person_name} at {date.strftime("%d %b %y")}' for value, date in zip(values, pr_counts.keys())]
+            plot_labels = zip(values_scaled, relative_positions, messages)
+
+            # Create a plot for each user
+            plot = {
+                'label': person_name,  # Use PersonName instead of User ID
+                'color': colors[count],
+                'points': plot_points,
+                'plot_labels': plot_labels
+            }
+            plots.append(plot)
+
+        # Return workout data with SVG dimensions and data points
+        return {
+            'title': title,
+            'vb_width': vb_width,
+            'vb_height': vb_height,
+            'plots': plots
+        }
+    
+    def get_workout_counts(self, workouts, period='week'):
+        df = pd.DataFrame(workouts)
+        
+        # Convert 'StartDate' to datetime and set period
+        df['StartDate'] = pd.to_datetime(df['StartDate'])
+        df['Period'] = df['StartDate'].dt.to_period('W' if period == 'week' else 'M')
+
+        # Group by PersonId, Period and count unique workouts
+        workout_counts = df.groupby(['PersonId', 'Period'])['WorkoutId'].nunique().reset_index()
+
+        # Convert 'Period' to timestamp using the start date of the period
+        workout_counts['Period'] = workout_counts['Period'].apply(lambda x: x.start_time)
+
+        # Pivot the result to get periods as columns
+        workout_counts_pivot = workout_counts.pivot(index='PersonId', columns='Period', values='WorkoutId').fillna(0)
+
+        # Include person names
+        names = df[['PersonId', 'PersonName']].drop_duplicates().set_index('PersonId')
+        workout_counts_final = names.join(workout_counts_pivot, how='left').fillna(0)
+
+        # Convert DataFrame to dictionary
+        result = workout_counts_final.reset_index().to_dict('records')
+
+        # Reformat the dictionary to desired structure
+        formatted_result = {}
+        for record in result:
+            person_id = record.pop('PersonId')
+            person_name = record.pop('PersonName')
+            pr_counts = {k: v for k, v in record.items()}
+            formatted_result[person_id] = {'PersonName': person_name, 'PRCounts': pr_counts}
+
+        return formatted_result
+
+    def count_prs_over_time(self, workouts, period='week'):
+        df = pd.DataFrame(workouts)
+        
+        # Convert 'StartDate' to datetime
+        df['StartDate'] = pd.to_datetime(df['StartDate'])
+
+        # Set period as week or month
+        df['Period'] = df['StartDate'].dt.to_period('W' if period == 'week' else 'M')
+
+        # Group by Person, Exercise, and Period to find max Estimated1RM in each period
+        period_max = df.groupby(['PersonId', 'ExerciseId', 'Period'])['Estimated1RM'].max().reset_index()
+
+        # Determine all-time max Estimated1RM up to the start of each period
+        period_max['AllTimeMax'] = period_max.groupby(['PersonId', 'ExerciseId'])['Estimated1RM'].cummax().shift(1)
+
+        # Identify PRs as entries where the period's max Estimated1RM exceeds the all-time max
+        period_max['IsPR'] = period_max['Estimated1RM'] > period_max['AllTimeMax']
+
+        # Count PRs in each period for each person
+        pr_counts = period_max.groupby(['PersonId', 'Period'])['IsPR'].sum().reset_index()
+
+        # Convert 'Period' to timestamp using the start date of the period
+        pr_counts['Period'] = pr_counts['Period'].apply(lambda x: x.start_time)
+
+        # Pivot table to get the desired output format
+        output = pr_counts.pivot(index='PersonId', columns='Period', values='IsPR').fillna(0)
+
+        # Convert only the PR count columns to integers
+        for col in output.columns:
+            output[col] = output[col].astype(int)
+
+        # Merge with names and convert to desired format
+        names = df[['PersonId', 'PersonName']].drop_duplicates().set_index('PersonId')
+        output = names.join(output, how='left').fillna(0)
+
+        # Reset the index to bring 'PersonId' back as a column
+        output.reset_index(inplace=True)
+
+        # Convert to the final dictionary format with PRCounts nested
+        result = {}
+        for index, row in output.iterrows():
+            person_id = row['PersonId']
+            person_name = row['PersonName']
+            pr_counts = row.drop(['PersonId', 'PersonName']).to_dict()
+            result[person_id] = {"PersonName": person_name, "PRCounts": pr_counts}
+
+        return result

templates/dashboard.html

@@ -2,12 +2,9 @@
 
 {% block content %}
 
-<div class="w-full mb-4 grid grid-cols-1 xl:grid-cols-2 gap-4">
+<div class="hidden" hx-get="{{ url_for('get_people_graphs') }}"
-    {% for graph in dashboard_graphs %}
+    hx-include="[name='exercise_id'],[name='min_date'],[name='max_date'],[name='person_id']" hx-trigger="load"
-    <div class="bg-white shadow rounded-lg p-4 sm:p-6 xl:p-8">
+    hx-target="this" hx-swap="outerHTML">
-        {{ render_partial('partials/svg_line_graph.html', **graph) }}
-    </div>
-    {% endfor %}
 </div>
 
 <div class="bg-white shadow rounded-lg pt-4 p-3 md:p-4 w-full mb-4">

templates/notes.html

@@ -3,6 +3,10 @@
 {% block content %}
 
 <div class="flex flex-grow flex-col bg-white sm:rounded shadow overflow-hidden">
+    <div class="hidden" hx-get="{{ url_for('get_people_graphs') }}" hx-vals='{"person_id": "{{ person_id }}"}'
+        hx-trigger="load" hx-target="this" hx-swap="outerHTML">
+    </div>
+
     <div class="flex items-center justify-between pt-2 pb-2">
         <div class="flex">
             <h2 class="ml-2 text-xl font-bold leading-none">Notes</h2>

templates/partials/people_graphs.html

@@ -0,0 +1,8 @@
+<div class="w-full mb-4 grid grid-cols-1 xl:grid-cols-2 gap-4" hx-get="{{ refresh_url }}" hx-target="this"
+    hx-swap="outerHTML" hx-trigger="refreshStats from:body">
+    {% for graph in graphs %}
+    <div class="bg-white shadow rounded-lg p-4 sm:p-6 xl:p-8">
+        {{ render_partial('partials/svg_line_graph.html', **graph) }}
+    </div>
+    {% endfor %}
+</div>

templates/partials/svg_line_graph.html

@@ -65,7 +65,7 @@
             {% endfor %}
 
         </svg>
-        {% if plots|length > 1  %} 
+        
             <div class="flex justify-center pt-2">
                 {% for plot in plots %}
                 <div class="flex items-center px-2 select-none cursor-pointer" 
@@ -76,6 +76,5 @@
                 </div>
             {% endfor %}
             </div>
-        {% endif %}
 
     </div>

templates/person_overview.html

@@ -5,6 +5,11 @@
 <div class="flex max-w-full overflow-x-hidden">
     <div class="bg-white shadow rounded-lg pt-2 pb-2 sm:w-full xl:p-8 md:w-full">
 
+        <div class="hidden" hx-get="{{ url_for('get_people_graphs') }}"
+            hx-include="[name='exercise_id'],[name='min_date'],[name='max_date']"
+            hx-vals='{"person_id": "{{ person_id }}"}' hx-trigger="load" hx-target="this" hx-swap="outerHTML">
+        </div>
+
         <div class="mb-4 flex items-center justify-between px-2 md:px-3">
             <div>
                 <h3 class="text-xl font-bold text-gray-900 mb-2">{{ person_name }}</h3>

utils.py

@@ -260,134 +260,6 @@ def get_exercise_graph_model(title, estimated_1rm, repetitions, weight, start_da
         'max_date': max_date
     }
 
-
-def get_workout_counts(workouts, period='week'):
-    df = pd.DataFrame(workouts)
-    
-    # Convert 'StartDate' to datetime and set period
-    df['StartDate'] = pd.to_datetime(df['StartDate'])
-    df['Period'] = df['StartDate'].dt.to_period('W' if period == 'week' else 'M')
-
-    # Group by PersonId, Period and count unique workouts
-    workout_counts = df.groupby(['PersonId', 'Period'])['WorkoutId'].nunique().reset_index()
-
-    # Convert 'Period' to timestamp using the start date of the period
-    workout_counts['Period'] = workout_counts['Period'].apply(lambda x: x.start_time)
-
-    # Pivot the result to get periods as columns
-    workout_counts_pivot = workout_counts.pivot(index='PersonId', columns='Period', values='WorkoutId').fillna(0)
-
-    # Include person names
-    names = df[['PersonId', 'PersonName']].drop_duplicates().set_index('PersonId')
-    workout_counts_final = names.join(workout_counts_pivot, how='left').fillna(0)
-
-    # Convert DataFrame to dictionary
-    result = workout_counts_final.reset_index().to_dict('records')
-
-    # Reformat the dictionary to desired structure
-    formatted_result = {}
-    for record in result:
-        person_id = record.pop('PersonId')
-        person_name = record.pop('PersonName')
-        pr_counts = {k: v for k, v in record.items()}
-        formatted_result[person_id] = {'PersonName': person_name, 'PRCounts': pr_counts}
-
-    return formatted_result
-
-def count_prs_over_time(workouts, period='week'):
-    df = pd.DataFrame(workouts)
-    
-    # Convert 'StartDate' to datetime
-    df['StartDate'] = pd.to_datetime(df['StartDate'])
-
-    # Set period as week or month
-    df['Period'] = df['StartDate'].dt.to_period('W' if period == 'week' else 'M')
-
-    # Group by Person, Exercise, and Period to find max Estimated1RM in each period
-    period_max = df.groupby(['PersonId', 'ExerciseId', 'Period'])['Estimated1RM'].max().reset_index()
-
-    # Determine all-time max Estimated1RM up to the start of each period
-    period_max['AllTimeMax'] = period_max.groupby(['PersonId', 'ExerciseId'])['Estimated1RM'].cummax().shift(1)
-
-    # Identify PRs as entries where the period's max Estimated1RM exceeds the all-time max
-    period_max['IsPR'] = period_max['Estimated1RM'] > period_max['AllTimeMax']
-
-    # Count PRs in each period for each person
-    pr_counts = period_max.groupby(['PersonId', 'Period'])['IsPR'].sum().reset_index()
-
-    # Convert 'Period' to timestamp using the start date of the period
-    pr_counts['Period'] = pr_counts['Period'].apply(lambda x: x.start_time)
-
-    # Pivot table to get the desired output format
-    output = pr_counts.pivot(index='PersonId', columns='Period', values='IsPR').fillna(0)
-
-    # Convert only the PR count columns to integers
-    for col in output.columns:
-        output[col] = output[col].astype(int)
-
-    # Merge with names and convert to desired format
-    names = df[['PersonId', 'PersonName']].drop_duplicates().set_index('PersonId')
-    output = names.join(output, how='left').fillna(0)
-
-    # Reset the index to bring 'PersonId' back as a column
-    output.reset_index(inplace=True)
-
-    # Convert to the final dictionary format with PRCounts nested
-    result = {}
-    for index, row in output.iterrows():
-        person_id = row['PersonId']
-        person_name = row['PersonName']
-        pr_counts = row.drop(['PersonId', 'PersonName']).to_dict()
-        result[person_id] = {"PersonName": person_name, "PRCounts": pr_counts}
-
-    return result
-
-def get_weekly_pr_graph_model(title, weekly_pr_data):
-    # Assuming weekly_pr_data is in the format {1: {"PersonName": "Alice", "PRCounts": {Timestamp('2022-01-01', freq='W-MON'): 0, ...}}, 2: {...}, ...}
-
-    # Find the overall date range for all users
-    all_dates = [date for user_data in weekly_pr_data.values() for date in user_data["PRCounts"].keys()]
-    min_date, max_date = min(all_dates), max(all_dates)
-    total_span = (max_date - min_date).days or 1
-    relative_positions = [(date - min_date).days / total_span for date in all_dates]
-
-    # Calculate viewBox dimensions
-    max_value = max(max(user_data["PRCounts"].values()) for user_data in weekly_pr_data.values()) or 1
-    min_value = 0
-    value_range = max_value - min_value
-    vb_width = 200
-    vb_height= 75
-
-    plots = []
-    colors = get_distinct_colors(len(weekly_pr_data.items()))
-    for count, (user_id, user_data) in enumerate(weekly_pr_data.items()):
-        pr_counts = user_data["PRCounts"]
-        person_name = user_data["PersonName"]
-
-        values = pr_counts.values()
-
-        values_scaled = [((value - min_value) / value_range) * vb_height for value in values]
-        plot_points = list(zip(values_scaled, relative_positions))
-        messages = [f'{value} for {person_name} at {date.strftime("%d %b %y")}' for value, date in zip(values, pr_counts.keys())]
-        plot_labels = zip(values_scaled, relative_positions, messages)
-
-        # Create a plot for each user
-        plot = {
-            'label': person_name,  # Use PersonName instead of User ID
-            'color': colors[count],
-            'points': plot_points,
-            'plot_labels': plot_labels
-        }
-        plots.append(plot)
-
-    # Return workout data with SVG dimensions and data points
-    return {
-        'title': title,
-        'vb_width': vb_width,
-        'vb_height': vb_height,
-        'plots': plots
-    }
-
 def get_distinct_colors(n):
     colors = []
     for i in range(n):