from rowers.forms import analysischoices from rowers.views.statements import * from jinja2 import Environment, FileSystemLoader from rowers.rower_rules import can_view_session from django.forms.widgets import SelectDateWidget, HiddenInput from polars.exceptions import ComputeError def floatformat(x, prec=2): # pragma: no cover return '{x}'.format(x=round(x, prec)) env = Environment(loader=FileSystemLoader(["rowers/templates"])) env.filters['floatformat'] = floatformat # generic Analysis view - defaultoptions = { 'includereststrokes': False, 'workouttypes': ['rower', 'dynamic', 'slides'], 'waterboattype': mytypes.waterboattype, 'function': 'boxplot', 'ranking': False, 'trendline': False, } @permission_required('rower.is_coach', fn=get_user_by_userid, raise_exception=True) def analysis_new(request, userid=0, function='boxplot', teamid=0, id='', session=0): r = getrequestrower(request, userid=userid) user = r.user userid = user.id worldclass = False firstworkout = None if id: firstworkout = get_workout(id) if not is_workout_team(request.user, firstworkout): # pragma: no cover raise PermissionDenied("You are not allowed to use this workout") #if workout_is_strava(firstworkout): # messages.error(request, "You cannot use Strava workouts for analysis") # raise PermissionDenied("You cannot use Strava workouts for analysis") firstworkoutquery = Workout.objects.filter(id=encoder.decode_hex(id)) try: theteam = Team.objects.get(id=teamid) except Team.DoesNotExist: theteam = None try: thesession = PlannedSession.objects.get(id=session) if not can_view_session(user, thesession): # pragma: no cover raise PermissionDenied("you cannot view this session") except PlannedSession.DoesNotExist: thesession = None if 'options' in request.session: options = request.session['options'] else: options = defaultoptions options['userid'] = userid try: modalities = options['modalities'] modality = modalities[0] except KeyError: modalities = [m[0] for m in mytypes.workouttypes_ordered.items()] modality = 'all' try: ranking = options['ranking'] except KeyError: ranking = False try: worldclass = options['cpoverlay'] except KeyError: worldclass = False if 'startdate' in request.session: startdate = iso8601.parse_date(request.session['startdate']) else: startdate = timezone.now()-datetime.timedelta(days=42) if function not in [c[0] for c in analysischoices]: # pragma: no cover function = 'boxplot' if 'enddate' in request.session: enddate = iso8601.parse_date(request.session['enddate']) else: enddate = timezone.now() waterboattype = mytypes.waterboattype try: waterboattype = options['waterboattype'] except KeyError: # pragma: no cover pass if request.method == 'POST': thediv = get_call() dateform = DateRangeForm(request.POST) if dateform.is_valid(): startdate = dateform.cleaned_data['startdate'] enddate = dateform.cleaned_data['enddate'] startdatestring = startdate.strftime('%Y-%m-%d') enddatestring = enddate.strftime('%Y-%m-%d') request.session['startdate'] = startdatestring request.session['enddate'] = enddatestring optionsform = AnalysisOptionsForm(request.POST) if optionsform.is_valid(): for key, value in optionsform.cleaned_data.items(): options[key] = value modality = optionsform.cleaned_data['modality'] waterboattype = optionsform.cleaned_data['waterboattype'] ranking = optionsform.cleaned_data['ranking'] if modality == 'all': modalities = [m[0] for m in mytypes.workouttypes] else: # pragma: no cover modalities = [modality] if modality not in ['water','rower']: waterboattype = [b[0] for b in mytypes.boattypes] options['modalities'] = modalities options['waterboattype'] = waterboattype options['ranking'] = ranking try: worldclass = options['cpoverlay'] except KeyError: worldclass = False options['cpoverlay'] = worldclass chartform = AnalysisChoiceForm(request.POST) if chartform.is_valid(): options['form'] = chartform.cleaned_data for key, value in chartform.cleaned_data.items(): options[key] = value form = WorkoutMultipleCompareForm(request.POST) if form.is_valid(): cd = form.cleaned_data selectedworkouts = cd['workouts'] ids = [int(w.id) for w in selectedworkouts] options['ids'] = ids else: # pragma: no cover ids = [] options['ids'] = ids else: thediv = '' dateform = DateRangeForm(initial={ 'startdate': startdate, 'enddate': enddate, }) negtypes = [] for b in mytypes.boattypes: if b[0] not in waterboattype: # pragma: no cover negtypes.append(b[0]) for b in mytypes.ergtypes: if b[0] not in waterboattype: # pragma: no cover negtypes.append(b[0]) startdate = datetime.datetime.combine(startdate, datetime.time()) enddate = datetime.datetime.combine(enddate, datetime.time(23, 59, 59)) if enddate < startdate: # pragma: no cover s = enddate enddate = startdate startdate = s # make sure the dates are not naive try: startdate = pytz.utc.localize(startdate) except (ValueError, AttributeError): # pragma: no cover pass try: enddate = pytz.utc.localize(enddate) except (ValueError, AttributeError): # pragma: no cover pass rankingtypes = [False,True] if ranking: # pragma: no cover rankingtypes = [True] if theteam is not None and (theteam.viewing == 'allmembers' or theteam.manager == request.user): # pragma: no cover workouts = Workout.objects.filter(team=theteam, startdatetime__gte=startdate, startdatetime__lte=enddate, workouttype__in=modalities, rankingpiece__in=rankingtypes, )#.exclude(workoutsource='strava') elif theteam is not None and theteam.viewing == 'coachonly': # pragma: no cover workouts = Workout.objects.filter(team=theteam, user=r, startdatetime__gte=startdate, startdatetime__lte=enddate, workouttype__in=modalities, rankingpiece__in=rankingtypes, )#.exclude(workoutsource='strava') elif thesession is not None: workouts = get_workouts_session(r, thesession) else: workouts = Workout.objects.filter(user=r, startdatetime__gte=startdate, startdatetime__lte=enddate, workouttype__in=modalities, rankingpiece__in=rankingtypes, ) if firstworkout: workouts = firstworkoutquery | workouts workouts = workouts.order_by( "-date", "-starttime" ).exclude(boattype__in=negtypes) query = request.POST.get('q') if query: # pragma: no cover query_list = query.split() try: workouts = workouts.filter( reduce(operator.and_, (Q(name__icontains=q) for q in query_list)) | reduce(operator.and_, (Q(notes__icontains=q) for q in query_list)) ) searchform = SearchForm(initial={'q': query}) except TypeError: searchform = SearchForm() else: searchform = SearchForm() if request.method != 'POST': form = WorkoutMultipleCompareForm() if id: form.fields["workouts"].initial = [firstworkout] chartform = AnalysisChoiceForm(initial={'function': function}) try: chartformoptions = options['form'] chartformoptions['function'] = function chartform = AnalysisChoiceForm(initial=chartformoptions) except KeyError: pass selectedworkouts = Workout.objects.none() else: selectedworkouts = Workout.objects.filter(id__in=ids) form.fields["workouts"].queryset = (workouts | selectedworkouts)#.exclude(workoutsource='strava') optionsform = AnalysisOptionsForm(initial={ 'modality': modality, 'waterboattype': waterboattype, 'ranking': ranking, }) if r.birthdate: age = calculate_age(r.birthdate) else: age = 0 startdatestring = startdate.strftime('%Y-%m-%d') enddatestring = enddate.strftime('%Y-%m-%d') request.session['startdate'] = startdatestring request.session['enddate'] = enddatestring request.session['options'] = options savedata = options.get('savedata',False) if savedata: # pragma: no cover ids = options.get('ids',[]) tw = [] for id in ids: try: tw.append(Workout.objects.get(id=id)) except Workout.DoesNotExist: pass df = pd.DataFrame() function = options.get('function','') if function == 'boxplot': df = boxplotdata(tw, options) elif function == 'trendflex': # pragma: no cover df = trendflexdata(tw, options, userid=userid) elif function == 'histo': # pragma: no cover df = histodata(tw, options) elif function == 'flexall': # pragma: no cover df = flexalldata(tw, options) elif function == 'compare': # pragma: no cover df = comparisondata(tw, options) elif function == 'cp': # pragma: no cover df = cpdata(tw, options) options['savedata'] = False request.session['options'] = options try: response = HttpResponse(df.to_csv()) except AttributeError: response = HttpResponse(df.write_csv()) code = str(uuid4()) filename = code+'.csv' chartform.fields['savedata'].initial = False response['Content-Disposition'] = 'attachment; filename="%s"' % filename response['Content-Type'] = 'application/octet-stream' return response if not request.user.is_superuser: chartform.fields['savedata'].widget = HiddenInput() chartform.fields['savedata'].initial = False breadcrumbs = [ { 'url': '/rowers/analysis', 'name': 'Analysis' }, { 'url': reverse('analysis_new', kwargs={'userid': userid}), 'name': 'Analysis Select' }, ] return render(request, 'user_analysis_select.html', {'workouts': workouts, 'dateform': dateform, 'startdate': startdate, 'enddate': enddate, 'rower': r, 'breadcrumbs': breadcrumbs, 'theuser': user, 'the_div': thediv, 'form': form, 'active': 'nav-analysis', 'chartform': chartform, 'searchform': searchform, 'optionsform': optionsform, 'worldclass': worldclass, 'age': age, 'sex': r.sex, 'weightcategory': r.weightcategory, 'teams': get_my_teams(request.user), }) def trendflexdata(workouts, options, userid=0): includereststrokes = options['includereststrokes'] palette = options['palette'] groupby = options['groupby'] binsize = options['binsize'] xparam = options['xparam'] yparam = options['yparam'] spmmin = options['spmmin'] spmmax = options['spmmax'] workmin = options['workmin'] workmax = options['workmax'] ploterrorbars = options['ploterrorbars'] ids = options['ids'] workstrokesonly = not includereststrokes savedata = options.get('savedata',False) #try: # workouts = workouts.exclude(workoutsource='strava') #except AttributeError: # pragma: no cover # workouts = [w for w in workouts if w.workoutsource != 'strava'] fieldlist, fielddict = dataprep.getstatsfields() fieldlist = [xparam, yparam, groupby, 'workoutid', 'spm', 'driveenergy', 'workoutstate'] # prepare data frame datadf, extracols = dataprep.read_cols_df_sql(ids, fieldlist) if xparam == groupby: # pragma: no cover datadf['groupby'] = datadf[xparam] #groupby = 'groupby' datadf = dataprep.clean_df_stats(datadf, workstrokesonly=workstrokesonly) datadf = dataprep.filter_df(datadf, 'spm', spmmin, largerthan=True) datadf = dataprep.filter_df(datadf, 'spm', spmmax, largerthan=False) datadf = dataprep.filter_df(datadf, 'driveenergy', workmin, largerthan=True) datadf = dataprep.filter_df(datadf, 'driveneergy', workmax, largerthan=False) if yparam == 'power': datadf = dataprep.filter_df(datadf, 'power', 1, largerthan=True) datadf.dropna(axis=0, how='any', inplace=True) datemapping = { w.id: w.date for w in workouts } datadf['date'] = datadf['workoutid'] datadf['date'].replace(datemapping, inplace=True) today = timezone.now() try: datadf['days ago'] = list(map(lambda x: x.days, datadf.date - today)) except TypeError: datadf['days ago'] = 0 if groupby != 'date': try: bins = np.arange(datadf[groupby].min()-binsize, datadf[groupby].max()+binsize, binsize) groups = datadf.groupby( pd.cut(datadf[groupby], bins, labels=False)) except (ValueError, AttributeError): # pragma: no cover return ('', 'Error: not enough data') except TypeError: return ('','Error: something wrong with the data') except KeyError: return ('','Error: no {groupby} data'.format(groupby=groupby)) else: # pragma: no cover try: bins = np.arange(datadf['days ago'].min()-binsize, datadf['days ago'].max()+binsize, binsize, ) groups = datadf.groupby(pd.cut(datadf['days ago'], bins, labels=False)) except (ValueError, AttributeError): # pragma: no cover return ('', 'Error: not enough data') xvalues = [] yvalues = [] xerror = [] yerror = [] groupsize = [] groupval = [] try: for key, item in groups: xvalues.append(groups.get_group(key)[xparam].mean()) yvalues.append(groups.get_group(key)[yparam].mean()) xerror.append(groups.get_group(key)[xparam].std()) yerror.append(groups.get_group(key)[yparam].std()) groupsize.append(len(groups.get_group(key)[xparam])) groupval.append(groups.get_group(key)[groupby].mean()) except TypeError: return('','Error: Unable to compute data') xvalues = pd.Series(xvalues) yvalues = pd.Series(yvalues) xerror = pd.Series(xerror) yerror = pd.Series(yerror) groupsize = pd.Series(groupsize) mask = groupsize <= min([0.01*groupsize.sum(), 0.2*groupsize.mean()]) xvalues.loc[mask] = np.nan yvalues.loc[mask] = np.nan xerror.loc[mask] = np.nan yerror.loc[mask] = np.nan groupsize.loc[mask] = np.nan xvalues.dropna(inplace=True) yvalues.dropna(inplace=True) xerror.dropna(inplace=True) yerror.dropna(inplace=True) groupsize.dropna(inplace=True) if len(groupsize) == 0: # pragma: no cover messages.error('No data in selection') url = reverse(user_multiflex_select) return HttpResponseRedirect(url) else: groupsize = 30.*np.sqrt(groupsize/float(groupsize.max())) df = pd.DataFrame({ xparam: xvalues, yparam: yvalues, 'x': xvalues, 'y': yvalues, 'xerror': xerror, 'yerror': yerror, 'groupsize': groupsize, }) #if yparam == 'pace': # df['y'] = dataprep.paceformatsecs(df['y']/1.0e3) aantal = len(df) if groupby != 'date': try: df['groupval'] = pd.Series(groupval) df.loc[mask, 'groupval'] = np.nan groupcols = df['groupval'] except (ValueError, AttributeError): # pragma: no cover df['groupval'] = groups.mean()[groupby].fillna(value=0) df.loc[mask, 'groupval'] = np.nan groupcols = df['groupval'] except KeyError: # pragma: no cover messages.error(request, 'Data selection error') url = reverse(user_multiflex_select) return HttpResponseRedirect(url) else: # pragma: no cover try: dates = groups.min()[groupby] dates.loc[mask] = np.nan dates.dropna(inplace=True) df['groupval'] = [x.strftime("%Y-%m-%d") for x in dates] df['groupval'].loc[mask] = np.nan groupcols = 100.*np.arange(aantal)/float(aantal) except AttributeError: df['groupval'] = groups.mean()['days ago'].fillna(value=0) groupcols = 100.*np.arange(aantal)/float(aantal) groupcols = (groupcols-groupcols.min())/(groupcols.max()-groupcols.min()) if aantal == 1: # pragma: no cover groupcols = np.array([1.]) colors = range_to_color_hex(groupcols, palette=palette) df['color'] = colors clegendx = np.arange(0, 1.2, .2) legcolors = range_to_color_hex(clegendx, palette=palette) if groupby != 'date': clegendy = df['groupval'].min()+clegendx * \ (df['groupval'].max()-df['groupval'].min()) else: # pragma: no cover clegendy = df.index.min()+clegendx*(df.index.max()-df.index.min()) colorlegend = zip(range(6), clegendy, legcolors) if userid == 0: extratitle = '' else: # pragma: no cover u = User.objects.get(id=userid) extratitle = ' '+u.first_name+' '+u.last_name savedata = options.get('savedata',False) if savedata: # pragma: no cover return df script, div = interactive_multiflex(df, xparam, yparam, groupby, extratitle=extratitle, ploterrorbars=ploterrorbars, binsize=binsize, colorlegend=colorlegend, spmmin=spmmin, spmmax=spmmax, workmin=workmin, workmax=workmax) return(script, div) def flexalldata(workouts, options): includereststrokes = options['includereststrokes'] xparam = options['xaxis'] yparam1 = options['yaxis1'] yparam2 = options['yaxis2'] trendline = options['trendline'] promember = True #try: # workouts = workouts.exclude(workoutsource='strava') #except AttributeError: # pragma: no cover # workouts = [w for w in workouts if w.workoutsource != 'strava'] workstrokesonly = not includereststrokes userid = options['userid'] if userid == 0: # pragma: no cover extratitle = '' else: u = User.objects.get(id=userid) extratitle = ' '+u.first_name+' '+u.last_name savedata = options.get('savedata',False) if savedata: # pragma: no cover workstrokesonly = not includereststrokes columns = [xparam, yparam1, yparam2, 'spm', 'driveenergy', 'distance'] ids = [int(w.id) for w in workouts] df = dataprep.read_data(columns, ids=ids, workstrokesonly=workstrokesonly, doclean=True, ) return df res = interactive_cum_flex_chart2(workouts, xparam=xparam, yparam1=yparam1, yparam2=yparam2, promember=promember, extratitle=extratitle, workstrokesonly=workstrokesonly, trendline=trendline, ) script = res[0] div = res[1] return(script, div) def histodata(workouts, options): includereststrokes = options['includereststrokes'] plotfield = options['plotfield'] spmmin = options['spmmin'] spmmax = options['spmmax'] workmin = options['workmin'] workmax = options['workmax'] userid = options['userid'] #try: # workouts = workouts.exclude(workoutsource='strava') #except AttributeError: # pragma: no cover # workouts = [w for w in workouts if w.workoutsource != 'strava'] if userid == 0: # pragma: no cover extratitle = '' else: u = User.objects.get(id=userid) extratitle = ' '+u.first_name+' '+u.last_name savedata = options.get('savedata',False) if savedata: # pragma: no cover workstrokesonly = not includereststrokes ids = [int(w.id) for w in workouts] df = dataprep.read_data([plotfield], ids=ids, workstrokesonly=workstrokesonly, doclean=True, ) return df script, div = interactive_histoall(workouts, plotfield, includereststrokes, spmmin=spmmin, spmmax=spmmax, extratitle=extratitle, workmin=workmin, workmax=workmax) return(script, div) def cpdata(workouts, options): start = timezone.now() userid = options['userid'] cpfit = options['cpfit'] cpoverlay = options['cpoverlay'] u = User.objects.get(id=userid) r = u.rower delta, cpvalue, avgpower, workoutnames, urls = dataprep.fetchcp_new( r, workouts) powerdf = pl.DataFrame({ 'Delta': delta, 'CP': cpvalue, 'workout': workoutnames, 'url': urls, }) savedata = options.get('savedata',False) if savedata: # pragma: no cover return powerdf.to_pandas() if powerdf.is_empty(): # pragma: no cover return('', '

No valid data found

') powerdf = powerdf.lazy().filter(pl.col("CP")>0) powerdf = powerdf.sort(["Delta", "CP"], descending=[False, True]) powerdf = powerdf.unique(subset="Delta", keep="first") powerdf = powerdf.fill_nan(None).drop_nulls() powerdf = powerdf.collect() rowername = r.user.first_name+" "+r.user.last_name wcdurations = [] wcpower = [] if len(powerdf) != 0: datefirst = pd.Series(w.date for w in workouts).min() datelast = pd.Series(w.date for w in workouts).max() title = 'CP chart for {name}, from {d1} to {d2}'.format( name=rowername, d1=datefirst, d2=datelast, ) wtype = 'water' if workouts[0].workouttype in mytypes.otetypes: # pragma: no cover wtype = 'erg' if workouts[0].workouttype == 'bikeerg': # pragma: no cover # for Mike wtype = 'erg' if cpoverlay: if r.birthdate: age = calculate_age(r.birthdate) else: # pragma: no cover age = 0 agerecords = CalcAgePerformance.objects.filter( age=age, sex=r.sex, weightcategory=r.weightcategory ) if len(agerecords) == 0: wcpower = [] wcdurations = [] else: # pragma: no cover wcdurations = [] wcpower = [] for record in agerecords: recordpower = record.power if wtype == 'water': recordpower = record.power*(100.-r.otwslack)/100. wcdurations.append(record.duration) wcpower.append(recordpower) res = interactive_otwcpchart(powerdf, promember=True, rowername=rowername, r=r, cpfit=cpfit, title=title, type=wtype, cpoverlay=cpoverlay, wcdurations=wcdurations, wcpower=wcpower) script = res[0] div = res[1] p1 = res[2] ratio = res[3] else: # pragma: no cover script = '' div = '

No ranking pieces found.

' p1 = [1, 1, 1, 1] ratio = 1 minutes = options['piece'] if minutes != 0: # minutes = 77 try: hourvalue, tvalue = divmod(minutes, 60) except: # pragma: no cover hourvalue = 0 tvalue = minutes # hourvalue = 1, tvalue = 17 try: hourvalue = int(hourvalue) except TypeError: # pragma: no cover hourvalue = 0 try: minutevalue = int(tvalue) except TypeError: # pragma: no cover minutevalue = 0 tvalue = int(60*(tvalue-minutevalue)) if hourvalue >= 24: # pragma: no cover hourvalue = 23 pieceduration = datetime.time( minute=minutevalue, hour=hourvalue, second=tvalue, ) pieceseconds = 3600.*pieceduration.hour+60. * \ pieceduration.minute+pieceduration.second # CP model pwr = p1[0]/(1+pieceseconds/p1[2]) pwr += p1[1]/(1+pieceseconds/p1[3]) if pwr <= 0: # pragma: no cover pwr = 50. if not np.isnan(pwr): try: pwr2 = pwr*ratio except: # pragma: no cover pwr2 = pwr duration = timedeltaconv(pieceseconds) power = int(pwr) upper = int(pwr2) else: # pragma: no cover duration = timedeltaconv(0) power = 0 upper = 0 htmly = env.get_template('otwcp.html') html_content = htmly.render({ 'script': script, 'the_div': div, 'duration': duration, 'power': power, 'upper': upper, }) return (script, html_content) def statsdata(workouts, options): #try: # workouts = workouts.exclude(workoutsource='strava') #except AttributeError: # pragma: no cover # workouts = [w for w in workouts if w.workoutsource != 'strava'] includereststrokes = options['includereststrokes'] ids = options['ids'] userid = options['userid'] if userid == 0: # pragma: no cover extratitle = '' else: u = User.objects.get(id=userid) extratitle = ' '+u.first_name+' '+u.last_name workstrokesonly = not includereststrokes ids = [w.id for w in workouts] fieldlist, fielddict = dataprep.getstatsfields() # prepare data frame datadf, extracols = dataprep.read_cols_df_sql(ids, fieldlist) datadf = dataprep.clean_df_stats(datadf, workstrokesonly=workstrokesonly) try: datadf['pace'] = datadf['pace']/1000. except KeyError: # pragma: no cover pass # Create stats stats = {} # fielddict.pop('workoutstate') # fielddict.pop('workoutid') for field, verbosename in fielddict.items(): try: thedict = { 'mean': datadf[field].mean(), 'min': datadf[field].min(), 'std': datadf[field].std(), 'max': datadf[field].max(), 'median': datadf[field].median(), 'firstq': datadf[field].quantile(q=0.25), 'thirdq': datadf[field].quantile(q=0.75), 'verbosename': verbosename, } stats[field] = thedict except KeyError: # pragma: no cover pass # Create a dict with correlation values cor = datadf.corr(method='spearman') cor.fillna(value=0, inplace=True) cordict = {} for field1, verbosename1 in fielddict.items(): thedict = {} for field2, verbosename2 in fielddict.items(): try: thedict[verbosename2] = cor.loc[field1, field2] except KeyError: # pragma: no cover thedict[verbosename2] = 0 cordict[verbosename1] = thedict context = { 'stats': stats, 'cordict': cordict, 'extratitle': extratitle, } htmly = env.get_template('statsdiv.html') html_content = htmly.render(context) return('', html_content) def comparisondata(workouts, options): includereststrokes = options['includereststrokes'] xparam = options['xaxis'] yparam1 = options['yaxis1'] plottype = options['plottype'] promember = True workstrokesonly = not includereststrokes ids = [w.id for w in workouts] labeldict = { int(w.id): w.__str__() for w in workouts } savedata = options.get('savedata',False) if savedata: # pragma: no cover workstrokesonly = not includereststrokes columns = [xparam, yparam1, 'ftime', 'distance', 'fpace', 'power', 'hr', 'spm', 'time', 'pace', 'workoutstate', 'workoutid'] df = dataprep.read_data(columns, ids=ids, workstrokesonly=workstrokesonly, doclean=True, ) return df res = interactive_multiple_compare_chart(ids, xparam, yparam1, promember=promember, plottype=plottype, workstrokesonly=workstrokesonly, labeldict=labeldict) script = res[0] div = res[1] return(script, div) def boxplotdata(workouts, options): #try: # workouts = workouts.exclude(workoutsource='strava') #except AttributeError: # workouts = [w for w in workouts if w.workoutsource != 'strava'] includereststrokes = options['includereststrokes'] spmmin = options['spmmin'] spmmax = options['spmmax'] workmin = options['workmin'] workmax = options['workmax'] ids = options['ids'] userid = options['userid'] plotfield = options['plotfield'] workstrokesonly = not includereststrokes datemapping = { w.id: w.date for w in workouts } fieldlist, fielddict = dataprep.getstatsfields() fieldlist = [plotfield, 'workoutid', 'spm', 'driveenergy', 'workoutstate'] ids = [w.id for w in workouts] # prepare data frame datadf = dataprep.read_data(fieldlist, ids) datadf = dataprep.remove_nulls_pl(datadf) try: datadf = datadf.filter( pl.col("spm")>=spmmin, pl.col("spm")<=spmmax, pl.col("driveenergy")>=workmin, pl.col("driveenergy")<=workmax, ) except ColumnNotFoundError: pass try: datadf = datadf.with_columns((pl.col("workoutid").apply(lambda x: datemapping[x])).alias("date")) except ComputeError as e: return ("","No Valid Data") if userid == 0: # pragma: no cover extratitle = '' else: u = User.objects.get(id=userid) extratitle = ' '+u.first_name+' '+u.last_name savedata = options.get('savedata',False) if savedata: # pragma: no cover return datadf.to_pandas() script, div = interactive_boxchart(datadf, plotfield, extratitle=extratitle, spmmin=spmmin, spmmax=spmmax, workmin=workmin, workmax=workmax) return(script, div) @permission_required('rower.is_coach', fn=get_user_by_userid, raise_exception=True) def analysis_view_data(request, userid=0): is_ajax = request_is_ajax(request) if settings.TESTING: is_ajax = True if not is_ajax: # pragma: no cover url = reverse('analysis_new') return HttpResponseRedirect(url) if 'options' in request.session: options = request.session['options'] else: # pragma: no cover options = defaultoptions if userid == 0: userid = request.user.id workouts = [] try: ids = options['ids'] except KeyError: # pragma: no cover return JSONResponse({ "script": '', "div": 'No data found' }) function = options['function'] if not ids: # pragma: no cover return JSONResponse({ "script": '', "div": 'No data found' }) for id in ids: try: w = Workout.objects.get(id=id) #if w.workoutsource != 'strava': # workouts.append(w) workouts.append(w) except Workout.DoesNotExist: # pragma: no cover pass if function == 'boxplot': script, div = boxplotdata(workouts, options) elif function == 'trendflex': # pragma: no cover script, div = trendflexdata(workouts, options, userid=userid) elif function == 'histo': # pragma: no cover script, div = histodata(workouts, options) elif function == 'flexall': # pragma: no cover script, div = flexalldata(workouts, options) elif function == 'stats': # pragma: no cover script, div = statsdata(workouts, options) elif function == 'compare': # pragma: no cover script, div = comparisondata(workouts, options) elif function == 'cp': # pragma: no cover script, div = cpdata(workouts, options) else: # pragma: no cover script = '' div = 'Unknown analysis functions' return JSONResponse({ "script": script, "div": div, }) @login_required() @permission_required('rower.is_coach', fn=get_user_by_userid, raise_exception=True) def create_marker_workouts_view(request, userid=0, startdate=timezone.now()-timezone.timedelta(days=42), enddate=timezone.now()): therower = getrequestrower(request, userid=userid) theuser = therower.user workouts = Workout.objects.filter(user=theuser.rower, date__gte=startdate, date__lte=enddate, workouttype__in=mytypes.rowtypes, duplicate=False).order_by('date')#.exclude(workoutsource='strava') for workout in workouts: _ = dataprep.check_marker(workout) url = reverse('goldmedalscores_view', kwargs={ 'userid': request.user.id, }) return HttpResponseRedirect(url) @login_required() @permission_required('rower.is_coach', fn=get_user_by_userid, raise_exception=True) def goldmedalscores_view(request, userid=0, startdate=timezone.now()-timezone.timedelta(days=365), enddate=timezone.now()): is_ajax = request_is_ajax(request) therower = getrequestrower(request, userid=userid) theuser = therower.user if request.method == 'POST': form = DateRangeForm(request.POST) if form.is_valid(): startdate = form.cleaned_data['startdate'] enddate = form.cleaned_data['enddate'] if startdate > enddate: # pragma: no cover s = enddate enddate = startdate startdate = s else: form = DateRangeForm(initial={ 'startdate': startdate, 'enddate': enddate, }) script, div, ids = goldmedalscorechart( theuser, startdate=startdate, enddate=enddate, ) bestworkouts = Workout.objects.filter(id__in=ids).order_by('-date')#.exclude(workoutsource='strava') breadcrumbs = [ { 'url': '/rower/analysis', 'name': 'Analysis', }, { 'url': reverse(goldmedalscores_view), 'name': 'Gold Medal Scores' } ] if is_ajax: # pragma: no cover response = json.dumps({ 'script': script, 'div': div, }) return(HttpResponse(response, content_type='application/json')) return render(request, 'goldmedalscores.html', { 'rower': therower, 'active': 'nav-analysis', 'chartscript': script, 'breadcrumbs': breadcrumbs, 'the_div': div, 'form': form, 'bestworkouts': bestworkouts, }) @permission_required('rower.is_coach', fn=get_user_by_userid, raise_exception=True) def trainingzones_view(request, userid=0): r = getrequestrower(request, userid=userid) enddate = timezone.now() startdate = enddate-datetime.timedelta(days=42) zones = 'hr' date_agg = 'week' yaxis = 'percentage' form = TrainingZonesForm({ 'startdate': startdate, 'enddate': enddate, }) if request.method == 'POST': # pragma: no cover form = TrainingZonesForm(request.POST) if form.is_valid(): startdate = form.cleaned_data['startdate'] enddate = form.cleaned_data['enddate'] if date_agg == 'week': startdate = startdate - datetime.timedelta(days=startdate.weekday()) else: # pragma: no cover startdate = startdate - datetime.timedelta(days=(startdate.day-1)) form = TrainingZonesForm(initial={ 'startdate': startdate, 'enddate': enddate, 'zones': zones, 'dates': date_agg, 'yaxis': yaxis, }) script = '' div = get_call() breadcrumbs = [ { 'url': '/rowers/analysis', 'name': 'Analysis' }, { 'url': reverse('trainingzones_view'), 'name': 'Training Zones' } ] return render(request, 'trainingzones.html', { 'active': 'nav-analysis', 'breadcrumbs': breadcrumbs, 'rower': r, 'the_script': script, 'the_div': div, 'form': form, 'startdate': startdate, 'enddate': enddate, 'zones': zones, 'dates': date_agg, 'yaxis': yaxis, } ) @login_required() @permission_required('rower.is_coach', fn=get_user_by_userid, raise_exception=True) def trainingzones_view_data(request, userid=0): r = getrequestrower(request, userid=userid) startdate = timezone.now()-datetime.timedelta(days=365) enddate = timezone.now() zones = 'hr' date_agg = 'week' yaxis = 'percentage' zones = request.GET.get('zones', zones) date_agg = request.GET.get('dates', date_agg) yaxis = request.GET.get('yaxis', yaxis) if request.GET.get('startdate'): startdate = datetime.datetime.strptime( request.GET.get('startdate'), "%Y-%m-%d") startdate = arrow.get(startdate).datetime if request.GET.get('enddate'): enddate = datetime.datetime.strptime( request.GET.get('enddate'), "%Y-%m-%d") enddate = arrow.get(enddate).datetime if (enddate-startdate).days > 200: date_agg = "month" data = get_zones_report_pl(r, startdate, enddate, trainingzones=zones, date_agg=date_agg, yaxis=yaxis) script, div = interactive_zoneschart2( r, data, startdate, enddate, trainingzones=zones, date_agg=date_agg, yaxis=yaxis) return JSONResponse({ 'script': script, 'div': div, }) @permission_required('rower.is_coach', fn=get_user_by_userid, raise_exception=True) def performancemanager_view(request, userid=0, mode='rower', startdate=timezone.now()-timezone.timedelta(days=365), enddate=timezone.now()): is_ajax = request_is_ajax(request) therower = getrequestrower(request, userid=userid) theuser = therower.user kfitness = therower.kfit kfatigue = therower.kfatigue metricchoice = 'hrtss' doform = therower.showfresh dofatigue = therower.showfit script = '' thediv = 'Submit form to create chart' endfitness = 0 endfatigue = 0 endform = 0 ids = [] if request.method == 'POST': form = PerformanceManagerForm(request.POST) if form.is_valid(): startdate = form.cleaned_data['startdate'] enddate = form.cleaned_data['enddate'] metricchoice = form.cleaned_data['metricchoice'] dofatigue = form.cleaned_data['dofatigue'] doform = form.cleaned_data['doform'] therower.showfresh = doform therower.showfatigue = dofatigue therower.save() script, thediv, endfitness, endfatigue, endform, ids = performance_chart( theuser, startdate=startdate, enddate=enddate, kfitness=kfitness, kfatigue=kfatigue, metricchoice=metricchoice, doform=doform, dofatigue=dofatigue, showtests=True, ) else: markerworkouts = Workout.objects.filter( user = therower, date__gte=startdate-datetime.timedelta(days=90), date__lte=enddate, duplicate=False, rankingpiece=True, workouttype__in=mytypes.rowtypes).order_by('date')#.exclude(workoutsource='strava') ids = [w.id for w in markerworkouts] form = PerformanceManagerForm(initial={ 'doform': doform, 'dofatigue': dofatigue, }) ids = pd.Series(ids, dtype='int').dropna().values bestworkouts = Workout.objects.filter(id__in=ids).order_by('-date')#.exclude(workoutsource='strava') breadcrumbs = [ { 'url': '/rowers/analysis', 'name': 'Analysis' }, { 'url': reverse('performancemanager_view'), 'name': 'Performance Manager' } ] if is_ajax: # pragma: no cover response = json.dumps({ 'script': script, 'div': thediv, 'endform': int(endform), 'endfitness': int(endfitness), 'endfatigue': int(endfatigue), }) return(HttpResponse(response, content_type='application/json')) return render(request, 'performancemanager.html', { 'rower': therower, 'active': 'nav-analysis', 'chartscript': script, 'breadcrumbs': breadcrumbs, 'the_div': thediv, 'mode': mode, 'form': form, 'endfitness': int(endfitness), 'endfatigue': int(endfatigue), 'endform': int(endform), 'bestworkouts': bestworkouts, }) @login_required() def ajax_agegrouprecords(request, age=25, sex='female', weightcategory='hwt', userid=0): durations = [1, 4, 30, 60] distances = [100, 500, 1000, 2000, 5000, 6000, 10000, 21097, 42195] df = pd.DataFrame( list( C2WorldClassAgePerformance.objects.filter( sex=sex, weightcategory=weightcategory ).values() ) ) jsondf = df.to_json() job = myqueue(queue, handle_getagegrouprecords, jsondf, distances, durations, age, sex, weightcategory, ) return JSONResponse( { 'job': job.id } ) def agegrouprecordview(request, sex='male', weightcategory='hwt', distance=2000, duration=None): if not duration: df = pd.DataFrame( list( C2WorldClassAgePerformance.objects.filter( distance=distance, sex=sex, weightcategory=weightcategory ).values() ) ) else: duration = int(duration) * 60 df = pd.DataFrame( list( C2WorldClassAgePerformance.objects.filter( duration=duration, sex=sex, weightcategory=weightcategory ).values() ) ) script, div = interactive_agegroup_plot(df, sex=sex, distance=distance, duration=duration, weightcategory=weightcategory) return render(request, 'agegroupchart.html', { 'interactiveplot': script, 'active': 'nav-analysis', 'the_div': div, }) #instroke analysis delete view class ForceCurveAnalysisDelete(DeleteView): login_required = True model = ForceCurveAnalysis template_name = 'forcecurveanalysis_delete_confirm.html' # extra parameters def get_context_data(self, **kwargs): context = super(ForceCurveAnalysisDelete, self).get_context_data(**kwargs) if 'userid' in kwargs: # pragma: no cover userid = kwargs['userid'] else: userid = 0 context['rower'] = getrequestrower(self.request, userid=userid) context['alert'] = self.object breadcrumbs = [ { 'url': '/rowers/analysis', 'name': 'Analysis' }, { 'url': reverse('forcecurveanalysis_view'), 'name': 'Force Curve Analysis', }, { 'url': reverse('workout_forcecurve_view', kwargs={'userid': userid, 'id': encoder.encode_hex(self.object.workout.id), 'analysis': self.object.pk}), 'name': self.object.name, }, { 'url': reverse('forcecurve_analysis_delete_view', kwargs={'pk': self.object.pk}), 'name': 'Delete' } ] context['breadcrumbs'] = breadcrumbs return context def get_success_url(self): return reverse('forcecurveanalysis_view') def get_object(self, *args, **kwargs): obj = super(ForceCurveAnalysisDelete, self).get_object(*args, **kwargs) if obj.rower != self.request.user.rower: # pragma: no cover raise PermissionDenied("You are not allowed to delete this Analysis") return obj class SavedAnalysisView(UserPassesTestMixin, View): form_class = InStrokeMultipleCompareForm initial = {} selected = [] div = "" script = "" url = '/rowers/analysis/instrokeanalysis/' name = 'In-Stroke Analysis' breadcrumbs = [] analysis_class = InStrokeAnalysis template_name = 'instroke_analysis.html' r = None analyses = None chart = instroke_multi_interactive_chart def handle_no_permission(self): # pragma: no cover if self.request.user.is_anonymous: url = reverse('login')+'?next='+self.request.path return HttpResponseRedirect(url) messages.error(self.request, "This functionality requires a Pro plan or higher." " If you are already a Pro user, please log in to access this functionality") return HttpResponseRedirect("/rowers/paidplans") def test_func(self): return ispromember(self.request.user) def __init__(self, *args, **kwargs): self.userid = kwargs.get('userid',0) self.form_class = kwargs.get('form_class',self.form_class) self.initial = kwargs.get('initial', self.initial) self.selected = kwargs.get('selected', self.selected) self.name = kwargs.get('name', self.name) self.url = kwargs.get('url', self.url) self.analysis_class = kwargs.get('analysis_class', self.analysis_class) self.template_name = kwargs.get('template_name', self.template_name) self.chart = kwargs.get('chart', self.chart) self.breadcrumbs = breadcrumbs = [ { 'url': '/rowers/analysis', 'name': 'Analysis' }, { 'url': self.url, 'name': self.name, }, ] def setup(self, request, *args, **kwargs): super(SavedAnalysisView, self).setup(request, *args, **kwargs) self.r = getrequestrower(request, userid=self.userid) if self.r is not None and not is_coach_user(self.request.user, self.r.user): # pragma: no cover raise PermissionDenied("You are not allowed to see analyses for this user") self.analyses = self.analysis_class.objects.filter(rower=self.r).order_by("-date","-id") query = request.GET.get('q') startdate = request.GET.get('startdate') enddate = request.GET.get('enddate') self.searchform = SearchForm() if query: # pragma: no cover query_list = query.split() if self.analysis_class == InStrokeAnalysis: self.analyses = self.analyses.filter( reduce(operator.and_, (Q(name__icontains=q) for q in query_list)) | reduce(operator.and_, (Q(notes__icontains=q) for q in query_list)) | reduce(operator.and_, (Q(metric__icontains=q) for q in query_list)) | reduce(operator.and_, (Q(workout__name__icontains=q) for q in query_list)) ) else: self.analyses = self.analyses.filter( reduce(operator.and_, (Q(name__icontains=q) for q in query_list)) | reduce(operator.and_, (Q(notes__icontains=q) for q in query_list)) | reduce(operator.and_, (Q(workout__name__icontains=q) for q in query_list)) ) self.searchform = SearchForm(initial={'q': query}) date_initial = {} if startdate: # pragma: no cover try: self.analyses = self.analyses.filter(date__gte=startdate) date_initial['startdate'] = startdate except ValidationError: messages.error(request,"{startdate} is not a valid start date".format(startdate=startdate)) if enddate: # pragma: no cover self.analyses = self.analyses.filter(date__lte=enddate) date_initial['enddate'] = enddate self.dateform = DateRangeForm(initial=date_initial) def get(self, request, *args, **kwargs): return render(request, self.template_name, { 'breadcrumbs': self.breadcrumbs, 'analyses': self.analyses, 'rower': self.r, 'the_script': self.script, 'the_div': self.div, 'selected': self.selected, 'searchform': self.searchform, 'dateform': self.dateform, }) def post(self, request, *args, **kwargs): # pragma: no cover form = self.form_class(request.POST) if form.is_valid(): cd = form.cleaned_data self.selected = cd['analyses'] request.session['analyses'] = [a.id for a in self.selected] # now should redirect to analysis self.script, self.div = self.chart(self.selected) return render(request, self.template_name, { 'breadcrumbs': self.breadcrumbs, 'analyses': self.analyses, 'rower': self.r, 'the_script': self.script, 'the_div': self.div, 'selected': self.selected, 'searchform': self.searchform, 'dateform': self.dateform, }) #instroke analysis delete view class InStrokeAnalysisDelete(DeleteView): login_required = True model = InStrokeAnalysis template_name = 'instrokeanalysis_delete_confirm.html' # extra parameters def get_context_data(self, **kwargs): context = super(InStrokeAnalysisDelete, self).get_context_data(**kwargs) if 'userid' in kwargs: # pragma: no cover userid = kwargs['userid'] else: userid = 0 context['rower'] = getrequestrower(self.request, userid=userid) context['alert'] = self.object breadcrumbs = [ { 'url': '/rowers/analysis', 'name': 'Analysis' }, { 'url': reverse('instrokeanalysis_view'), 'name': 'In-Stroke Analysis', }, { 'url': reverse('instroke_chart_interactive', kwargs={'userid': userid, 'id': encoder.encode_hex(self.object.workout.id), 'analysis': self.object.pk}), 'name': self.object.name, }, { 'url': reverse('instroke_analysis_delete_view', kwargs={'pk': self.object.pk}), 'name': 'Delete' } ] context['breadcrumbs'] = breadcrumbs return context def get_success_url(self): return reverse('instrokeanalysis_view') def get_object(self, *args, **kwargs): obj = super(InStrokeAnalysisDelete, self).get_object(*args, **kwargs) if obj.rower != self.request.user.rower: # pragma: no cover raise PermissionDenied("You are not allowed to delete this Analysis") return obj @login_required @permission_required('rower.is_coach', fn=get_user_by_userid, raise_exception=True) def alerts_view(request, userid=0): r = getrequestrower(request, userid=userid) alerts = Alert.objects.filter(rower=r).order_by('next_run') stats = [] for alert in alerts: # pragma: no cover stats.append(alert_get_stats(alert)) breadcrumbs = [ { 'url': '/rowers/analysis', 'name': 'Analysis' }, { 'url': reverse('alerts_view'), 'name': 'Alerts', }, ] return render(request, 'alerts.html', { 'breadcrumbs': breadcrumbs, 'alerts': alerts, 'rower': r, 'stats': stats, }) # alert create view @user_passes_test(ispromember, login_url="/rowers/paidplans", message="This functionality requires a Pro plan or higher." " If you are already a Pro user, please log in to access this functionality", redirect_field_name=None) @permission_required('rower.is_coach', fn=get_user_by_userid, raise_exception=True) def alert_create_view(request, userid=0): r = getrequestrower(request, userid=userid) FilterFormSet = formset_factory( ConditionEditForm, formset=BaseConditionFormSet, extra=1) filter_formset = FilterFormSet() if request.method == 'POST': form = AlertEditForm(request.POST) measuredform = ConditionEditForm(request.POST) filter_formset = FilterFormSet(request.POST) if form.is_valid() and measuredform.is_valid() and filter_formset.is_valid(): ad = form.cleaned_data measured = measuredform.cleaned_data period = ad['period'] emailalert = ad['emailalert'] reststrokes = ad['reststrokes'] workouttype = ad['workouttype'] boattype = ad['boattype'] name = ad['name'] filters = [] for filter_form in filter_formset: metric = filter_form.cleaned_data.get('metric') condition = filter_form.cleaned_data.get('condition') value1 = filter_form.cleaned_data.get('value1') value2 = filter_form.cleaned_data.get('value2') filters.append( { 'metric': metric, 'condition': condition, 'value1': value1, 'value2': value2, } ) result, message = create_alert(request.user, r, measured, period=period, emailalert=emailalert, reststrokes=reststrokes, workouttype=workouttype, boattype=boattype, filter=filters, name=name) if result: messages.info(request, message) url = reverse('alert_edit_view', kwargs={'id': result}) return HttpResponseRedirect(url) else: form = AlertEditForm() measuredform = ConditionEditForm() breadcrumbs = [ { 'url': '/rowers/analysis', 'name': 'Analysis' }, { 'url': reverse('alerts_view'), 'name': 'Alerts', }, { 'url': reverse('alert_create_view'), 'name': 'Create' } ] return render(request, 'alert_create.html', { 'breadcrumbs': breadcrumbs, 'formset': filter_formset, 'rower': r, 'form': form, 'measuredform': measuredform, }) # alert report view @login_required() @permission_required('rower.is_coach', fn=get_user_by_userid, raise_exception=True) def alert_report_view(request, id=0, userid=0, nperiod=0): r = getrequestrower(request, userid=userid) if userid == 0: # pragma: no cover userid = request.user.id alert = Alert.objects.get(id=id) nperiod = int(nperiod) try: alert = Alert.objects.get(id=id) except Alert.DoesNotExist: # pragma: no cover raise Http404("This alert doesn't exist") if not checkalertowner(alert, request.user): # pragma: no cover raise PermissionDenied('You are not allowed to edit this Alert') stats = alert_get_stats(alert, nperiod=nperiod) is_ajax = request_is_ajax(request) if is_ajax: # pragma: no cover return JSONResponse({ "stats": stats, }) breadcrumbs = [ { 'url': '/rowers/analysis', 'name': 'Analysis' }, { 'url': reverse('alerts_view'), 'name': 'Alerts', }, { 'url': reverse('alert_edit_view', kwargs={'userid': userid, 'id': alert.id}), 'name': alert.name, }, { 'url': reverse('alert_report_view', kwargs={'userid': userid, 'id': alert.id}), 'name': 'Report', }, ] # pragma: no cover return render(request, 'alert_stats.html', { 'breadcrumbs': breadcrumbs, 'stats': stats, 'rower': r, 'alert': alert, 'nperiod': nperiod, }) # pragma: no cover # alert edit view @user_passes_test(ispromember, login_url="/rowers/paidplans", message="This functionality requires a Pro plan or higher." " If you are already a Pro user, please log in to access this functionality", redirect_field_name=None) @permission_required('rower.is_coach', fn=get_user_by_userid, raise_exception=True) def alert_edit_view(request, id=0, userid=0): r = getrequestrower(request, userid=userid) try: alert = Alert.objects.get(id=id) except Alert.DoesNotExist: # pragma: no cover raise Http404("This alert doesn't exist") if alert.manager != request.user: # pragma: no cover raise PermissionDenied('You are not allowed to edit this Alert') FilterFormSet = formset_factory( ConditionEditForm, formset=BaseConditionFormSet, extra=0) if len(alert.filter.all()) == 0: # pragma: no cover FilterFormSet = formset_factory( ConditionEditForm, formset=BaseConditionFormSet, extra=1) filter_data = [{'metric': m.metric, 'value1': m.value1, 'value2': m.value2, 'condition': m.condition} for m in alert.filter.all()] if request.method == 'POST': form = AlertEditForm(request.POST) measuredform = ConditionEditForm(request.POST) filter_formset = FilterFormSet(request.POST) if form.is_valid() and measuredform.is_valid() and filter_formset.is_valid(): ad = form.cleaned_data measured = measuredform.cleaned_data period = ad['period'] emailalert = ad['emailalert'] reststrokes = ad['reststrokes'] workouttype = ad['workouttype'] boattype = ad['boattype'] name = ad['name'] m = alert.measured m.metric = measured['metric'] m.value1 = measured['value1'] m.value2 = measured['value2'] m.condition = measured['condition'] m.save() alert.period = period alert.emailalert = emailalert alert.reststrokes = reststrokes alert.workouttype = workouttype alert.boattype = boattype alert.name = name alert.save() filters = [] for filter_form in filter_formset: metric = filter_form.cleaned_data.get('metric') condition = filter_form.cleaned_data.get('condition') value1 = filter_form.cleaned_data.get('value1') value2 = filter_form.cleaned_data.get('value2') filters.append( { 'metric': metric, 'condition': condition, 'value1': value1, 'value2': value2, } ) _ = alert_add_filters(alert, filters) messages.info(request, 'Alert was changed') else: form = AlertEditForm(instance=alert) measuredform = ConditionEditForm(instance=alert.measured) filter_formset = FilterFormSet(initial=filter_data) breadcrumbs = [ { 'url': '/rowers/analysis', 'name': 'Analysis' }, { 'url': reverse('alerts_view'), 'name': 'Alerts', }, { 'url': reverse('alert_edit_view', kwargs={'userid': userid, 'id': alert.id}), 'name': alert.name, }, ] return render(request, 'alert_edit.html', { 'breadcrumbs': breadcrumbs, 'rower': r, 'form': form, 'measuredform': measuredform, 'formset': filter_formset, 'alert': alert, }) # alert delete view class AlertDelete(DeleteView): login_required = True model = Alert template_name = 'alert_delete_confirm.html' # extra parameters def get_context_data(self, **kwargs): context = super(AlertDelete, self).get_context_data(**kwargs) if 'userid' in kwargs: # pragma: no cover userid = kwargs['userid'] else: userid = 0 context['rower'] = getrequestrower(self.request, userid=userid) context['alert'] = self.object breadcrumbs = [ { 'url': '/rowers/analysis', 'name': 'Analysis' }, { 'url': reverse('alerts_view'), 'name': 'Alerts', }, { 'url': reverse('alert_edit_view', kwargs={'userid': userid, 'id': self.object.pk}), 'name': 'Alert', }, { 'url': reverse('alert_delete_view', kwargs={'pk': self.object.pk}), 'name': 'Delete' } ] context['breadcrumbs'] = breadcrumbs return context def get_success_url(self): return reverse('alerts_view') def get_object(self, *args, **kwargs): obj = super(AlertDelete, self).get_object(*args, **kwargs) if obj.manager != self.request.user: # pragma: no cover raise PermissionDenied("You are not allowed to delete this Alert") # some checks return obj @permission_required('rower.is_coach', fn=get_user_by_userid, raise_exception=True) def history_view(request, userid=0): r = getrequestrower(request, userid=userid) usertimezone = pytz.timezone(r.defaulttimezone) time_min = datetime.time(hour=0, minute=0, second=0) time_max = datetime.time(hour=23, minute=59, second=59) if request.GET.get('startdate'): startdate, enddate = get_dates_timeperiod(request) sstartdate = startdate senddate = enddate activity_startdate = usertimezone.localize( timezone.datetime.combine(startdate, time_min)) activity_enddate = usertimezone.localize( timezone.datetime.combine(enddate, time_max)) else: activity_enddate = timezone.now() activity_enddate = usertimezone.localize( timezone.datetime.combine(activity_enddate.date(), time_max)) startdate = timezone.now() - datetime.timedelta(days=14) activity_startdate = usertimezone.localize( timezone.datetime.combine(startdate.date(), time_min)) sstartdate = activity_startdate.date senddate = activity_enddate.date startdate = sstartdate enddate = senddate typeselect = 'All' if request.GET.get('workouttype'): typeselect = request.GET.get('workouttype') yaxis = request.GET.get('yaxis', 'duration') if typeselect not in mytypes.checktypes: typeselect = 'All' form = HistorySelectForm(initial={'startdate': activity_startdate, 'enddate': activity_enddate, 'workouttype': typeselect, 'yaxis': yaxis}) g_workouts = Workout.objects.filter( user=r, startdatetime__gte=activity_startdate, startdatetime__lte=activity_enddate, duplicate=False, privacy='visible' ).order_by("-startdatetime") tscript, tdiv = interactive_workouttype_piechart(g_workouts) totalmeters, totalhours, totalminutes, totalseconds = get_totals( g_workouts) # meters, duration per workout type wtypes = list(set([w.workouttype for w in g_workouts])) typechoices = [("All", "All")] for wtype in wtypes: if wtype in mytypes.checktypes: typechoices.append((wtype, mytypes.workouttypes_ordered[wtype])) form.fields['workouttype'].choices = typechoices listofdicts = [] for wtype in wtypes: a_workouts = g_workouts.filter(workouttype=wtype) wmeters, whours, wminutes, wseconds = get_totals(a_workouts) ddict = {} ddict['id'] = wtype try: ddict['wtype'] = mytypes.workouttypes_ordered[wtype] except KeyError: # pragma: no cover ddict['wtype'] = wtype ddict['distance'] = wmeters ddict['duration'] = "{whours}:{wminutes:02d}:{wseconds:02d}".format( whours=whours, wminutes=wminutes, wseconds=wseconds, ) ddict['distance'] = wmeters ddict['nrworkouts'] = a_workouts.count() listofdicts.append(ddict) # interactive hr pie chart totalscript = '' totaldiv = get_call() totalsdict = {} totalsdict['duration'] = "{totalhours:02d}:{totalminutes:02d}:{totalseconds:02d}".format( totalhours=totalhours, totalminutes=totalminutes, totalseconds=totalseconds, ) totalsdict['distance'] = totalmeters totalsdict['nrworkouts'] = g_workouts.count() breadcrumbs = [ { 'url': '/rowers/analysis', 'name': 'Analysis', }, { 'url': reverse('history_view'), 'name': 'History', }, ] lastseven = timezone.now() - datetime.timedelta(days=7) lastfourteen = timezone.now() - datetime.timedelta(days=14) last28 = timezone.now() - datetime.timedelta(days=28) today = timezone.now() lastyear = today - datetime.timedelta(days=365) #lastyear = datetime.datetime( # year=today.year - 1, month=today.month, day=today.day) firstmay = datetime.datetime( year=today.year, month=5, day=1).astimezone(usertimezone) if firstmay > today: # pragma: no cover firstmay = datetime.datetime(year=today.year - 1, month=5, day=1) return render(request, 'history.html', { 'tscript': tscript, 'tdiv': tdiv, 'rower': r, 'breadcrumbs': breadcrumbs, 'active': 'nav-analysis', 'totalsdict': totalsdict, 'typedicts': listofdicts, 'totalscript': totalscript, 'totaldiv': totaldiv, 'form': form, 'startdate': activity_startdate, 'enddate': activity_enddate, 'lastseven': lastseven, 'lastfourteen': lastfourteen, 'last28': last28, 'lastyear': lastyear, 'today': today, 'workouttype': typeselect, 'yaxis': yaxis, 'firstmay': firstmay, 'sstartdate': sstartdate, 'senddate': senddate, }) @login_required() def history_view_data(request, userid=0): r = getrequestrower(request, userid=userid) usertimezone = pytz.timezone(r.defaulttimezone) time_min = datetime.time(hour=0, minute=0, second=0) time_max = datetime.time(hour=23, minute=59, second=59) startdate = timezone.now() - datetime.timedelta(days=14) enddate = timezone.now() activity_enddate = usertimezone.localize( timezone.datetime.combine(enddate.date(), time_max)) activity_startdate = usertimezone.localize( timezone.datetime.combine(startdate.date(), time_min)) if request.GET.get('startdate'): startdate = datetime.datetime.strptime( request.GET.get('startdate'), "%Y-%m-%d") activity_startdate = usertimezone.localize( timezone.datetime.combine(startdate, time_min)) if request.GET.get('enddate'): enddate = datetime.datetime.strptime( request.GET.get('enddate'), "%Y-%m-%d") activity_enddate = usertimezone.localize( timezone.datetime.combine(enddate, time_max)) typeselect = 'All' if request.GET.get('workouttype'): typeselect = request.GET.get('workouttype') if typeselect not in mytypes.checktypes: typeselect = 'All' yaxis = request.GET.get('yaxis', 'duration') if yaxis.lower() not in ['duration', 'rscore', 'trimp', 'distance']: # pragma: no cover yaxis = 'duration' g_workouts = Workout.objects.filter( user=r, startdatetime__gte=activity_startdate, startdatetime__lte=activity_enddate, duplicate=False, privacy='visible' ).order_by("-startdatetime") ids = [w.id for w in g_workouts] columns = ['hr', 'power', 'time', 'workoutstate', 'workoutid'] df = dataprep.read_data(columns, ids=ids) df = dataprep.remove_nulls_pl(df) try: df = df.with_columns(pl.col('time').diff().clip(lower_bound=0).alias("deltat")) except KeyError: # pragma: no cover pass except ColumnNotFoundError: pass totalmeters, totalhours, totalminutes, totalseconds = get_totals( g_workouts) # meters, duration per workout type wtypes = list(set([w.workouttype for w in g_workouts])) typechoices = [("All", "All")] for wtype in wtypes: if wtype in mytypes.checktypes: typechoices.append((wtype, mytypes.workouttypes_ordered[wtype])) listofdicts = [] for wtype in wtypes: a_workouts = g_workouts.filter(workouttype=wtype) wmeters, whours, wminutes, wseconds = get_totals(a_workouts) ddict = {} try: # pragma: no cover ddict['wtype'] = mytypes.workouttypes_ordered[wtype] except KeyError: # pragma: no cover ddict['wtype'] = wtype ddict['id'] = wtype ddict['distance'] = wmeters ddict['duration'] = "{whours}:{wminutes:02d}:{wseconds:02d}".format( whours=whours, wminutes=wminutes, wseconds=wseconds, ) ddf = dataprep.read_data(columns, ids=[w.id for w in a_workouts]) ddf = dataprep.remove_nulls_pl(ddf) try: ddf = ddf.with_columns(pl.col("time").diff().clip(lower_bound=0).alias("deltat")) except KeyError: # pragma: no cover pass except ColumnNotFoundError: pass ddf = dataprep.clean_df_stats_pl(ddf, workstrokesonly=False, ignoreadvanced=True) try: ddict['hrmean'] = int(wavg(ddf, 'hr', 'deltat')) except (KeyError, ValueError, AttributeError, ColumnNotFoundError): # pragma: no cover ddict['hrmean'] = 0 try: ddict['hrmax'] = int(ddf['hr'].max()) except (KeyError, ValueError, AttributeError, ColumnNotFoundError): # pragma: no cover ddict['hrmax'] = 0 except ColumnNotFoundError: ddict['hrmax'] = 0 ddict['powermean'] = int(wavg(ddf, 'power', 'deltat')) try: ddict['powermax'] = int(ddf['power'].max()) except (KeyError, ColumnNotFoundError): # pragma: no cover ddict['powermax'] = 0 ddict['nrworkouts'] = a_workouts.count() listofdicts.append(ddict) totalsdict = {} totalsdict['duration'] = "{totalhours}:{totalminutes}".format( totalhours=totalhours, totalminutes=totalminutes ) totalsdict['distance'] = totalmeters try: totalsdict['powermean'] = int(wavg(df, 'power', 'deltat')) totalsdict['powermax'] = int(df['power'].max()) except (KeyError, ColumnNotFoundError): # pragma: no cover totalsdict['powermean'] = 0 totalsdict['powermax'] = 0 try: totalsdict['hrmean'] = int(wavg(df, 'hr', 'deltat')) totalsdict['hrmax'] = int(df['hr'].max()) except (KeyError, ColumnNotFoundError): # pragma: no cover totalsdict['hrmean'] = 0 totalsdict['hrmax'] = 0 totalsdict['nrworkouts'] = g_workouts.count() # activity chart activity_script, activity_div = interactive_activitychart2( g_workouts, startdate, enddate, yaxis=yaxis) # interactive hr pie chart if typeselect == 'All': totalseconds = 3600 * totalhours + 60 * totalminutes + totalseconds totalscript, totaldiv = interactive_hr_piechart(df, r, 'All Workouts', totalseconds=totalseconds) else: a_workouts = g_workouts.filter(workouttype=typeselect) meters, hours, minutes, seconds = get_totals(a_workouts) totalseconds = 3600 * hours + 60 * minutes + seconds ddf = dataprep.read_data(columns, ids=[w.id for w in a_workouts]) ddf = dataprep.remove_nulls_pl(ddf) if ddf.is_empty(): totalscript = "" totaldiv = "No data" try: ddf = ddf.with_columns(pl.col("time").diff().clip(lower_bound=0).alias("deltat")) except KeyError: pass except ColumnNotFoundError: totalscript = "" totaldiv = "No data" ddf = dataprep.clean_df_stats_pl(ddf, workstrokesonly=True, ignoreadvanced=True) totalscript, totaldiv = interactive_hr_piechart( ddf, r, mytypes.workouttypes_ordered[typeselect], totalseconds=totalseconds) return JSONResponse({ 'script': totalscript, 'div': totaldiv, 'totalsdict': totalsdict, 'listofdicts': listofdicts, 'activities_script': activity_script, 'activities_chart': activity_div, })