How I Added Functionality To Download Workouts CSV From The Peloton API To My Workouts Dashboard

Photo by bruce mars on Unsplash

How I Added Functionality To Download Workouts CSV From The Peloton API To My Workouts Dashboard

Using Requests to Download Peloton CSV, Waiting as a locking mechanism and load_dotenv for environment variables

This is a follow-up to my previous blog post How to build your own Peloton workout dashboard.

Previously, I created a dashboard for my Peloton workouts using Plotly Dash. However, there was a necessary manual step of downloading your workouts in that project that I wanted to automate. In this blog post, I will walk through how I:

  1. Extended the PylotonCycle API to download the workouts CSV.

  2. Allowed users to specify their login credentials with a .env file.

  3. Made my Dash app wait to download the CSV before launching

The code for this project currently lives in my GitHub repo.

Extending the PylotonCycle to download the workouts CSV

Problem

In its current state, the PylotonCycle API allows for getting recent workouts, getting a list of n workouts and getting a single workout in JSON format. However, Dash requires loading data into a Pandas Dataframe and rather than constructing the CSV myself I wondered whether it was possible to get grab the data with one call to the API using the same method I would use to download it manually through my Peloton workouts tab.

Indeed, hovering over the workouts button revealed to me that I can get the workouts CSV by accessing the endpoint using an authenticated user.

Solution

PylotonCycle provides a requests session to an authenticated user if you pass in login credentials to the class. Once we are authenticated, we can construct the endpoint to download the CSV using the base URL and User ID provided by the class.

"""
This is just a snippet.
Full code available here: https://github.com/justmedude/pylotoncycle/blob/main/pylotoncycle/pylotoncycle.py
"""

class PylotonCycle:
    def __init__(self, username, password):
        """
        This will be the base url we will use.
        """
        self.base_url = 'https://api.onepeloton.com'
        """
        This will be our session.
        """
        self.s = requests.Session()
        self.headers = {
            'Content-Type': 'application/json',
            'User-Agent': 'pylotoncycle'
        }

        # Initialize a couple of variables that will get reused
        # userid - our userid
        # instructor_id_dict - dictionary that will allow us to cache
        #                      information
        #                      format is: instructor_id : instructor_dict
        self.userid = None
        self.instructor_id_dict = {}

        """
        Once this login method is called in the constructor, we will have
        access to the userid class variable shown above.
        """
        self.login(username, password)

    def login(self, username, password):
        auth_login_url = '%s/auth/login' % self.base_url
        auth_payload = {
            'username_or_email': username,
            'password': password
        }
        headers = {
            'Content-Type': 'application/json',
            'User-Agent': 'pyloton'
        }
        resp = self.s.post(
            auth_login_url,
            json=auth_payload, headers=headers, timeout=10).json()

        if (('status' in resp) and (resp['status'] == 401)):
            raise PelotonLoginException(resp['message'] if ('message' in resp)
                  else "Login Failed")

        self.userid = resp['user_id']

    def GetMe(self):
        url = '%s/api/me' % self.base_url
        resp = self.s.get(url, timeout=10).json()
        self.username = resp['username']
        self.userid = resp['id']
        self.total_workouts = resp['total_workouts']
        return resp

Using the class above, we can write a small script that constructs the necessary URL endpoint we need for downloading the CSV. This will look like workouts_url = f"{connection.base_url}/api/user/{connection.userid}/workout_history_csv?timezone={timezone}" where connection is an instance of the PylotonCycle class and timezone is a string parameter passed in to specify which timezone we want the workout timestamps to be in. Afterward, we can make a get request using the authenticated requests session to the URL. We should also specify the path where we would like to save this CSV and have a default timezone in case nothing is passed in.

#!./venv/bin/python
import os
import csv
from pathlib import Path, PurePath

from pylotoncycle import PylotonCycle


def GetWorkoutCSV(
    connection: PylotonCycle, path: str = "", timezone: str = "America/New_York"
) -> dict:
    workouts_url = f"{connection.base_url}/api/user/{connection.userid}/workout_history_csv?timezone={timezone}"
    resp = connection.s.get(workouts_url, headers=connection.headers, timeout=10)
    p = Path(path).joinpath("workouts.csv") if path else PurePath("workouts.csv")
    writer = csv.writer(open(p, "w"))
    for row in csv.reader(resp.text.splitlines()):
        writer.writerow(row)
    print(f"Workouts saved to {r['path']}")
    return {"response": resp, "path": p}


if __name__ == "__main__":
    username = os.environ["USERNAME"]
    password = os.environ["PASSWORD"]

    conn = PylotonCycle(username, password)
    r = GetWorkoutCSV(connection=conn, timezone="America/New_York")

At a high level, the code above:

  1. Instantiates the PylotonCycle class with the USERNAME and PASSWORD environment variables (we will talk about these later!)

  2. Passes the PylotonCycle object into the GetWorkoutsCSV function.

  3. The GetWorkoutsCSV function constructs the workouts CSV download API endpoint, makes a get request to the endpoint and writes the response text to a file called workouts.csv. If there is no path passed in, this file is written to the directory from which the script is called.

We now have our function for creating our CSV! Whenever this function is called, it will overwrite the previous workouts.csv.

Loading login credentials from .env file

Problem

The script above requires login credentials for the user. However, we do not want to add these to the file itself as we may accidentally commit it. We would like to store these in a file that can be ignored by git by adding it to .gitignore and still be able to use the credentials in the script.

Solution

We can use the load-dotenv package to load environment variables from our .env file. Then, we will be able to use these variables using the os.environ dictionary. This requires us to run pip install load-dotenv to install the load-dotenv dependency and add the following two lines to our script.

from dotenv import load_dotenv

load_dotenv()

Now our finished CSV download script looks like the one below.

#!./venv/bin/python
import os

from dotenv import load_dotenv

load_dotenv()
import csv
from pathlib import Path, PurePath

from pylotoncycle import PylotonCycle


def GetWorkoutCSV(
    connection: PylotonCycle, path: str = "", timezone: str = "America/New_York"
) -> dict:
    workouts_url = f"{connection.base_url}/api/user/{connection.userid}/workout_history_csv?timezone={timezone}"
    resp = connection.s.get(workouts_url, headers=connection.headers, timeout=10)
    p = Path(path).joinpath("workouts.csv") if path else PurePath("workouts.csv")
    writer = csv.writer(open(p, "w"))
    for row in csv.reader(resp.text.splitlines()):
        writer.writerow(row)
    print(f"Workouts saved to {r['path']}")
    return {"response": resp, "path": p}


if __name__ == "__main__":
    username = os.environ["USERNAME"]
    password = os.environ["PASSWORD"]

    conn = PylotonCycle(username, password)
    r = GetWorkoutCSV(connection=conn, timezone="America/New_York")

We can now create our .env file and to store our credentials and have them be used by our script.

USERNAME = YOUR_PELOTON_USERNAME_OR_EMAIL
PASSWORD = YOUR_PELOTON_PASSWORD

Making Dash app wait until CSV is downloaded

Problem

Now that the script for downloading the CSV of workouts is completed, we would like to call this from within Dash. However, as soon as I tried calling the code I realized that the call to GetWorkoutCSV is non-blocking and the pd.read_csv() method is called before the workouts.csv file exists. I needed a way to have this function wait until the workouts.csv exists (if it doesn't already exist).

Solution

I discovered the waiting module which serves this purpose. It waits for a function to return True before it continues onto the rest of the code. We can use this to poll whether the workouts.csv file exists and if it does not, it will wait up until a timeout of 10 seconds for the lambda function to return True while the file is written. Since I plan to later use the get_workouts_csv.py script for a standalone cron job, I decided to execute it as a shell script in the code snippet below.

from waiting import wait
import os

os.system("python get_workout_csv.py")


def is_something_ready():
    return bool(os.path.exists("workouts.csv"))


wait(
    lambda: is_something_ready(),
    timeout_seconds=10,
    waiting_for="workouts.csv to be ready",
)
df = pd.read_csv("workouts.csv")

That is it! Now the Dash app will wait for the workouts.csv file to download and I no longer have to manually have users of my Dash app download their Peloton workouts CSV. Instead, they simply provide credentials in the .env file and it will download workouts before the app launches.

Final Thoughts

I am excited to have resurrected this project from the 2021 dustbin and to have been able to add a new feature. I learned about how the Peloton API returns the CSV data in the response text (I don't know why I thought it would return the file object) and I learned about the waiting module which seems very useful. However, I now realize the front end needs some work! Upon adding this feature I came up with many more ideas for future features to add to this dashboard and am excited to keep implementing them.

I hope you learned something from this article. If you did (or if you are also a fan of working out with Peloton), please let me know in the comments section below!