[trails] Pull data out of FIT/GPX files

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,