Alexander/FitTrackee
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

API: add an activity (w/ gpx file)

Browse Source
This commit is contained in:
Sam 2018-05-01 17:51:38 +02:00
parent ba613079f4
commit f4f0d5a2aa
8 changed files with 319 additions and 16 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

86
mpwo_api/mpwo_api/activities/activities.py
View File

@ -1,7 +1,15 @@
from flask import Blueprint, jsonify
import json
import os
from datetime import timedelta
from ..users.utils import authenticate
from flask import Blueprint, current_app, jsonify, request
from mpwo_api import appLog, db
from sqlalchemy import exc
from werkzeug.utils import secure_filename
from ..users.utils import authenticate, verify_extension
from .models import Activity
from .utils import get_gpx_info
activities_blueprint = Blueprint('activities', __name__)
@ -29,3 +37,77 @@ def get_activities(auth_user_id):
}
}
return jsonify(response_object), 200
@activities_blueprint.route('/activities', methods=['POST'])
@authenticate
def post_activity(auth_user_id):
"""Post an activity"""
code = 400
response_object = verify_extension('activity', request)
if response_object['status'] != 'success':
return jsonify(response_object), code
activity_data = json.loads(request.form["data"])
if not activity_data or activity_data.get('sport_id') is None:
response_object = {
'status': 'error',
'message': 'Invalid payload.'
}
return jsonify(response_object), 400
activity_file = request.files['file']
filename = secure_filename(activity_file.filename)
dirpath = os.path.join(
current_app.config['UPLOAD_FOLDER'],
'activities',
str(auth_user_id)
)
if not os.path.exists(dirpath):
os.makedirs(dirpath)
filepath = os.path.join(dirpath, filename)
try:
activity_file.save(filepath)
gpx_data = get_gpx_info(filepath)
except Exception as e:
appLog.error(e)
response_object = {
'status': 'error',
'message': 'Error during activity file save.'
}
return jsonify(response_object), 500
try:
new_activity = Activity(
user_id=auth_user_id,
sport_id=activity_data.get('sport_id'),
activity_date=gpx_data['start'],
duration=timedelta(seconds=gpx_data['duration'])
)
new_activity.gpx = filepath
new_activity.pauses = timedelta(seconds=gpx_data['stop_time'])
new_activity.moving = timedelta(seconds=gpx_data['moving_time'])
new_activity.distance = gpx_data['distance']
new_activity.min_alt = gpx_data['elevation_min']
new_activity.max_alt = gpx_data['elevation_max']
new_activity.descent = gpx_data['downhill']
new_activity.ascent = gpx_data['uphill']
new_activity.max_speed = gpx_data['max_speed']
new_activity.ave_speed = gpx_data['average_speed']
db.session.commit()
response_object = {
'status': 'created',
'message': 'Activity added.'
}
return jsonify(response_object), 201
except (exc.IntegrityError, ValueError) as e:
db.session.rollback()
appLog.error(e)
response_object = {
'status': 'fail',
'message': 'Error during activity save.'
}
return jsonify(response_object), 500

49
mpwo_api/mpwo_api/activities/utils.py Normal file
View File

@ -0,0 +1,49 @@
import gpxpy.gpx
def get_gpx_info(gpx_file):
gpx_data = {'filename': gpx_file}
gpx_file = open(gpx_file, 'r')
gpx = gpxpy.parse(gpx_file)
max_speed = 0
start = 0
for track in gpx.tracks:
for segment in track.segments:
for point_idx, point in enumerate(segment.points):
if point_idx == 0:
start = point.time
speed = segment.get_speed(point_idx)
try:
if speed > max_speed:
max_speed = speed
except Exception:
pass
gpx_data['max_speed'] = (max_speed / 1000) * 3600
gpx_data['start'] = start
duration = gpx.get_duration()
gpx_data['duration'] = duration
ele = gpx.get_elevation_extremes()
gpx_data['elevation_max'] = ele.maximum
gpx_data['elevation_min'] = ele.minimum
hill = gpx.get_uphill_downhill()
gpx_data['uphill'] = hill.uphill
gpx_data['downhill'] = hill.downhill
mv = gpx.get_moving_data()
gpx_data['moving_time'] = mv.moving_time
gpx_data['stop_time'] = mv.stopped_time
distance = mv.moving_distance + mv.stopped_distance
gpx_data['distance'] = distance/1000
average_speed = distance / duration
gpx_data['average_speed'] = (average_speed / 1000) * 3600
return gpx_data

1
mpwo_api/mpwo_api/config.py
View File

@ -15,6 +15,7 @@ class BaseConfig:
current_app.root_path, 'uploads'
)
PICTURE_ALLOWED_EXTENSIONS = {'jpg', 'png', 'gif'}
ACTIVITY_ALLOWED_EXTENSIONS = {'gpx'}
class DevelopmentConfig(BaseConfig):

151
mpwo_api/mpwo_api/tests/test_activities.py
View File

@ -1,8 +1,30 @@
import datetime
import json
from io import BytesIO
from mpwo_api.tests.utils import add_activity, add_sport, add_user
gpx_file = (
'<?xml version=\'1.0\' encoding=\'UTF-8\'?>'
'<gpx xmlns:gpxdata="http://www.cluetrust.com/XML/GPXDATA/1/0" xmlns:gpxtpx="http://www.garmin.com/xmlschemas/TrackPointExtension/v1" xmlns:gpxext="http://www.garmin.com/xmlschemas/GpxExtensions/v3" xmlns="http://www.topografix.com/GPX/1/1">' # noqa
' <metadata/>'
' <trk>'
' <trkseg>'
' <trkpt lat="77.2261324" lon="-42.1223054">'
' <time>2015-09-20T13:48:44+00:00</time>'
' </trkpt>'
' <trkpt lat="77.2261324" lon="-42.1223054">'
' <time>2015-09-20T13:48:46+00:00</time>'
' <ele>223.28399658203125</ele>'
' </trkpt>'
' <trkpt lat="77.2261324" lon="-42.1223054">'
' <time>2015-09-20T13:48:46+00:00</time>'
' </trkpt>'
' </trkseg>'
' </trk>'
'</gpx>'
)
def test_get_all_activities(app):
add_user('test', 'test@test.com', '12345678')
@ -54,3 +76,132 @@ def test_get_all_activities(app):
assert 3600 == data['data']['activities'][0]['duration']
assert 1024 == data['data']['activities'][1]['duration']
def test_add_an_activity(app):
add_user('test', 'test@test.com', '12345678')
add_sport('cycling')
client = app.test_client()
resp_login = client.post(
'/api/auth/login',
data=json.dumps(dict(
email='test@test.com',
password='12345678'
)),
content_type='application/json'
)
response = client.post(
'/api/activities',
data=dict(
file=(BytesIO(str.encode(gpx_file)), 'example.gpx'),
data='{"sport_id": 1}'
),
headers=dict(
content_type='multipart/form-data',
Authorization='Bearer ' + json.loads(
resp_login.data.decode()
)['auth_token']
)
)
data = json.loads(response.data.decode())
assert response.status_code == 201
assert 'created' in data['status']
def test_add_an_activity_invalid_file(app):
add_user('test', 'test@test.com', '12345678')
add_sport('cycling')
client = app.test_client()
resp_login = client.post(
'/api/auth/login',
data=json.dumps(dict(
email='test@test.com',
password='12345678'
)),
content_type='application/json'
)
response = client.post(
'/api/activities',
data=dict(
file=(BytesIO(str.encode(gpx_file)), 'example.png'),
data='{"sport_id": 1}'
),
headers=dict(
content_type='multipart/form-data',
Authorization='Bearer ' + json.loads(
resp_login.data.decode()
)['auth_token']
)
)
data = json.loads(response.data.decode())
assert response.status_code == 400
assert data['status'] == 'fail'
assert data['message'] == 'File extension not allowed.'
def test_add_an_activity_no_sport_id(app):
add_user('test', 'test@test.com', '12345678')
add_sport('cycling')
client = app.test_client()
resp_login = client.post(
'/api/auth/login',
data=json.dumps(dict(
email='test@test.com',
password='12345678'
)),
content_type='application/json'
)
response = client.post(
'/api/activities',
data=dict(
file=(BytesIO(str.encode(gpx_file)), 'example.gpx'),
data='{}'
),
headers=dict(
content_type='multipart/form-data',
Authorization='Bearer ' + json.loads(
resp_login.data.decode()
)['auth_token']
)
)
data = json.loads(response.data.decode())
assert response.status_code == 400
assert data['status'] == 'error'
assert data['message'] == 'Invalid payload.'
def test_add_an_activity_no_file(app):
add_user('test', 'test@test.com', '12345678')
add_sport('cycling')
client = app.test_client()
resp_login = client.post(
'/api/auth/login',
data=json.dumps(dict(
email='test@test.com',
password='12345678'
)),
content_type='application/json'
)
response = client.post(
'/api/activities',
data=dict(
data='{}'
),
headers=dict(
content_type='multipart/form-data',
Authorization='Bearer ' + json.loads(
resp_login.data.decode()
)['auth_token']
)
)
data = json.loads(response.data.decode())
assert response.status_code == 400
assert data['status'] == 'fail'
assert data['message'] == 'No file part.'

1
mpwo_api/mpwo_api/tests/test_sports.py
View File

@ -1,4 +1,3 @@
import datetime
import json
from mpwo_api.tests.utils import add_admin, add_sport, add_user

17
mpwo_api/mpwo_api/users/auth.py
View File

@ -7,7 +7,7 @@ from sqlalchemy import exc, or_
from werkzeug.utils import secure_filename
from .models import User
from .utils import allowed_picture, authenticate, register_controls
from .utils import authenticate, register_controls, verify_extension
auth_blueprint = Blueprint('auth', __name__)
@ -236,20 +236,11 @@ def edit_user(user_id):
@authenticate
def edit_picture(user_id):
code = 400
if 'file' not in request.files:
response_object = {'status': 'fail', 'message': 'No file part.'}
return jsonify(response_object), code
file = request.files['file']
if file.filename == '':
response_object = {'status': 'fail', 'message': 'No selected file.'}
return jsonify(response_object), code
if not allowed_picture(file.filename):
response_object = {
'status': 'fail',
'message': 'File extension not allowed.'
}
response_object = verify_extension('picture', request)
if response_object['status'] != 'success':
return jsonify(response_object), code
file = request.files['file']
filename = secure_filename(file.filename)
dirpath = os.path.join(
current_app.config['UPLOAD_FOLDER'],

29
mpwo_api/mpwo_api/users/utils.py
View File

@ -6,12 +6,40 @@ from flask import current_app, jsonify, request
from .models import User
def allowed_activity(filename):
return '.' in filename and \
filename.rsplit('.', 1)[1].lower() in \
current_app.config.get('ACTIVITY_ALLOWED_EXTENSIONS')
def allowed_picture(filename):
return '.' in filename and \
filename.rsplit('.', 1)[1].lower() in \
current_app.config.get('PICTURE_ALLOWED_EXTENSIONS')
def verify_extension(file_type, req):
response_object = {'status': 'success'}
if 'file' not in req.files:
response_object = {'status': 'fail', 'message': 'No file part.'}
return response_object
file = req.files['file']
if file.filename == '':
response_object = {'status': 'fail', 'message': 'No selected file.'}
return response_object
if ((file_type == 'picture' and not allowed_picture(file.filename)) or
(file_type == 'activity' and not allowed_activity(file.filename))):
response_object = {
'status': 'fail',
'message': 'File extension not allowed.'
}
return response_object
def authenticate(f):
@wraps(f)
def decorated_function(*args, **kwargs):
@ -34,6 +62,7 @@ def authenticate(f):
if not user:
return jsonify(response_object), code
return f(resp, *args, **kwargs)
return decorated_function

1
mpwo_api/requirements.txt
View File

@ -13,6 +13,7 @@ Flask-Bcrypt==0.7.1
Flask-Migrate==2.1.1
Flask-SQLAlchemy==2.3.2
Flask-Testing==0.6.2
gpxpy==1.2.0
isort==4.2.15
itsdangerous==0.24
Jinja2==2.10