doing trainingpeaks, untested

rowers/integrations/__init__.py

@@ -3,3 +3,5 @@ from .strava import StravaIntegration
 from .nk import NKIntegration
 from .sporttracks import SportTracksIntegration
 from .rp3 import RP3Integration
+from .trainingpeaks import TPIntegration
+

rowers/integrations/rp3.py

@@ -244,8 +244,4 @@ class RP3Integration(SyncIntegration):
         return super(RP3Integration, self).token_refresh(*args, **kwargs)
 
     
-# just as a quick test during development
-u = User.objects.get(id=1)
-
-integration_1 = RP3Integration(u)
 

rowers/integrations/trainingpeaks.py

@@ -0,0 +1,134 @@
+from .integrations import SyncIntegration, NoTokenError
+from rowers.models import User, Rower, Workout, TombStone
+
+import django_rq
+queue = django_rq.get_queue('default')
+queuelow = django_rq.get_queue('low')
+queuehigh = django_rq.get_queue('high')
+
+from rowers.utils import myqueue, dologging, myqueue
+
+import requests
+
+from rowingdata import rowingdata
+
+from rowers.rower_rules import is_workout_user
+import time
+from django_rq import job
+
+from rowers.tasks import check_tp_workout_id, handle_workout_tp_upload
+
+from rowsandall_app.settings import (
+    TP_CLIENT_ID, TP_CLIENT_SECRET,
+    TP_REDIRECT_URI, TP_CLIENT_KEY,TP_API_LOCATION,
+    TP_OAUTH_LOCATION,
+)
+
+import gzip
+
+import base64
+from io import BytesIO
+
+
+tpapilocation = TP_API_LOCATION
+
+
+
+class TPIntegration(SyncIntegration):
+    def __init__(self, *args, **kwargs):
+        super(TPIntegration, self).__init__(*args, **kwargs)
+        self.oauth_data = {
+            'client_id': TP_CLIENT_ID,
+            'client_secret': TP_CLIENT_SECRET,
+            'redirect_uri': TP_REDIRECT_URI,
+            'autorization_uri': "https://oauth.trainingpeaks.com/oauth/authorize?",
+            'content_type': 'application/x-www-form-urlencoded',
+            'tokenname': 'tptoken',
+            'refreshtokenname': 'tprefreshtoken',
+            'expirydatename': 'tptokenexpirydate',
+            'bearer_auth': False,
+            'base_url': "https://oauth.trainingpeaks.com/oauth/token",
+            'scope': 'write',
+        }
+
+    def createworkoutdata(self, w, *args, **kwargs):
+        filename = w.csvfilename
+        row = rowingdata(csvfile=filename)
+        tcxfilename = filename[:-4]+'.tcx'
+        try:
+            newnotes = w.notes+'\n from '+w.workoutsource+' via rowsandall.com'
+        except TypeError:
+            newnotes = 'from '+w.workoutsource+' via rowsandall.com'
+
+        row.exporttotcx(tcxfilename, notes=newnotes)
+
+        return tcxfilename
+
+
+    def workout_export(self, workout, *args, **kwargs) -> str:
+        thetoken = self.open()
+        tcxfilename = self.createworkoutdata(workout)
+        job = myqueue(
+            queue,
+            handle_workout_tp_upload,
+            workout,
+            thetoken,
+            tcxfilename
+        )
+        return job.id
+
+
+    def get_workouts(self, *args, **kwargs) -> int:
+        raise NotImplementedError("not implemented")
+
+    def get_workout(self, id) -> int:
+        raise NotImplementedError("not implemented")
+
+    def get_workout_list(self, *args, **kwargs) -> list:
+        raise NotImplementedError("not implemented")
+
+    def make_authorization_url(self, *args, **kwargs) -> str:  # pragma: no cover
+        return super(TPIntegration, self).make_authorization_url(self, *args, **kwargs)
+
+    def get_token(self, code, *args, **kwargs) -> (str, int, str):
+            # 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,
+        }
+
+        response = requests.post(
+            TP_OAUTH_LOCATION+"/oauth/token/",
+            data=post_data, verify=False,
+        )
+
+        if response.status_code != 200:
+            raise NoTokenError
+
+        try:
+            token_json = response.json()
+            thetoken = token_json['access_token']
+            expires_in = token_json['expires_in']
+            refresh_token = token_json['refresh_token']
+        except KeyError:  # pragma: no cover
+            thetoken = ""
+            expires_in = 0
+            refresh_token = ""
+
+        return thetoken, expires_in, refresh_token
+
+
+    def open(self, *args, **kwargs) -> str:
+        return super(TPIntegration, self).open(*args, **kwargs)
+
+    def token_refresh(self, *args, **kwargs) -> str:
+        return super(TPIntegration, self).token_refresh(*args, **kwargs)
+
+# just as a quick test during development
+u = User.objects.get(id=1)
+
+integration_1 = TPIntegration(u)
+