From 7c33095d871b6375932807056d0e45b7414ee1ea Mon Sep 17 00:00:00 2001 From: Colin Powell Date: Thu, 21 May 2026 11:17:13 -0400 Subject: [PATCH] [trails] Pull data out of FIT/GPX files --- tests/trails_tests/test_gpx_importer.py | 60 ++++++- .../apps/scrobbles/importers/trail_gpx.py | 148 ++++++++++++++++-- vrobbler/apps/trails/models.py | 8 + 3 files changed, 198 insertions(+), 18 deletions(-) diff --git a/tests/trails_tests/test_gpx_importer.py b/tests/trails_tests/test_gpx_importer.py index 907968a..78b23e6 100644 --- a/tests/trails_tests/test_gpx_importer.py +++ b/tests/trails_tests/test_gpx_importer.py @@ -8,11 +8,12 @@ from django.core.files import File from locations.models import GeoLocation from scrobbles.importers.trail_gpx import ( + compute_trail_stats, import_trail_gpx, parse_trackpoints, ) from scrobbles.models import Scrobble, TrailGPXImport -from trails.models import Trail +from trails.models import Trail, TrailLogData User = get_user_model() @@ -33,9 +34,11 @@ def sample_gpx_path(): class TestParseTrackpoints: def test_parses_gpx(self, sample_gpx_path): - points, name = parse_trackpoints(sample_gpx_path) + result = parse_trackpoints(sample_gpx_path) + points = result["points"] assert len(points) == 837 - assert name == "Morning Run ⛅" + assert result["name"] == "Morning Run ⛅" + assert result["description"] == "Run" lat, lon, ele, t = points[0] assert round(lat, 6) == 34.190598 assert round(lon, 6) == -118.844015 @@ -43,12 +46,23 @@ class TestParseTrackpoints: assert t is not None def test_first_and_last_times(self, sample_gpx_path): - points, _ = parse_trackpoints(sample_gpx_path) + result = parse_trackpoints(sample_gpx_path) + points = result["points"] first_time = points[0][3] last_time = points[-1][3] duration = (last_time - first_time).total_seconds() assert duration == pytest.approx(3770, abs=5) + def test_gpx_extra_metadata(self, sample_gpx_path): + result = parse_trackpoints(sample_gpx_path) + extra = result["extra"] + assert extra["avg_heartrate"] == 159 + assert extra["max_heartrate"] == 183 + assert extra["avg_speed_kmh"] == pytest.approx(9.82, abs=0.1) + assert extra["activity_type"] == "Run" + assert extra["moving_time_seconds"] == 3008 + assert extra["total_elevation_gain_m"] == 246.4 + class TestImportTrailGPX: def test_creates_trail(self, user, sample_gpx_path): @@ -100,6 +114,44 @@ class TestImportTrailGPX: import_trail_gpx(sample_gpx_path, user.id) assert Scrobble.objects.filter(source="GPX Import").count() == 1 + def test_scrobble_log_has_stats(self, user, sample_gpx_path): + import_trail_gpx(sample_gpx_path, user.id) + scrobble = Scrobble.objects.filter(source="GPX Import").first() + log = scrobble.log + assert log["distance_km"] == pytest.approx(8.2, abs=0.2) + assert log["elevation_gain_m"] == pytest.approx(260, abs=20) + assert log["moving_time_seconds"] == pytest.approx(3770, abs=10) + assert log["avg_speed_kmh"] == pytest.approx(7.8, abs=0.5) + assert log["description"] == "Run" + + def test_scrobble_playback_position(self, user, sample_gpx_path): + import_trail_gpx(sample_gpx_path, user.id) + scrobble = Scrobble.objects.filter(source="GPX Import").first() + assert scrobble.playback_position_seconds == pytest.approx(3770, abs=5) + + def test_scrobble_log_extra_metadata(self, user, sample_gpx_path): + import_trail_gpx(sample_gpx_path, user.id) + scrobble = Scrobble.objects.filter(source="GPX Import").first() + log = scrobble.log + assert log["avg_heartrate"] == 159 + assert log["max_heartrate"] == 183 + assert log["activity_type"] == "Run" + + def test_scrobble_log_no_calories_in_gpx(self, user, sample_gpx_path): + import_trail_gpx(sample_gpx_path, user.id) + scrobble = Scrobble.objects.filter(source="GPX Import").first() + assert scrobble.log.get("calories") is None + + +class TestComputeTrailStats: + def test_computes_distance_and_elevation(self, sample_gpx_path): + result = parse_trackpoints(sample_gpx_path) + stats = compute_trail_stats(result["points"]) + assert stats["distance_km"] == pytest.approx(8.2, abs=0.2) + assert stats["elevation_gain_m"] == pytest.approx(260, abs=20) + assert stats["moving_time_seconds"] == pytest.approx(3770, abs=10) + assert stats["avg_speed_kmh"] == pytest.approx(7.8, abs=0.5) + class TestTrailGPXImportModel: def test_create_import_model(self, db, user, sample_gpx_path): diff --git a/vrobbler/apps/scrobbles/importers/trail_gpx.py b/vrobbler/apps/scrobbles/importers/trail_gpx.py index 1224d11..c62fbf4 100644 --- a/vrobbler/apps/scrobbles/importers/trail_gpx.py +++ b/vrobbler/apps/scrobbles/importers/trail_gpx.py @@ -1,7 +1,9 @@ +import json import logging import os import tempfile from datetime import timezone as dt_timezone +from math import asin, cos, radians, sin, sqrt from typing import Optional from django.contrib.auth import get_user_model @@ -9,13 +11,57 @@ from django.core.files import File from locations.models import GeoLocation from scrobbles.models import Scrobble -from trails.models import Trail +from trails.models import Trail, TrailLogData logger = logging.getLogger(__name__) User = get_user_model() +def haversine(lat1, lon1, lat2, lon2): + R = 6371000 + dlat = radians(lat2 - lat1) + dlon = radians(lon2 - lon1) + a = sin(dlat / 2) ** 2 + cos(radians(lat1)) * cos(radians(lat2)) * sin(dlon / 2) ** 2 + return R * 2 * asin(sqrt(a)) + + +def compute_trail_stats(points): + distance_m = 0.0 + elevation_gain = 0.0 + first_time = None + last_time = None + + for i in range(1, len(points)): + lat1, lon1, ele1, t1 = points[i - 1] + lat2, lon2, ele2, t2 = points[i] + + if lat1 is not None and lon1 is not None and lat2 is not None and lon2 is not None: + distance_m += haversine(lat1, lon1, lat2, lon2) + + if ele1 is not None and ele2 is not None and ele2 > ele1: + elevation_gain += ele2 - ele1 + + if first_time is None and t1 is not None: + first_time = t1 + if t2 is not None: + last_time = t2 + + moving_time = ( + (last_time - first_time).total_seconds() + if first_time and last_time + else 0 + ) + avg_speed_kmh = (distance_m / 1000) / (moving_time / 3600) if moving_time > 0 else 0.0 + + return { + "distance_km": round(distance_m / 1000, 2), + "elevation_gain_m": round(elevation_gain, 1), + "moving_time_seconds": int(moving_time), + "avg_speed_kmh": round(avg_speed_kmh, 2), + } + + def parse_trackpoints(file_path): _, ext = os.path.splitext(file_path) ext = ext.lower() @@ -34,15 +80,45 @@ def _parse_gpx(file_path): with open(file_path) as f: gpx = gpxpy.parse(f) - track_name = None points = [] + track_name = None + description = None + extra = {} + for track in gpx.tracks: if track.name and not track_name: track_name = track.name + if track.description and not description: + description = track.description + if track.comment and not extra: + try: + cmt_data = json.loads(track.comment) + data = cmt_data.get("activity_dict") or cmt_data + hr_avg = data.get("average_heartrate") + if hr_avg is not None: + extra["avg_heartrate"] = int(round(hr_avg)) + hr_max = data.get("max_heartrate") + if hr_max is not None: + extra["max_heartrate"] = int(round(hr_max)) + extra["calories"] = data.get("calories") + extra["moving_time_seconds"] = data.get("moving_time") + extra["total_elevation_gain_m"] = data.get("total_elevation_gain") + avg_speed = data.get("average_speed") + if avg_speed is not None: + extra["avg_speed_kmh"] = round(avg_speed * 3.6, 2) + extra["activity_type"] = cmt_data.get("type") or data.get("type") + except (json.JSONDecodeError, TypeError): + pass for seg in track.segments: for pt in seg.points: points.append((pt.latitude, pt.longitude, pt.elevation, pt.time)) - return points, track_name or os.path.splitext(os.path.basename(file_path))[0] + + return { + "points": points, + "name": track_name or os.path.splitext(os.path.basename(file_path))[0], + "description": description, + "extra": extra, + } def _parse_fit(file_path): @@ -53,6 +129,29 @@ def _parse_fit(file_path): points = [] track_name = os.path.splitext(os.path.basename(file_path))[0] + extra = {} + + session_msgs = list(fitfile.get_messages("session")) + for msg in session_msgs: + for field in msg: + if field.name == "sport" and field.value: + track_name = str(field.value) + elif field.name == "total_distance" and field.value: + extra["distance_km"] = round(field.value / 1000, 2) + elif field.name == "total_timer_time" and field.value: + extra["moving_time_seconds"] = int(field.value) + elif field.name == "total_elapsed_time" and field.value: + extra.setdefault("moving_time_seconds", int(field.value)) + elif field.name == "total_ascent" and field.value: + extra["total_elevation_gain_m"] = field.value + elif field.name == "avg_heart_rate" and field.value: + extra["avg_heartrate"] = int(round(field.value)) + elif field.name == "max_heart_rate" and field.value: + extra["max_heartrate"] = int(round(field.value)) + elif field.name == "avg_speed" and field.value: + extra["avg_speed_kmh"] = round(field.value * 3.6, 2) + elif field.name == "total_calories" and field.value: + extra["calories"] = int(field.value) for msg in messages: lat = None @@ -71,20 +170,20 @@ def _parse_fit(file_path): if lat is not None and lon is not None: points.append((lat, lon, ele, t)) - session_msgs = list(fitfile.get_messages("session")) - for msg in session_msgs: - for field in msg: - if field.name == "sport": - track_name = str(field.value) - break - - return points, track_name + return { + "points": points, + "name": track_name, + "description": None, + "extra": extra, + } def convert_fit_to_gpx(file_path): import gpxpy.gpx - points, track_name = _parse_fit(file_path) + result = _parse_fit(file_path) + points = result["points"] + track_name = result["name"] gpx = gpxpy.gpx.GPX() gpx_track = gpxpy.gpx.GPXTrack(name=track_name) gpx.tracks.append(gpx_track) @@ -104,7 +203,11 @@ def import_trail_gpx(file_path, user_id, original_filename=None): user = User.objects.get(id=user_id) new_scrobbles = [] - points, track_name = parse_trackpoints(file_path) + result = parse_trackpoints(file_path) + points = result["points"] + track_name = result["name"] + description = result.get("description") + extra = result.get("extra", {}) if not points: logger.warning(f"No trackpoints found in {file_path}") @@ -141,13 +244,30 @@ def import_trail_gpx(file_path, user_id, original_filename=None): logger.debug(f"Skipping existing scrobble for trail {trail}") return [] + computed = compute_trail_stats(points) + + logdata = TrailLogData( + description=description, + distance_km=computed["distance_km"], + elevation_gain_m=computed["elevation_gain_m"], + moving_time_seconds=computed["moving_time_seconds"], + activity_type=extra.get("activity_type"), + avg_speed_kmh=computed["avg_speed_kmh"], + avg_heartrate=extra.get("avg_heartrate"), + max_heartrate=extra.get("max_heartrate"), + calories=extra.get("calories"), + ) + + duration = int((stop_timestamp - timestamp).total_seconds()) if stop_timestamp else 0 + scrobble = Scrobble( user=user, timestamp=timestamp, stop_timestamp=stop_timestamp, + playback_position_seconds=duration, source="GPX Import", trail=trail, - log={}, + log=logdata.asdict, played_to_completion=True, in_progress=False, media_type=Scrobble.MediaType.TRAIL, diff --git a/vrobbler/apps/trails/models.py b/vrobbler/apps/trails/models.py index 9f4a90d..8827c78 100644 --- a/vrobbler/apps/trails/models.py +++ b/vrobbler/apps/trails/models.py @@ -25,6 +25,14 @@ def haversine(lat1, lon1, lat2, lon2): class TrailLogData(BaseLogData, WithPeopleLogData): effort: Optional[str] = None difficulty: Optional[str] = None + distance_km: Optional[float] = None + elevation_gain_m: Optional[float] = None + moving_time_seconds: Optional[int] = None + activity_type: Optional[str] = None + avg_speed_kmh: Optional[float] = None + avg_heartrate: Optional[int] = None + max_heartrate: Optional[int] = None + calories: Optional[int] = None class Trail(ScrobblableMixin):