# All the functionality needed to connect to Runkeeper # 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 import gzip from math import sin,cos,atan2,sqrt import os,sys import urllib import base64 from io import BytesIO # 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 from rowsandall_app.settings import ( C2_CLIENT_ID, C2_REDIRECT_URI, C2_CLIENT_SECRET, STRAVA_CLIENT_ID, STRAVA_REDIRECT_URI, STRAVA_CLIENT_SECRET, TP_CLIENT_ID, TP_CLIENT_SECRET, TP_REDIRECT_URI,TP_CLIENT_KEY, ) tpapilocation = "https://api.sandbox.trainingpeaks.com" # Custom error class - to raise a NoTokenError class TPNoTokenError(Exception): def __init__(self,value): self.value=value def __str__(self): return repr(self.value) # Exponentially weighted moving average # Used for data smoothing of the jagged data obtained by Strava # See bitbucket issue 72 def ewmovingaverage(interval,window_size): # Experimental code using Exponential Weighted moving average try: 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 except ValueError: interval2 = interval return interval2 from utils import geo_distance # Custom exception handler, returns a 401 HTTP message # with exception details in the json data 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 # Refresh ST token using refresh token def do_refresh_token(refreshtoken): client_auth = requests.auth.HTTPBasicAuth(TP_CLIENT_KEY, TP_CLIENT_SECRET) post_data = {"grant_type": "refresh_token", "client_secret": TP_CLIENT_SECRET, "client_id":TP_CLIENT_KEY, "refresh_token": refreshtoken, } headers = {'user-agent': 'sanderroosendaal', 'Accept': 'application/json', 'Content-Type': 'application/x-www-form-urlencoded', } url = "https://oauth.sandbox.trainingpeaks.com/oauth/token" response = requests.post(url, data=post_data, headers=headers) if response.status_code == 200: token_json = response.json() thetoken = token_json['access_token'] expires_in = token_json['expires_in'] try: refresh_token = token_json['refresh_token'] except KeyError: refresh_token = refreshtoken else: return [0,0,0] return [thetoken,expires_in,refresh_token] # Exchange access code for long-lived access token def get_token(code): client_auth = requests.auth.HTTPBasicAuth(TP_CLIENT_KEY, TP_CLIENT_SECRET) post_data = { "client_id":TP_CLIENT_KEY, "grant_type": "authorization_code", "code": code, "redirect_uri":TP_REDIRECT_URI, "client_secret": TP_CLIENT_SECRET, } headers = { 'Content-Type': 'application/x-www-form-urlencoded', } response = requests.post("https://oauth.sandbox.trainingpeaks.com/oauth/token", data=post_data) try: token_json = response.json() thetoken = token_json['access_token'] expires_in = token_json['expires_in'] refresh_token = token_json['refresh_token'] except KeyError: thetoken = 0 return thetoken,expires_in,refresh_token # Make authorization URL including random string 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": TP_CLIENT_KEY, "response_type": "code", "redirect_uri": TP_REDIRECT_URI, "scope": "file:write", } url = "https://oauth.sandbox.trainingpeaks.com/oauth/authorize?" +urllib.urlencode(params) return HttpResponseRedirect(url) def getidfromresponse(response): t = json.loads(response.text) links = t["_links"] id = links["self"][0]["id"] return int(id) def createtpworkoutdata(w): filename = w.csvfilename try: row = rowingdata(filename) tcxfilename = filename[:-4]+'.tcx' row.exporttotcx(tcxfilename,notes=w.notes+'\n from '+w.workoutsource+' via rowsandall.com') return tcxfilename except: tcxfilename = 0 return tcxfilename def tp_check(access_token): headers = { "Content-Type": "application/json", 'Accept': 'application/json', 'authorization': 'Bearer %s' % access_token } resp = requests.post(tpapilocation+"/v1/info/version", headers=headers) return resp def uploadactivity(access_token,filename,description='', name='Rowsandall.com workout'): data_gz = BytesIO() with file(filename,'rb') as inF: s = inF.read() with gzip.GzipFile(fileobj=data_gz,mode="w") as gzf: gzf.write(s) headers = { "Content-Type": "application/json", 'Accept': 'application/json', 'Authorization': 'Bearer %s' % access_token } data = { "UploadClient": "rowsandall", "Filename": filename, "SetWorkoutPublic": True, "Title":name, "Type": "rowing", "Comment": description, "Data": base64.b64encode(data_gz.getvalue()).decode("ascii") } resp = requests.post(tpapilocation+"/v1/file", data = json.dumps(data), headers=headers) if resp.status_code != 200: if settings.DEBUG: print resp.status_code print resp.reason print "" print headers print "" return 0,resp.reason,resp.status_code,headers else: return resp.json()[0]["Id"],"ok",200,"" return 0