Skip to content

Instantly share code, notes, and snippets.

@al-codaio
Last active January 21, 2021 17:42
Show Gist options
  • Save al-codaio/b9cc9fe33a46cb014eef22f95cd4d459 to your computer and use it in GitHub Desktop.
Save al-codaio/b9cc9fe33a46cb014eef22f95cd4d459 to your computer and use it in GitHub Desktop.
Python script to sync Peloton workout data to a table in a Coda doc
# One-way data sync from Peloton API to Coda in Python
# Author: Al Chen ([email protected])
# Last Updated: January 21st, 2021
# Writeup and copyable template here: https://coda.io/@atc/analyze-your-peloton-workout-stats-with-real-time-updates
################# Setup and global variables ###########
api_key = 'YOUR_CODA_API_KEY'
coda_doc_id = 'YOUR_CODA_DOC_ID'
peloton_username = 'YOUR_PELOTON_USERNAME'
peloton_pw = 'YOUR_PELOTON_PASSWORD'
########################################################
import requests
import json
# Current workout IDs
table_name = 'Workouts'
headers = {'Authorization': 'Bearer ' + api_key}
current_workout_ids = []
current_rows_uri = 'https://coda.io/apis/v1/docs/' + coda_doc_id + '/tables/' + table_name + '/rows'
current_rows = requests.get(current_rows_uri, headers=headers, params={})
for workout_id in current_rows.json()['items']:
current_workout_ids.append(workout_id['name'])
# Get workouts from Peloton API
s = requests.Session()
base_url = 'https://api.onepeloton.com'
payload = {'username_or_email': peloton_username, 'password': peloton_pw}
s.post(base_url + '/auth/login', json=payload)
userID = s.get(base_url + '/api/me').json()['id']
workouts = json.loads(json.dumps(s.get(base_url + '/api/user/' + userID + '/workouts?limit=1000').json()).replace('null', '""'))
workout_data = {}
for workout in workouts['data']:
workout_id = workout['id']
workout_data[workout_id] = {}
workout_summary = json.loads(json.dumps(s.get(base_url + '/api/workout/' + workout_id).json()).replace('null', '""'))
workout_performance = json.loads(json.dumps(s.get(base_url + '/api/workout/' + workout_id + '/performance_graph?every_n=1000').json()).replace('null', '""'))
workout_data[workout_id]['summary'] = workout_summary
workout_data[workout_id]['performance'] = workout_performance
# Get net new workouts
for workout_id in current_workout_ids:
if workout_id in workout_data:
workout_data.pop(workout_id)
# Push new workouts to Coda
rows = []
for key, values in workout_data.items():
cells = []
avg_summary_length = len(values['performance']['average_summaries'])
if avg_summary_length > 0: # for bike workouts
cells.extend((
{'column': 'Workout ID', 'value': key},
{'column': 'Date', 'value': values['summary']['created_at']},
{'column': 'Workout Type', 'value': values['summary']['fitness_discipline']},
{'column': 'Difficulty', 'value': values['summary']['ride']['difficulty_estimate']},
{'column': 'Duration', 'value': values['summary']['ride']['duration']},
{'column': 'Class Thumbnail', 'value': values['summary']['ride']['image_url']},
{'column': 'Instructor ID', 'value': values['summary']['ride']['instructor_id']},
{'column': 'Total Ratings', 'value': values['summary']['ride']['overall_rating_count']},
{'column': 'Workout Name', 'value': values['summary']['ride']['title']},
{'column': 'Total Workouts', 'value': values['summary']['ride']['total_workouts']},
{'column': 'Start Time', 'value': values['summary']['start_time']},
{'column': 'Leaderboard Rank', 'value': values['summary']['leaderboard_rank']},
{'column': 'Leaderboard Users', 'value': values['summary']['total_leaderboard_users']},
{'column': 'Status', 'value': values['summary']['status']},
{'column': 'Workout Description', 'value': values['summary']['ride']['description']},
{'column': 'Avg Output (kj)', 'value': values['performance']['average_summaries'][0]['value'] if 0 < avg_summary_length else 0},
{'column': 'Avg Cadence', 'value': values['performance']['average_summaries'][1]['value'] if 1 < avg_summary_length else 0},
{'column': 'Avg Resistance', 'value': values['performance']['average_summaries'][2]['value'] if 2 < avg_summary_length else 0},
{'column': 'Avg Speed (mph)', 'value': values['performance']['average_summaries'][3]['value'] if 3 < avg_summary_length else 0},
{'column': 'Total Output (kj)', 'value': values['performance']['summaries'][0]['value']},
{'column': 'Distance (mi)', 'value': values['performance']['summaries'][1]['value']},
{'column': 'Calories (kcal)', 'value': values['performance']['summaries'][2]['value']}
))
else: # for non-bike workouts
cells.extend((
{'column': 'Workout ID', 'value': key},
{'column': 'Date', 'value': values['summary']['created_at']},
{'column': 'Workout Type', 'value': values['summary']['fitness_discipline']},
{'column': 'Difficulty', 'value': values['summary']['ride']['difficulty_estimate']},
{'column': 'Duration', 'value': values['summary']['ride']['duration']},
{'column': 'Class Thumbnail', 'value': values['summary']['ride']['image_url']},
{'column': 'Instructor ID', 'value': values['summary']['ride']['instructor_id']},
{'column': 'Total Ratings', 'value': values['summary']['ride']['overall_rating_count']},
{'column': 'Workout Name', 'value': values['summary']['ride']['title']},
{'column': 'Total Workouts', 'value': values['summary']['ride']['total_workouts']},
{'column': 'Start Time', 'value': values['summary']['start_time']},
{'column': 'Leaderboard Rank', 'value': values['summary']['leaderboard_rank']},
{'column': 'Leaderboard Users', 'value': values['summary']['total_leaderboard_users']},
{'column': 'Status', 'value': values['summary']['status']},
{'column': 'Workout Description', 'value': values['summary']['ride']['description']}
))
rows.append({'cells': cells})
new_rows_uri = 'https://coda.io/apis/v1/docs/' + coda_doc_id + '/tables/' + table_name+ '/rows'
payload = {'rows': rows}
new_rows = requests.post(new_rows_uri, headers=headers, json=payload)
print(new_rows.json())
# Current instructors
table_name_instructors = 'Instructors'
current_instructor_ids = []
current_instructors_uri = 'https://coda.io/apis/v1/docs/' + coda_doc_id + '/tables/' + table_name_instructors + '/rows'
current_instructors = requests.get(current_instructors_uri, headers=headers, params={})
for instructor_id in current_instructors.json()['items']:
current_instructor_ids.append(instructor_id['name'])
# Get instructors from API
instructors = json.loads(json.dumps(s.get(base_url + '/api/instructor?limit=100').json()).replace('null', '""'))['data']
# Get new instructors
new_instructors = []
for instructor in instructors:
if instructor['id'] not in current_instructor_ids:
new_instructors.append(instructor)
# Push new instructors to Coda
rows = []
for instructor in new_instructors:
cells = []
cells.extend((
{'column': 'Instructor ID', 'value': instructor['id']},
{'column': 'Image URL', 'value': instructor['image_url']},
{'column': 'Background', 'value': instructor['bio']},
{'column': 'Name', 'value': instructor['name']},
{'column': 'Instagram', 'value': instructor['instagram_profile']},
{'column': 'Twitter', 'value': instructor['twitter_profile']},
{'column': 'Quote', 'value': instructor['quote']},
{'column': 'Spotify Playlist', 'value': instructor['spotify_playlist_uri']},
{'column': 'Peloton Username', 'value': instructor['username']}
))
rows.append({'cells': cells})
new_instructor_rows_uri = 'https://coda.io/apis/v1/docs/' + coda_doc_id + '/tables/' + table_name_instructors + '/rows'
payload = {'rows': rows}
new_instructor_rows = requests.post(new_instructor_rows_uri, headers=headers, json=payload)
print(new_instructor_rows.json())
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment