cardio/app.py

from sqlalchemy import func
from datetime import timedelta
import io
from flask_sqlalchemy import SQLAlchemy
from flask import Flask, make_response, render_template, request, jsonify
import jinja_partials
from flask_htmx import HTMX
import matplotlib.pyplot as plt
import matplotlib.dates as mdates
import os


app = Flask(__name__)
# TODO CHANGE SECRET KEY TO ENVIRONMENT VARIABLE
app.config['SECRET_KEY'] = 'secret!'
uri = os.getenv("DATABASE_URL")
if uri and uri.startswith("postgres://"):
    uri = uri.replace("postgres://", "postgresql://", 1)

app.config['SQLALCHEMY_DATABASE_URI'] = uri
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False
jinja_partials.register_extensions(app)
htmx = HTMX(app)
db = SQLAlchemy(app)


class User(db.Model):
    __tablename__ = 'users'
    id = db.Column(db.Integer, primary_key=True)
    name = db.Column(db.String(255), nullable=False)
    workouts = db.relationship('Workout', backref='user', lazy=True)


class Workout(db.Model):
    __tablename__ = 'workouts'
    id = db.Column(db.Integer, primary_key=True)
    user_id = db.Column(db.Integer, db.ForeignKey(
        'users.id', ondelete='CASCADE'), nullable=False)
    created_at = db.Column(db.DateTime, nullable=False, default=db.func.now())
    cadence_readings = db.relationship(
        'CadenceReading', backref='workout', lazy=True)


class CadenceReading(db.Model):
    __tablename__ = 'cadence_readings'
    id = db.Column(db.Integer, primary_key=True)
    workout_id = db.Column(db.Integer, db.ForeignKey(
        'workouts.id', ondelete='CASCADE'), nullable=False)
    created_at = db.Column(db.DateTime, nullable=False)
    rpm = db.Column(db.Integer, nullable=False)


@app.route('/', methods=['GET'])
def get_workouts():
    users = User.query.all()
    users_data = []
    for user in users:
        workouts = Workout.query.filter_by(user_id=user.id).all()
        user_data = {
            'id': user.id,
            'name': user.name,
            'workouts_count': len(workouts),
            'workouts': get_workouts_for_user(user.id)
        }
        users_data.append(user_data)
    return render_template('users_and_workouts.html', users=users_data)


@app.route("/user/<int:user_id>/new_workout")
def new_workout(user_id):
    user = User.query.get(user_id)
    return render_template('new_workout.html', user=user)


@app.route('/users', methods=['GET', 'POST'])
def users():
    if request.method == 'GET':
        # get a list of all users in the database
        users = User.query.all()
        users_list = [{'id': user.id, 'name': user.name,
                       'workouts': len(user.workouts)} for user in users]
        return jsonify(users_list), 200

    elif request.method == 'POST':
        # create a new user
        data = request.json
        name = data['name']

        # create a new user and add it to the database
        new_user = User(name=name)
        db.session.add(new_user)
        db.session.commit()

        return jsonify({'message': 'User created successfully.'}), 201


@app.route('/user/<int:user_id>', methods=['DELETE'])
def delete_user(user_id):
    user = User.query.get(user_id)
    if user:
        db.session.delete(user)
        db.session.commit()
        return jsonify({'message': 'User deleted successfully.'}), 200
    else:
        return jsonify({'error': 'User not found.'}), 404


@app.route('/user/<int:user_id>/workouts', methods=['GET', 'POST'])
def workouts(user_id):
    if request.method == 'GET':
        workouts_data = get_workouts_for_user(user_id)
        return render_template('workouts_list.html', workouts=workouts_data)

    elif request.method == 'POST':
        data = request.json
        rpm_readings = data['workout']

        # create a new workout
        workout = Workout(user_id=user_id)
        db.session.add(workout)
        db.session.commit()

        # add cadence readings to the workout
        for reading in rpm_readings:
            cadence_reading = CadenceReading(
                workout_id=workout.id, created_at=reading['timestamp'], rpm=reading['rpm'])
            db.session.add(cadence_reading)

        db.session.commit()

        return jsonify({'message': 'Workout created successfully.'}), 201


@app.route('/user/<int:user_id>/workout/<int:workout_id>', methods=['GET', 'DELETE'])
def workout(user_id, workout_id):
    workout = Workout.query.filter_by(user_id=user_id, id=workout_id).first()
    if workout:
        if request.method == 'GET':
            # Get the cadence readings for the workout
            cadence_readings = CadenceReading.query.filter_by(
                workout_id=workout_id).all()
            if cadence_readings:
                # Create a graph of cadence readings
                x_values = [reading.created_at for reading in cadence_readings]
                y_values = [reading.rpm for reading in cadence_readings]
                fig, ax = plt.subplots()
                ax.plot(x_values, y_values)
                ax.set_xlabel('Time')
                ax.set_ylabel('Cadence (RPM)')

                # ax.set_title(
                #    'Cadence Readings for Workout {}'.format(workout_id))
                ax.xaxis.set_major_formatter(mdates.DateFormatter("%H:%M"))

                # set the y-axis limits to start at 0
                ax.set_ylim(bottom=0)

                # Save the graph to a bytes buffer
                buffer = io.BytesIO()
                plt.savefig(buffer, format='png',
                            transparent=True, bbox_inches='tight')
                buffer.seek(0)
                # Create a response object with the graph image
                response = make_response(buffer.getvalue())
                response.headers['Content-Type'] = 'image/png'
                response.headers['Content-Disposition'] = 'attachment; filename=cadence.png'
                return response, 200
            else:
                return jsonify({'message': 'No cadence readings for workout {}.'.format(workout_id)}), 404
        elif request.method == 'DELETE':
            # Delete the workout and its associated cadence readings
            CadenceReading.query.filter_by(workout_id=workout_id).delete()
            db.session.delete(workout)
            db.session.commit()
            workouts_data = get_workouts_for_user(user_id)
            return render_template('workouts_list.html', workouts=workouts_data)
    else:
        return jsonify({'message': 'Workout {} not found for user {}.'.format(workout_id, user_id)}), 404


@app.route('/user/<int:user_id>/workouts', methods=['GET', 'DELETE'])
def workouts_for_user(user_id):
    workouts_data = get_workouts_for_user(user_id)
    return render_template('workouts_list.html', workouts=workouts_data)


def get_workouts_for_user(user_id):
    workouts_data = []
    workouts = Workout.query.filter_by(user_id=user_id).order_by(
        Workout.created_at.desc()).all()
    for workout in workouts:
        cadence_readings = CadenceReading.query.filter_by(
            workout_id=workout.id).all()
        if cadence_readings:
            start_time = min(
                reading.created_at for reading in cadence_readings)
            end_time = max(
                reading.created_at for reading in cadence_readings)
            duration = end_time - start_time
            average_rpm = sum(
                reading.rpm for reading in cadence_readings) / len(cadence_readings)
            workouts_data.append({
                'id': workout.id,
                'user_id': user_id,
                'start_time': start_time.strftime('%Y-%m-%d %H:%M:%S'),
                'duration': format_duration(duration),
                'average_rpm': int(average_rpm)
            })
    return workouts_data


def format_duration(duration):
    hours, remainder = divmod(duration.seconds, 3600)
    minutes, _ = divmod(remainder, 60)

    if duration >= timedelta(hours=1):
        return f"{hours}h {minutes}m"
    else:
        return f"{minutes}m"


if __name__ == '__main__':
    # Bind to PORT if defined, otherwise default to 5000.
    port = int(os.environ.get('PORT', 5000))
    app.run(host='127.0.0.1', port=port)