Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
import BaseHTTPServer
import urlparse
import webbrowser
import pandas as pd
import datetime
#Global Variables - put your data in the file 'client.secret' and separate the fields with a comma!
client_id, secret = open('client.secret').read().strip().split(',')
port = 5000
url = 'http://localhost:%d/authorized' % port
allDone = False
types = ['time', 'distance', 'latlng', 'altitude', 'velocity_smooth', 'moving', 'grade_smooth', 'temp']
limit = 10
#Create the strava client, and open the web browser for authentication
client = stravalib.client.Client()
authorize_url = client.authorization_url(client_id=client_id, redirect_uri=url)
print 'Opening: %s' % authorize_url
webbrowser.open(authorize_url)
#Define the web functions to call from the strava API
def UseCode(code):
#Retrieve the login code from the Strava server
access_token = client.exchange_code_for_token(client_id=client_id,
client_secret=secret, code=code)
# Now store that access token somewhere (for now, it's just a local variable)
client.access_token = access_token
athlete = client.get_athlete()
print("For %(id)s, I now have an access token %(token)s" %
{'id': athlete.id, 'token': access_token})
return client
def strava_authorized():
code = request.args.get('code') # or whatever your framework does
client = stravalib.client.Client()
access_token = client.exchange_code_for_token(client_id=app.config['STRAVA_CLIENT_ID'], client_secret=app.config['STRAVA_CLIENT_SECRET'], code=code)
current_user.preferences['strava'] = UserPreference('strava', access_token)
db.session.commit()
flash("Access to Strava API granted")
return redirect(url_for('move_import'))
flash("imported '%s': move %d" % (xmlfile.filename, move_id.id))
return redirect(url_for('move', id=move_id.id))
else:
flash("imported %d moves" % len(imported_moves))
return redirect(url_for('moves'))
else:
model = {'has_strava': current_user.has_strava()}
if current_user.has_strava():
associated_activities, known_activities, new_activities = strava.associate_activities(current_user)
model['new_strava_activities'] = new_activities
model['associated_strava_activities'] = associated_activities
model['known_strava_activities'] = known_activities
elif 'STRAVA_CLIENT_ID' in app.config:
client = stravalib.client.Client()
client_id_ = app.config['STRAVA_CLIENT_ID']
strava_authorize_url = client.authorization_url(client_id=client_id_,
redirect_uri=url_for('strava_authorized', _external=True),
scope='activity:read_all')
model['strava_authorize_url'] = strava_authorize_url
return render_template('import.html', **model)
Columns names are translated to sweat terminology (e.g. "heart_rate" > "heartrate").
Two API calls are made to the Strava API: 1 to retrieve activity metadata, 1 to retrieve the raw data ("streams").
Args:
activity_id: The id of the activity
access_token: The Strava access token
refresh_token: The Strava refresh token. Optional.
client_id: The Strava client id. Optional. Used for token refresh.
client_secret: The Strava client secret. Optional. Used for token refresh.
resample: whether or not the data frame needs to be resampled to 1Hz
interpolate: whether or not missing data in the data frame needs to be interpolated
Returns:
A pandas data frame with all the data.
"""
client = Client()
client.access_token = access_token
client.refresh_token = refresh_token
activity = client.get_activity(activity_id)
start_datetime = activity.start_date_local
streams = client.get_activity_streams(
activity_id=activity_id, types=STREAM_TYPES, series_type="time",
)
raw_data = dict()
for key, value in streams.items():
if key == "latlng":
latitude, longitude = list(zip(*value.data))
raw_data["latitude"] = latitude
raw_data["longitude"] = longitude
def get_strava_api(secret, ID):
all_act = []
client = Client(access_token=secret)
tot = total_num(client)
me = client.get_athlete(ID)
activities = client.get_activities()
for i in trange(tot):
df = pd.DataFrame()
_a = activities.next()
_streams = client.get_activity_streams(_a.id, types=types)
for item in types:
if item in _streams.keys():
df[item] = pd.Series(_streams[item].data, index=None)
df['act_id'] = _a.id
df['act_name'] = _a.name
df['act_type'] = _a.type
def strava_upload(player_id, activity):
try:
from stravalib.client import Client
except ImportError:
logger.warn("stravalib is not installed. Skipping Strava upload attempt.")
return
profile_dir = '%s/%s' % (STORAGE_DIR, player_id)
strava = Client()
try:
with open('%s/strava_token.txt' % profile_dir, 'r') as f:
client_id = f.readline().rstrip('\r\n')
client_secret = f.readline().rstrip('\r\n')
strava.access_token = f.readline().rstrip('\r\n')
refresh_token = f.readline().rstrip('\r\n')
expires_at = f.readline().rstrip('\r\n')
except:
logger.warn("Failed to read %s/strava_token.txt. Skipping Strava upload attempt." % profile_dir)
return
try:
if time.time() > int(expires_at):
refresh_response = strava.refresh_access_token(client_id=client_id, client_secret=client_secret,
refresh_token=refresh_token)
with open('%s/strava_token.txt' % profile_dir, 'w') as f:
f.write(client_id + '\n')
def get_strava_client(current_user):
strava_access_token = current_user.get_strava_access_token()
client = stravalib.client.Client(access_token=strava_access_token)
return client