-
Notifications
You must be signed in to change notification settings - Fork 0
/
peloton_coda_python.py
139 lines (126 loc) · 7.54 KB
/
peloton_coda_python.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
# One-way data sync from Peloton API to Coda in Python
# Author: Al Chen (al@coda.io)
# 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())