[trails] Add auto GPX and FIT file importing

vrobbler/apps/scrobbles/importers/trail_gpx.py

@@ -0,0 +1,178 @@
+import logging
+import os
+import tempfile
+from datetime import timezone as dt_timezone
+from typing import Optional
+
+from django.contrib.auth import get_user_model
+from django.core.files import File
+
+from locations.models import GeoLocation
+from scrobbles.models import Scrobble
+from trails.models import Trail
+
+logger = logging.getLogger(__name__)
+
+User = get_user_model()
+
+
+def parse_trackpoints(file_path):
+    _, ext = os.path.splitext(file_path)
+    ext = ext.lower()
+
+    if ext == ".gpx":
+        return _parse_gpx(file_path)
+    elif ext == ".fit":
+        return _parse_fit(file_path)
+    else:
+        raise ValueError(f"Unsupported file type: {ext}")
+
+
+def _parse_gpx(file_path):
+    import gpxpy
+
+    with open(file_path) as f:
+        gpx = gpxpy.parse(f)
+
+    track_name = None
+    points = []
+    for track in gpx.tracks:
+        if track.name and not track_name:
+            track_name = track.name
+        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]
+
+
+def _parse_fit(file_path):
+    import fitparse
+
+    fitfile = fitparse.FitFile(file_path)
+    messages = list(fitfile.get_messages("record"))
+
+    points = []
+    track_name = os.path.splitext(os.path.basename(file_path))[0]
+
+    for msg in messages:
+        lat = None
+        lon = None
+        ele = None
+        t = None
+        for field in msg:
+            if field.name == "position_lat":
+                lat = field.value * (180.0 / (2**31)) if field.value else None
+            elif field.name == "position_long":
+                lon = field.value * (180.0 / (2**31)) if field.value else None
+            elif field.name == "altitude":
+                ele = field.value
+            elif field.name == "timestamp":
+                t = field.value.replace(tzinfo=dt_timezone.utc) if field.value else None
+        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
+
+
+def convert_fit_to_gpx(file_path):
+    import gpxpy.gpx
+
+    points, track_name = _parse_fit(file_path)
+    gpx = gpxpy.gpx.GPX()
+    gpx_track = gpxpy.gpx.GPXTrack(name=track_name)
+    gpx.tracks.append(gpx_track)
+    gpx_seg = gpxpy.gpx.GPXTrackSegment()
+    gpx_track.segments.append(gpx_seg)
+
+    for lat, lon, ele, t in points:
+        gpx_seg.points.append(gpxpy.gpx.GPXTrackPoint(lat, lon, elevation=ele, time=t))
+
+    tmp = tempfile.NamedTemporaryFile(mode="w", suffix=".gpx", delete=False)
+    tmp.write(gpx.to_xml())
+    tmp.close()
+    return tmp.name
+
+
+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)
+
+    if not points:
+        logger.warning(f"No trackpoints found in {file_path}")
+        return []
+
+    first_lat, first_lon, _, first_time = points[0]
+    _, _, _, last_time = points[-1]
+
+    if first_time is None:
+        logger.warning(f"No timestamps in {file_path}")
+        return []
+
+    geo, _ = GeoLocation.objects.get_or_create(
+        lat=round(first_lat, 6),
+        lon=round(first_lon, 6),
+        defaults={"altitude": None},
+    )
+
+    trail = Trail.find_by_trailhead(first_lat, first_lon, tolerance_m=100)
+    if not trail:
+        trail = Trail.find_or_create(track_name)
+        trail.trailhead_location = geo
+        trail.save(update_fields=["trailhead_location"])
+
+    timestamp = first_time
+    stop_timestamp = last_time
+
+    existing = Scrobble.objects.filter(
+        timestamp=timestamp,
+        trail=trail,
+        user=user,
+    ).first()
+    if existing:
+        logger.debug(f"Skipping existing scrobble for trail {trail}")
+        return []
+
+    scrobble = Scrobble(
+        user=user,
+        timestamp=timestamp,
+        stop_timestamp=stop_timestamp,
+        source="GPX Import",
+        trail=trail,
+        log={},
+        played_to_completion=True,
+        in_progress=False,
+        media_type=Scrobble.MediaType.TRAIL,
+    )
+
+    _, ext = os.path.splitext(file_path)
+    if ext.lower() == ".fit":
+        gpx_path = convert_fit_to_gpx(file_path)
+        with open(gpx_path, "rb") as f:
+            scrobble.gpx_file.save(
+                f"{original_filename or 'trail'}.gpx",
+                File(f),
+                save=False,
+            )
+        os.unlink(gpx_path)
+    else:
+        with open(file_path, "rb") as f:
+            scrobble.gpx_file.save(
+                original_filename or os.path.basename(file_path),
+                File(f),
+                save=False,
+            )
+
+    new_scrobbles.append(scrobble)
+
+    created = Scrobble.objects.bulk_create(new_scrobbles)
+    logger.info(f"Created {len(created)} trail scrobbles")
+    return created

vrobbler/apps/scrobbles/importers/webdav.py

@@ -1,9 +1,11 @@
 import logging
+import os
+import tempfile
 
 from books.koreader import fetch_file_from_webdav
 from profiles.models import UserProfile
-from scrobbles.models import KoReaderImport
-from scrobbles.tasks import process_koreader_import
+from scrobbles.models import KoReaderImport, TrailGPXImport
+from scrobbles.tasks import process_koreader_import, process_trail_gpx_import
 from scrobbles.utils import get_file_md5_hash
 from webdav.client import get_webdav_client
 
@@ -13,7 +15,6 @@ logger = logging.getLogger(__name__)
 def import_from_webdav_for_all_users(restart=False):
     """Grab a list of all users with WebDAV enabled and kickoff imports for them"""
 
-    # WebDavImport = apps.get_model("scrobbles", "WebDavImport")
     webdav_enabled_user_ids = UserProfile.objects.filter(
         webdav_url__isnull=False,
         webdav_user__isnull=False,
@@ -23,6 +24,7 @@ def import_from_webdav_for_all_users(restart=False):
     logger.info(f"Start import of {webdav_enabled_user_ids.count()} webdav accounts")
 
     koreader_import_count = 0
+    trail_gpx_import_count = 0
 
     for user_id in webdav_enabled_user_ids:
         webdav_client = get_webdav_client(user_id)
@@ -78,4 +80,54 @@ def import_from_webdav_for_all_users(restart=False):
 
             process_koreader_import.delay(koreader_import.id)
             koreader_import_count += 1
-    return koreader_import_count
+
+        trail_gpx_import_count += scan_webdav_for_gpx(webdav_client, user_id)
+
+    return koreader_import_count, trail_gpx_import_count
+
+
+def scan_webdav_for_gpx(webdav_client, user_id):
+    try:
+        webdav_client.info("gpx/")
+    except:
+        logger.info("No gpx/ directory on webdav", extra={"user_id": user_id})
+        return 0
+
+    try:
+        files = webdav_client.list("gpx/")
+    except Exception as e:
+        logger.warning("Could not list gpx/", extra={"user_id": user_id, "error": str(e)})
+        return 0
+
+    gpx_extensions = (".gpx", ".fit")
+    new_imports = 0
+    already_imported = set(
+        TrailGPXImport.objects.filter(user_id=user_id)
+        .values_list("original_filename", flat=True)
+    )
+
+    for fname in files:
+        fname = os.path.basename(fname)
+        if not fname.lower().endswith(gpx_extensions):
+            continue
+        if fname in already_imported:
+            logger.debug(f"Skipping already-imported {fname}")
+            continue
+
+        tmp = tempfile.NamedTemporaryFile(delete=False, suffix=fname)
+        try:
+            webdav_client.download_sync(remote_path=f"gpx/{fname}", local_path=tmp.name)
+            imp = TrailGPXImport.objects.create(
+                user_id=user_id,
+                original_filename=fname,
+            )
+            with open(tmp.name, "rb") as f:
+                imp.gpx_file.save(fname, f, save=True)
+            process_trail_gpx_import.delay(imp.id)
+            new_imports += 1
+        except Exception as e:
+            logger.error(f"Failed to import {fname}: {e}")
+        finally:
+            os.unlink(tmp.name)
+
+    return new_imports