from .integrations import SyncIntegration, NoTokenError, create_or_update_syncrecord, get_known_ids from rowers.models import User, Rower, Workout, TombStone from rowingdata import rowingdata from rowers.tasks import handle_sporttracks_sync, handle_sporttracks_workout_from_data from rowers.rower_rules import is_workout_user import rowers.mytypes as mytypes from rowsandall_app.settings import ( SPORTTRACKS_CLIENT_SECRET, SPORTTRACKS_CLIENT_ID, SPORTTRACKS_REDIRECT_URI ) import re import numpy import pytz import json import requests import datetime import pandas as pd 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, uniqify def getidfromuri(uri): # pragma: no cover m = re.search('/(\w.*)\/(\d+)', uri) return m.group(2) def getidfromresponse(response): # pragma: no cover t = response.json() uri = t['uris'][0] regex = '.*?sporttracks\.mobi\/api\/v2\/fitnessActivities/(\d+)\.json$' m = re.compile(regex).match(uri).group(1) id = int(m) return int(id) def default(o): # pragma: no cover if isinstance(o, numpy.int64): return int(o) raise TypeError class SportTracksIntegration(SyncIntegration): def __init__(self, *args, **kwargs): super(SportTracksIntegration, self).__init__(*args, **kwargs) self.oauth_data = { 'client_id': SPORTTRACKS_CLIENT_ID, 'client_secret': SPORTTRACKS_CLIENT_SECRET, 'redirect_uri': SPORTTRACKS_REDIRECT_URI, 'authorization_uri': "https://api.sporttracks.mobi/oauth2/authorize?", 'content_type': 'application/json', 'tokenname': 'sporttrackstoken', 'refreshtokenname': 'sporttracksrefreshtoken', 'expirydatename': 'sporttrackstokenexpirydate', 'bearer_auth': False, 'base_url': "https://api.sporttracks.mobi/oauth2/token", 'scope': 'write', } def get_name(self): return "SportTracks" def get_shortname(self): return "sporttracks" def open(self, *args, **kwargs) -> str: return super(SportTracksIntegration, self).open(*args, **kwargs) def createworkoutdata(self, w, *args, **kwargs): timezone = pytz.timezone(w.timezone) filename = w.csvfilename try: row = rowingdata(csvfile=filename) except: # pragma: no cover return {} try: averagehr = int(row.df[' HRCur (bpm)'].mean()) maxhr = int(row.df[' HRCur (bpm)'].max()) except KeyError: # pragma: no cover averagehr = 0 maxhr = 0 try: duration = w.duration.hour*3600 duration += w.duration.minute*60 duration += w.duration.second duration += +1.0e-6*w.duration.microsecond except AttributeError: # pragma: no cover return {} t = row.df.loc[:, 'TimeStamp (sec)'].values - \ row.df.loc[:, 'TimeStamp (sec)'].iloc[0] try: t[0] = t[1] except IndexError: # pragma: no cover return {} d = row.df.loc[:, 'cum_dist'].values d[0] = d[1] t = t.astype(int) d = d.astype(int) spm = row.df[' Cadence (stokes/min)'].astype(int).values spm[0] = spm[1] hr = row.df[' HRCur (bpm)'].astype(int).values haslatlon = 1 try: lat = row.df[' latitude'].values lon = row.df[' longitude'].values if not lat.std() and not lon.std(): # pragma: no cover haslatlon = 0 except KeyError: haslatlon = 0 haspower = 1 try: power = row.df[' Power (watts)'].astype(int).values except KeyError: # pragma: no cover haspower = 0 locdata = [] hrdata = [] spmdata = [] distancedata = [] powerdata = [] t = t.tolist() hr = hr.tolist() d = d.tolist() spm = spm.tolist() if haslatlon: lat = lat.tolist() lon = lon.tolist() power = power.tolist() for i in range(len(t)): hrdata.append(t[i]) hrdata.append(hr[i]) distancedata.append(t[i]) distancedata.append(d[i]) spmdata.append(t[i]) spmdata.append(spm[i]) if haslatlon: locdata.append(t[i]) locdata.append([lat[i], lon[i]]) if haspower: powerdata.append(t[i]) powerdata.append(power[i]) try: w.notes = w.notes+'\n from '+w.workoutsource+' via rowsandall.com' except TypeError: w.notes = 'from '+w.workoutsource+' via rowsandall.com' st = w.startdatetime.astimezone(timezone) st = st.replace(microsecond=0) data = { "type": "Rowing", "name": w.name, "start_time": st.isoformat(), "total_distance": int(w.distance), "duration": duration, "notes": w.notes, "avg_heartrate": averagehr, "max_heartrate": maxhr, "distance": distancedata, "cadence": spmdata, "heartrate": hrdata, } if haslatlon: data = { "type": "Rowing", "name": w.name, "start_time": st.isoformat(), "total_distance": int(w.distance), "duration": duration, "notes": w.notes, "avg_heartrate": averagehr, "max_heartrate": maxhr, "location": locdata, "distance": distancedata, "cadence": spmdata, "heartrate": hrdata, } if haspower: data['power'] = powerdata return data def workout_export(self, workout, *args, **kwargs) -> str: thetoken = self.open() stid = "0" # ready to upload. Hurray if not(is_workout_user(self.user, workout)): return "0" authorizationstring = str('Bearer ' + thetoken) headers = {'Authorization': authorizationstring, 'user-agent': 'sanderroosendaal', 'Content-Type': 'application/json'} data = self.createworkoutdata(workout) if not data: return "0" url = "https://api.sporttracks.mobi/api/v2/fitnessActivities.json" _ = myqueue( queue, handle_sporttracks_sync, workout.id, url, headers, json.dumps(data, default=default)) return 1 def get_workouts(self, *args, **kwargs) -> int: r = self.rower workouts_json = self.get_workout_list_json(*args, **kwargs) stids = [int(getidfromuri(item['uri'])) for item in workouts_json['items']] knownstids = get_known_ids(r, 'sporttracksid') newids = [stid for stid in stids if stid not in knownstids] for sporttracksid in newids: id = self.get_workout(sporttracksid) return 1 def get_workout(self, id, *args, **kwargs) -> int: _ = self.open() r = self.rower record = create_or_update_syncrecord(r, None, sporttracksid=id) job = myqueue( queue, handle_sporttracks_workout_from_data, self.user, id, 'sporttracks', 'sporttracks' ) return job.id def get_workout_list_json(self, *args, **kwargs) -> dict: _ = self.open() r = self.rower authorizationstring = str('Bearer ' + r.sporttrackstoken) headers = {'Authorization': authorizationstring, 'user-agent': 'sanderroosendaal', 'Content-Type': 'application/json'} url = "https://api.sporttracks.mobi/api/v2/fitnessActivities" res = requests.get(url, headers=headers) if (res.status_code != 200): s = "Token doesn't exist. Need to authorize" raise NoTokenError(s) return res.json() def get_workout_list(self, *args, **kwargs) -> list: r = self.rower workouts_json = self.get_workout_list_json(*args, **kwargs) workouts = [] knownstids = get_known_ids(r, 'sporttracksid') for item in workouts_json['items']: d = int(float(item['total_distance'])) i = int(getidfromuri(item['uri'])) if i in knownstids: # pragma: no cover nnn = '' else: nnn = 'NEW' n = item['name'] ttot = str(datetime.timedelta(seconds=int(float(item['duration'])))) s = item['start_time'] r = item['type'] keys = ['id', 'distance', 'duration', 'starttime', 'rowtype', 'source', 'name', 'new'] values = [i, d, ttot, s, r, None, n, nnn] res = dict(zip(keys, values)) workouts.append(res) return workouts def make_authorization_url(self, *args, **kwargs) -> str: # pragma: no cover return super(SportTracksIntegration, self).make_authorization_url(*args, **kwargs) def get_token(self, code, *args, **kwargs) -> (str, int, str): return super(SportTracksIntegration, self).get_token(code, *args, **kwargs) def token_refresh(self, *args, **kwargs) -> str: return super(SportTracksIntegration, self).token_refresh(*args, **kwargs)