266 lines
7.5 KiB
Python
266 lines
7.5 KiB
Python
# Python
|
|
import oauth2 as oauth
|
|
import cgi
|
|
import requests
|
|
import requests.auth
|
|
import json
|
|
from django.utils import timezone
|
|
from datetime import datetime
|
|
import numpy as np
|
|
from dateutil import parser
|
|
import time
|
|
import math
|
|
from math import sin,cos,atan2,sqrt
|
|
|
|
# Django
|
|
from django.shortcuts import render_to_response
|
|
from django.http import HttpResponseRedirect, HttpResponse,JsonResponse
|
|
from django.conf import settings
|
|
from django.contrib.auth import authenticate, login, logout
|
|
from django.contrib.auth.models import User
|
|
from django.contrib.auth.decorators import login_required
|
|
|
|
# Project
|
|
# from .models import Profile
|
|
from rowingdata import rowingdata
|
|
import pandas as pd
|
|
from rowers.models import Rower,Workout
|
|
|
|
import stravalib
|
|
|
|
from rowsandall_app.settings import C2_CLIENT_ID, C2_REDIRECT_URI, C2_CLIENT_SECRET, STRAVA_CLIENT_ID, STRAVA_REDIRECT_URI, STRAVA_CLIENT_SECRET
|
|
|
|
def ewmovingaverage(interval,window_size):
|
|
# Experimental code using Exponential Weighted moving average
|
|
|
|
intervaldf = pd.DataFrame({'v':interval})
|
|
idf_ewma1 = intervaldf.ewm(span=window_size)
|
|
idf_ewma2 = intervaldf[::-1].ewm(span=window_size)
|
|
|
|
i_ewma1 = idf_ewma1.mean().ix[:,'v']
|
|
i_ewma2 = idf_ewma2.mean().ix[:,'v']
|
|
|
|
interval2 = np.vstack((i_ewma1,i_ewma2[::-1]))
|
|
interval2 = np.mean( interval2, axis=0) # average
|
|
|
|
return interval2
|
|
|
|
def geo_distance(lat1,lon1,lat2,lon2):
|
|
""" Approximate distance and bearing between two points
|
|
defined by lat1,lon1 and lat2,lon2
|
|
This is a slight underestimate but is close enough for our purposes,
|
|
We're never moving more than 10 meters between trackpoints
|
|
|
|
Bearing calculation fails if one of the points is a pole.
|
|
|
|
"""
|
|
|
|
# radius of earth in km
|
|
R = 6373.0
|
|
|
|
# pi
|
|
pi = math.pi
|
|
|
|
lat1 = math.radians(lat1)
|
|
lat2 = math.radians(lat2)
|
|
lon1 = math.radians(lon1)
|
|
lon2 = math.radians(lon2)
|
|
|
|
dlon = lon2 - lon1
|
|
dlat = lat2 - lat1
|
|
|
|
a = sin(dlat / 2)**2 + cos(lat1) * cos(lat2) * sin(dlon / 2)**2
|
|
c = 2 * atan2(sqrt(a), sqrt(1 - a))
|
|
|
|
distance = R * c
|
|
|
|
tc1 = atan2(sin(lon2-lon1)*cos(lat2),
|
|
cos(lat1)*sin(lat2)-sin(lat1)*cos(lat2)*cos(lon2-lon1))
|
|
|
|
tc1 = tc1 % (2*pi)
|
|
|
|
bearing = math.degrees(tc1)
|
|
|
|
return [distance,bearing]
|
|
|
|
|
|
def custom_exception_handler(exc,message):
|
|
|
|
response = {
|
|
"errors": [
|
|
{
|
|
"code": str(exc),
|
|
"detail": message,
|
|
}
|
|
]
|
|
}
|
|
|
|
res = HttpResponse(message)
|
|
res.status_code = 401
|
|
res.json = json.dumps(response)
|
|
|
|
return res
|
|
|
|
def get_token(code):
|
|
client_auth = requests.auth.HTTPBasicAuth(STRAVA_CLIENT_ID, STRAVA_CLIENT_SECRET)
|
|
post_data = {"grant_type": "authorization_code",
|
|
"code": code,
|
|
"redirect_uri": STRAVA_REDIRECT_URI,
|
|
"client_secret": STRAVA_CLIENT_SECRET,
|
|
"client_id":STRAVA_CLIENT_ID,
|
|
}
|
|
headers = {'user-agent': 'sanderroosendaal'}
|
|
response = requests.post("https://www.strava.com/oauth/token",
|
|
data=post_data,
|
|
headers=headers)
|
|
token_json = response.json()
|
|
thetoken = token_json['access_token']
|
|
|
|
return [thetoken]
|
|
|
|
|
|
def make_authorization_url(request):
|
|
# Generate a random string for the state parameter
|
|
# Save it for use later to prevent xsrf attacks
|
|
from uuid import uuid4
|
|
state = str(uuid4())
|
|
|
|
params = {"client_id": STRAVA_CLIENT_ID,
|
|
"response_type": "code",
|
|
"redirect_uri": STRAVA_REDIRECT_URI,
|
|
"scope":"write"}
|
|
import urllib
|
|
url = "https://www.strava.com/oauth/authorize" +urllib.urlencode(params)
|
|
|
|
return HttpResponseRedirect(url)
|
|
|
|
def get_strava_workout_list(user):
|
|
r = Rower.objects.get(user=user)
|
|
if (r.stravatoken == '') or (r.stravatoken is None):
|
|
s = "Token doesn't exist. Need to authorize"
|
|
return custom_exception_handler(401,s)
|
|
else:
|
|
# ready to fetch. Hurray
|
|
authorizationstring = str('Bearer ' + r.stravatoken)
|
|
headers = {'Authorization': authorizationstring,
|
|
'user-agent': 'sanderroosendaal',
|
|
'Content-Type': 'application/json'}
|
|
url = "https://www.strava.com/api/v3/athlete/activities"
|
|
s = requests.get(url,headers=headers)
|
|
|
|
return s
|
|
|
|
def get_strava_workout(user,stravaid):
|
|
r = Rower.objects.get(user=user)
|
|
if (r.stravatoken == '') or (r.stravatoken is None):
|
|
s = "Token doesn't exist. Need to authorize"
|
|
return custom_exception_handler(401,s)
|
|
else:
|
|
# ready to fetch. Hurray
|
|
fetchresolution = 'high'
|
|
authorizationstring = str('Bearer ' + r.stravatoken)
|
|
headers = {'Authorization': authorizationstring,
|
|
'user-agent': 'sanderroosendaal',
|
|
'Content-Type': 'application/json',
|
|
'resolution': 'medium',}
|
|
url = "https://www.strava.com/api/v3/activities/"+str(stravaid)
|
|
workoutsummary = requests.get(url,headers=headers).json()
|
|
|
|
workoutsummary['timezone'] = "Etc/UTC"
|
|
startdatetime = workoutsummary['start_date']
|
|
|
|
url = "https://www.strava.com/api/v3/activities/"+str(stravaid)+"/streams/cadence?resolution="+fetchresolution
|
|
spmjson = requests.get(url,headers=headers)
|
|
url = "https://www.strava.com/api/v3/activities/"+str(stravaid)+"/streams/heartrate?resolution="+fetchresolution
|
|
hrjson = requests.get(url,headers=headers)
|
|
url = "https://www.strava.com/api/v3/activities/"+str(stravaid)+"/streams/time?resolution="+fetchresolution
|
|
timejson = requests.get(url,headers=headers)
|
|
url = "https://www.strava.com/api/v3/activities/"+str(stravaid)+"/streams/velocity_smooth?resolution="+fetchresolution
|
|
velojson = requests.get(url,headers=headers)
|
|
url = "https://www.strava.com/api/v3/activities/"+str(stravaid)+"/streams/distance?resolution="+fetchresolution
|
|
distancejson = requests.get(url,headers=headers)
|
|
url = "https://www.strava.com/api/v3/activities/"+str(stravaid)+"/streams/latlng?resolution="+fetchresolution
|
|
latlongjson = requests.get(url,headers=headers)
|
|
|
|
t = np.array(timejson.json()[0]['data'])
|
|
d = np.array(distancejson.json()[0]['data'])
|
|
nr_rows = len(t)
|
|
|
|
try:
|
|
spm = np.array( spmjson.json()[1]['data'])
|
|
except IndexError:
|
|
spm = np.zeros(nr_rows)
|
|
|
|
try:
|
|
hr = np.array(hrjson.json()[1]['data'])
|
|
except IndexError:
|
|
hr = np.zeros(nr_rows)
|
|
|
|
try:
|
|
velo = np.array(velojson.json()[1]['data'])
|
|
except IndexError:
|
|
velo = np.zeros(nr_rows)
|
|
|
|
dt = np.diff(t).mean()
|
|
wsize = round(5./dt)
|
|
|
|
velo2 = ewmovingaverage(velo,wsize)
|
|
coords = np.array(latlongjson.json()[0]['data'])
|
|
try:
|
|
lat = coords[:,0]
|
|
lon = coords[:,1]
|
|
except IndexError:
|
|
lat = np.zeros(len(t))
|
|
lon = np.zeros(len(t))
|
|
|
|
|
|
|
|
|
|
strokelength = velo*60./(spm)
|
|
strokelength[np.isinf(strokelength)] = 0.0
|
|
|
|
pace = 500./(1.0*velo2)
|
|
pace[np.isinf(pace)] = 0.0
|
|
|
|
df = pd.DataFrame({'t':10*t,
|
|
'd':10*d,
|
|
'p':10*pace,
|
|
'spm':spm,
|
|
'hr':hr,
|
|
'lat':lat,
|
|
'lon':lon,
|
|
'strokelength':strokelength,
|
|
})
|
|
|
|
# startdatetime = datetime.datetime.strptime(startdatetime,"%Y-%m-%d-%H:%M:%S")
|
|
|
|
return [workoutsummary,df]
|
|
|
|
def createstravaworkoutdata(w):
|
|
filename = w.csvfilename
|
|
row = rowingdata(filename)
|
|
tcxfilename = filename[:-4]+'.tcx'
|
|
row.exporttotcx(tcxfilename,notes=w.notes)
|
|
|
|
return tcxfilename
|
|
|
|
def handle_stravaexport(file,workoutname,stravatoken,description=''):
|
|
# w = Workout.objects.get(id=workoutid)
|
|
client = stravalib.Client(access_token=stravatoken)
|
|
|
|
act = client.upload_activity(file,'tcx',name=workoutname)
|
|
res = act.wait(poll_interval=5.0)
|
|
|
|
|
|
# description doesn't work yet. Have to wait for stravalib to update
|
|
act = client.update_activity(res.id,activity_type='Rowing',description=description)
|
|
|
|
|
|
# w.uploadedtostrava = res.id
|
|
# w.save()
|
|
file.close()
|
|
|
|
return res.id
|
|
|
|
|