adding update wps to upload routine

rowers/dataprep.py

@@ -28,7 +28,7 @@ from rowingdata import (
 
 from rowers.tasks import (
     handle_sendemail_unrecognized,handle_setcp,
-    handle_getagegrouprecords
+    handle_getagegrouprecords, handle_update_wps
     )
 from rowers.tasks import handle_zip_file
 
@@ -1263,6 +1263,44 @@ def setcp(workout,background=False,recurrance=True):
 
     return pd.DataFrame({'delta':[],'cp':[]}),pd.Series(dtype='float'),pd.Series(dtype='float')
 
+def update_wps(r,types,mode='water',asynchron=True):
+    firstdate = datetime.date.today()-datetime.timedelta(days=r.cprange)
+    workouts = Workout.objects.filter(
+        date__gte=firstdate,
+        workouttype__in=types,
+        user = r
+    )
+
+    ids = [w.id for w in workouts]
+    if asynchron:
+        job = myqueue(
+            queue,
+            handle_update_wps,
+            r.id,
+            types,
+            ids,
+            mode
+        )
+
+    df = getsmallrowdata_db(['time','driveenergy'],ids=ids)
+
+    try:
+        mask = df['driveenergy'] > 100
+    except KeyError:
+        return False
+    wps_median = int(df.loc[mask,'driveenergy'].median())
+
+    if mode == 'water':
+        r.median_wps = wps_median
+    else:
+        r.median_wps_erg = wps_median
+
+    r.save()
+
+    return True
+
+
+
 def update_rolling_cp(r,types,mode='water'):
     firstdate = datetime.date.today()-datetime.timedelta(days=r.cprange)
     workouts = Workout.objects.filter(
@@ -1801,6 +1839,8 @@ def save_workout_database(f2, r, dosmooth=True, workouttype='rower',
 
     isbreakthrough, ishard = checkbreakthrough(w, r)
     marker = check_marker(w)
+    result = update_wps(r,mytypes.otwtypes)
+    result = update_wps(r,mytypes.otetypes)
 
     job = myqueue(queuehigh,handle_calctrimp,w.id,f2,r.ftp,r.sex,r.hrftp,r.max,r.rest)
 
@@ -2436,7 +2476,7 @@ def getsmallrowdata_db(columns, ids=[], doclean=True,workstrokesonly=True,comput
                 #df = dd.read_parquet(f,columns=columns,engine='pyarrow')
                 df = pd.read_parquet(f,columns=columns)
                 data.append(df)
-            except OSError: # pragma: no cover
+            except (OSError,ArrowInvalid): # pragma: no cover
                 rowdata, row = getrowdata(id=id)
                 if rowdata and len(rowdata.df):
                     datadf = dataprep(rowdata.df,id=id,bands=True,otwpower=True,barchart=True)

rowers/tasks.py

@@ -2797,6 +2797,33 @@ def add2(x, y,debug=False,**kwargs): # pragma: no cover
 
 graphql_url = "https://rp3rowing-app.com/graphql"
 
+@app.task
+def handle_update_wps(rid,types,ids,mode,debug=False,**kwargs):
+    df = getsmallrowdata_db(['time','driveenergy'],ids=ids)
+    try:
+        mask = df['driveenergy'] > 100
+    except KeyError:
+        return 0
+    wps_median = int(df.loc[mask,'driveenergy'].median())
+
+    if mode == 'water':
+        query = "UPDATE `rowers_rower` SET `median_wps` = '%s' WHERE `id` = '%s'" % (wps_median,rid)
+    else:
+        query = "UPDATE `rowers_rower` SET `median_wps_erg` = '%s' WHERE `id` = '%s'" % (wps_median,rid)
+
+    if debug: # pragma: no cover
+        engine = create_engine(database_url_debug, echo=False)
+    else:
+        engine = create_engine(database_url, echo=False)
+
+    with engine.connect() as conn, conn.begin():
+        result = conn.execute(query)
+
+    conn.close()
+    engine.dispose()
+
+    return wps_median
+
 @app.task
 def handle_rp3_async_workout(userid,rp3token,rp3id,startdatetime,max_attempts,debug=False,**kwargs):
     headers = {'Authorization': 'Bearer ' + rp3token }