increase test coverage
This commit is contained in:
Sander Roosendaal
@@ -838,25 +838,6 @@ def get_workout(user,c2id,do_async=True):
|
||||
|
||||
return 1
|
||||
|
||||
# Get stroke data belonging to C2 ID
|
||||
def get_c2_workout_strokes(user,c2id):
|
||||
r = Rower.objects.get(user=user)
|
||||
if (r.c2token == '') or (r.c2token is None): # pragma: no cover
|
||||
return custom_exception_handler(401,s)
|
||||
s = "Token doesn't exist. Need to authorize"
|
||||
elif (timezone.now()>r.tokenexpirydate): # pragma: no cover
|
||||
s = "Token expired. Needs to refresh."
|
||||
return custom_exception_handler(401,s)
|
||||
else:
|
||||
# ready to fetch. Hurray
|
||||
authorizationstring = str('Bearer ' + r.c2token)
|
||||
headers = {'Authorization': authorizationstring,
|
||||
'user-agent': 'sanderroosendaal',
|
||||
'Content-Type': 'application/json'}
|
||||
url = "https://log.concept2.com/api/users/me/results/"+str(c2id)+"/strokes"
|
||||
s = requests.get(url,headers=headers)
|
||||
|
||||
return s
|
||||
|
||||
# Get list of C2 workouts. We load only the first page,
|
||||
# assuming that users don't want to import their old workouts
|
||||
@@ -1028,190 +1009,3 @@ def rower_c2_token_refresh(user):
|
||||
return r.c2token
|
||||
else: # pragma: no cover
|
||||
return None
|
||||
|
||||
# Create workout data from Strava or Concept2
|
||||
# data and create the associated Workout object and save it
|
||||
def add_workout_from_data(user,importid,data,strokedata,
|
||||
source='c2',splitdata=None,
|
||||
workoutsource='concept2'):
|
||||
try:
|
||||
workouttype = mytypes.c2mappinginv[data['type']]
|
||||
except KeyError: # pragma: no cover
|
||||
workouttype = 'rower'
|
||||
|
||||
if workouttype not in [x[0] for x in Workout.workouttypes]: # pragma: no cover
|
||||
workouttype = 'other'
|
||||
try:
|
||||
comments = data['comments']
|
||||
except: # pragma: no cover
|
||||
comments = ' '
|
||||
|
||||
try:
|
||||
thetimezone = pytz.timezone(data['timezone'])
|
||||
except UnknownTimeZoneError:
|
||||
thetimezone = 'UTC'
|
||||
|
||||
|
||||
r = Rower.objects.get(user=user)
|
||||
|
||||
try:
|
||||
rowdatetime = iso8601.parse_date(data['date_utc'])
|
||||
thetimezone = 'UTC'
|
||||
except KeyError: # pragma: no cover
|
||||
rowdatetime = iso8601.parse_date(data['start_date'])
|
||||
rowdatetime = rowdatetime.make_aware(thetimezone)
|
||||
except ParseError: # pragma: no cover
|
||||
rowdatetime = iso8601.parse_date(data['date'])
|
||||
rowdatetime = rowdatetime.make_aware(thetimezone)
|
||||
|
||||
|
||||
try:
|
||||
c2intervaltype = data['workout_type']
|
||||
|
||||
except KeyError: # pragma: no cover
|
||||
c2intervaltype = ''
|
||||
|
||||
try:
|
||||
title = data['name']
|
||||
except KeyError:
|
||||
title = ""
|
||||
try:
|
||||
t = data['comments'].split('\n', 1)[0]
|
||||
title += t[:40]
|
||||
except: # pragma: no cover
|
||||
title = ''
|
||||
|
||||
try:
|
||||
comments = data['comments']
|
||||
except KeyError: # pragma: no cover
|
||||
comments = ''
|
||||
|
||||
starttimeunix = arrow.get(rowdatetime).timestamp()
|
||||
|
||||
res = make_cumvalues(0.1*strokedata['t'])
|
||||
cum_time = res[0]
|
||||
lapidx = res[1]
|
||||
|
||||
totaltime = data['time']/10.
|
||||
starttimeunix = starttimeunix - totaltime
|
||||
|
||||
|
||||
unixtime = cum_time+starttimeunix
|
||||
# unixtime[0] = starttimeunix
|
||||
seconds = 0.1*strokedata.loc[:,'t']
|
||||
|
||||
nr_rows = len(unixtime)
|
||||
|
||||
try: # pragma: no cover
|
||||
latcoord = strokedata.loc[:,'lat']
|
||||
loncoord = strokedata.loc[:,'lon']
|
||||
except:
|
||||
latcoord = np.zeros(nr_rows)
|
||||
loncoord = np.zeros(nr_rows)
|
||||
|
||||
|
||||
try:
|
||||
strokelength = strokedata.loc[:,'strokelength']
|
||||
except:
|
||||
strokelength = np.zeros(nr_rows)
|
||||
|
||||
dist2 = 0.1*strokedata.loc[:,'d']
|
||||
|
||||
try:
|
||||
spm = strokedata.loc[:,'spm']
|
||||
except KeyError: # pragma: no cover
|
||||
spm = 0*dist2
|
||||
|
||||
try:
|
||||
hr = strokedata.loc[:,'hr']
|
||||
except KeyError: # pragma: no cover
|
||||
hr = 0*spm
|
||||
pace = strokedata.loc[:,'p']/10.
|
||||
pace = np.clip(pace,0,1e4)
|
||||
pace = pace.replace(0,300)
|
||||
|
||||
velo = 500./pace
|
||||
power = 2.8*velo**3
|
||||
if workouttype in ['bike','bikeerg']: # pragma: no cover
|
||||
velo = 1000./pace
|
||||
pace = 500./velo
|
||||
|
||||
# save csv
|
||||
# Create data frame with all necessary data to write to csv
|
||||
df = pd.DataFrame({'TimeStamp (sec)':unixtime,
|
||||
' Horizontal (meters)': dist2,
|
||||
' Cadence (stokes/min)':spm,
|
||||
' HRCur (bpm)':hr,
|
||||
' longitude':loncoord,
|
||||
' latitude':latcoord,
|
||||
' Stroke500mPace (sec/500m)':pace,
|
||||
' Power (watts)':power,
|
||||
' DragFactor':np.zeros(nr_rows),
|
||||
' DriveLength (meters)':np.zeros(nr_rows),
|
||||
' StrokeDistance (meters)':strokelength,
|
||||
' DriveTime (ms)':np.zeros(nr_rows),
|
||||
' StrokeRecoveryTime (ms)':np.zeros(nr_rows),
|
||||
' AverageDriveForce (lbs)':np.zeros(nr_rows),
|
||||
' PeakDriveForce (lbs)':np.zeros(nr_rows),
|
||||
' lapIdx':lapidx,
|
||||
' WorkoutState': 4,
|
||||
' ElapsedTime (sec)':seconds
|
||||
})
|
||||
|
||||
|
||||
df.sort_values(by='TimeStamp (sec)',ascending=True)
|
||||
|
||||
timestr = strftime("%Y%m%d-%H%M%S")
|
||||
|
||||
|
||||
# Create CSV file name and save data to CSV file
|
||||
csvfilename ='media/{code}_{importid}.csv'.format(
|
||||
importid=importid,
|
||||
code = uuid4().hex[:16]
|
||||
)
|
||||
|
||||
res = df.to_csv(csvfilename+'.gz',index_label='index',
|
||||
compression='gzip')
|
||||
|
||||
|
||||
# with Concept2
|
||||
if source=='c2':
|
||||
try:
|
||||
totaldist = data['distance']
|
||||
totaltime = data['time']/10.
|
||||
except KeyError: # pragma: no cover
|
||||
totaldist = 0
|
||||
totaltime = 0
|
||||
else: # pragma: no cover
|
||||
totaldist = 0
|
||||
totaltime = 0
|
||||
|
||||
id,message = dataprep.save_workout_database(
|
||||
csvfilename,r,
|
||||
workouttype=workouttype,
|
||||
title=title,notes=comments,
|
||||
workoutsource=workoutsource,
|
||||
dosummary=True,dosmooth=False,
|
||||
)
|
||||
|
||||
|
||||
w = Workout.objects.get(id=id)
|
||||
|
||||
try:
|
||||
local_tz = pytz.timezone(data['timezone'])
|
||||
except UnknownTimeZoneError:
|
||||
local_tz = pytz.utc
|
||||
# local_tz = pytz.timezone(thetimezone)
|
||||
|
||||
w.startdatetime = w.startdatetime.astimezone(local_tz)
|
||||
w.starttime = w.startdatetime.strftime('%H:%M:%S')
|
||||
w.timezone = local_tz
|
||||
|
||||
w.duration = dataprep.totaltime_sec_to_string(totaltime)
|
||||
w.distance = totaldist
|
||||
|
||||
|
||||
w.save()
|
||||
|
||||
|
||||
return id,message
|
||||
|
||||
Reference in New Issue
Block a user