Merge pull request #111 from jedie/kml

Support KML track import created by Pentax K-1

Author: jedie

README.md

@@ -221,7 +221,12 @@ Because this is a project and not really a reuse-able-app ;)
 
 [comment]: <> (✂✂✂ auto generated history start ✂✂✂)
 
-* [**dev**](https://github.com/jedie/django-for-runners/compare/v0.17.4...main)
+* [v0.18.0.dev1](https://github.com/jedie/django-for-runners/compare/v0.17.4...v0.18.0.dev1)
+  * 2024-08-01 - Replace metaweather.com with open-meteo.com
+  * 2024-08-01 - Update test_add_gpx()
+  * 2024-08-01 - Update Leaflet to v1.9.4 and fix styles
+  * 2024-08-01 - Support KML track import created by Pentax K-1
+  * 2024-08-01 - Catch metaweather.com error
   * 2024-07-31 - Project updates
   * 2024-07-31 - Update requirements
   * 2024-01-18 - +typeguard +manageprojects updates

dev_scripts/download_leaflet.sh

@@ -12,7 +12,7 @@ cd ../for_runners/static/leaflet
 # download link from:
 # https://leafletjs.com/download.html
 #
-wget http://cdn.leafletjs.com/leaflet/v1.3.1/leaflet.zip
+wget http://cdn.leafletjs.com/leaflet/v1.9.4/leaflet.zip
 
 unzip -u leaflet.zip
 

for_runners/__init__.py

@@ -3,5 +3,5 @@
     Store your GPX tracks of your running (or other sports activity) in django.
 """
 
-__version__ = '0.17.4'
+__version__ = '0.18.0.dev1'
 __author__ = 'Jens Diemer <[email protected]>'

for_runners/app_settings.py

@@ -38,3 +38,6 @@
 # All cash values are stored as decimal field without any currency information
 # This symbol will be just added ;)
 FOR_RUNNERS_CURRENCY_SYMBOL = "€"
+
+# Use requests cache:
+REQUEST_CACHE = True

for_runners/geo.py

@@ -10,6 +10,8 @@
 from django.core.cache import cache
 from geopy.geocoders import Nominatim
 
+from for_runners.request_session import USER_AGENT
+
 
 log = logging.getLogger(__name__)
 
@@ -70,7 +72,7 @@ def reverse_geo(lat, lon) -> Optional[Address]:
         log.debug('reverse geo from cache')
         full_address, raw_address = address
     else:
-        geolocator = Nominatim(user_agent="django-for-runners")
+        geolocator = Nominatim(user_agent=USER_AGENT)
 
         # https://nominatim.org/release-docs/develop/api/Reverse/
         location = geolocator.reverse(
@@ -89,7 +91,7 @@ def reverse_geo(lat, lon) -> Optional[Address]:
         cache.set(cache_key, address, timeout=None)  # cache forever
 
     short_address = construct_short_address(address=raw_address)
-    log.info(f'short_address={short_address}')
+    log.info(f'lat={lat2} lon={lon2} {short_address=}')
 
     return Address(short_address, full_address)
 

for_runners/gpx.py

@@ -10,6 +10,7 @@
 
 import gpxpy
 from gpxpy.geo import distance as geo_distance
+from gpxpy.gpx import GPX
 
 # https://github.com/jedie/django-for-runners
 from for_runners.exceptions import GpxDataError
@@ -20,12 +21,18 @@
 )
 
 
-def get_identifier(gpxpy_instance):
+def get_identifier(gpxpy_instance: GPX) -> Identifier:
     """
     :return: Identifier named tuple
     """
     time_bounds = gpxpy_instance.get_time_bounds()
 
+    if not time_bounds.start_time:
+        raise GpxDataError("No start time found!")
+
+    if not time_bounds.end_time:
+        raise GpxDataError("No end time found!")
+
     try:
         first_track = gpxpy_instance.tracks[0]
     except IndexError:
@@ -106,21 +113,10 @@ def get_prefix_id(self):
         return result
 
 
-def parse_gpx(content):
-    # if 'creator="Garmin Connect"' in content:
-    #     work-a-round until
-    #     https://github.com/tkrajina/gpxpy/issues/115#issuecomment-392798245 fixed
-    #     return garmin2gpxpy(content)
-
-    return gpxpy.parse(content)
-
-
-def parse_gpx_file(filepath):
+def parse_gpx_file(filepath) -> GPX:
     assert filepath.is_file(), f"File not found: '{filepath}'"
-    with filepath.open("r") as f:
-        content = f.read()
-
-    return parse_gpx(content)
+    content = filepath.read_text()
+    return gpxpy.parse(content)
 
 
 def iter_points(gpxpy_instance):

for_runners/gpx_tools/__init__.py

@@ -0,0 +1,109 @@
+import dataclasses
+import datetime
+import logging
+import re
+
+import gpxpy.gpx
+from gpxpy.gpx import GPX, GPXTrackPoint
+from lxml import etree
+from lxml.etree import Element
+
+
+log = logging.getLogger(__name__)
+
+
+def get_single_text(node: Element, xpath: str, namespaces) -> str | None:
+    if elements := node.xpath(xpath, namespaces=namespaces):
+        assert len(elements) == 1, f'Expected 1 element, got {len(elements)}'
+        return elements[0].text
+
+
+@dataclasses.dataclass
+class Coordinates:
+    longitude: float
+    latitude: float
+    altitude: float
+
+
+def parse_coordinates(coordinates: str) -> Coordinates | None:
+    """
+    >>> parse_coordinates('2.444563,51.052540,8.0')
+    Coordinates(longitude=2.444563, latitude=51.05254, altitude=8.0)
+    """
+    match = re.match(r'\s*(-?\d+\.\d+)\s*,\s*(-?\d+\.\d+)\s*,\s*(-?\d+\.\d+)\s*', coordinates)
+    if match:
+        lon, lat, alt = map(float, match.groups())
+        return Coordinates(longitude=lon, latitude=lat, altitude=alt)
+
+
+def get_coordinates(placemark: Element, namespaces) -> Coordinates | None:
+    if coordinates := get_single_text(placemark, './/kml:coordinates', namespaces=namespaces):
+        return parse_coordinates(coordinates)
+
+
+def parse_datetime(datetime_str) -> datetime.datetime | None:
+    dt_part, tz_part = datetime_str.rsplit(' ', 1)
+
+    try:
+        dt = datetime.datetime.strptime(dt_part, '%Y/%m/%d %H:%M:%S')
+    except ValueError:
+        log.exception('Failed to parse datetime string %r', datetime_str)
+        return None
+
+    if not tz_part.startswith('UTC'):
+        log.warning('Timezone not in UTC format: %r', tz_part)
+        return None
+
+    sign = 1 if tz_part[3] == '+' else -1
+    try:
+        hours_offset = int(tz_part[4:6])
+        minutes_offset = int(tz_part[7:9])
+    except ValueError:
+        log.exception('Failed to parse timezone offset %r', tz_part)
+        return None
+
+    tz_offset = datetime.timedelta(hours=sign * hours_offset, minutes=sign * minutes_offset)
+    dt = dt.replace(tzinfo=datetime.timezone(tz_offset))
+    return dt
+
+
+def datetime_from_description(placemark: Element, namespaces):
+    if description := get_single_text(placemark, 'kml:description', namespaces=namespaces):
+        dt_str = description.partition('<br>')[0]
+        return parse_datetime(dt_str)
+
+
+def kml2gpx(kml_file) -> GPX:
+    """
+    Convert a KML file to a GPX object.
+    Notes:
+        * Only tested with KML files from a Pentax K-1 camera!
+    """
+    gpx = GPX()
+    track = gpxpy.gpx.GPXTrack()
+    gpx.tracks.append(track)
+
+    segment = gpxpy.gpx.GPXTrackSegment()
+    track.segments.append(segment)
+
+    root = etree.parse(kml_file).getroot()
+    namespaces = {'kml': root.nsmap.get(None)}
+
+    for placemark in root.xpath('//kml:Placemark', namespaces=namespaces):
+        dt = datetime_from_description(placemark, namespaces=namespaces)
+        if not dt:
+            continue
+
+        coordinates = get_coordinates(placemark, namespaces=namespaces)
+        if not coordinates:
+            continue
+
+        point = GPXTrackPoint(
+            latitude=coordinates.latitude,
+            longitude=coordinates.longitude,
+            elevation=coordinates.altitude,
+            time=dt,
+        )
+        segment.points.append(point)
+
+    return gpx

for_runners/gpx_tools/kml.py

@@ -0,0 +1,84 @@
+from datetime import datetime, timedelta, timezone
+
+from bx_py_utils.test_utils.datetime import parse_dt
+from django.test import SimpleTestCase
+from gpxpy.gpx import GPX
+
+from for_runners.gpx_tools.kml import Coordinates, kml2gpx, parse_coordinates, parse_datetime
+from for_runners.tests.fixture_files import get_fixture_path
+
+
+class KmlTestCase(SimpleTestCase):
+
+    def test_parse_coordinates(self):
+        self.assertEqual(
+            parse_coordinates('2.444563,51.052540,8.0'),
+            Coordinates(longitude=2.444563, latitude=51.05254, altitude=8.0),
+        )
+        self.assertEqual(
+            parse_coordinates('-2.444563,-51.052540,-8.0'),
+            Coordinates(longitude=-2.444563, latitude=-51.05254, altitude=-8.0),
+        )
+        self.assertEqual(
+            parse_coordinates('  2.444563 , 51.052540 , 8.0  '),
+            Coordinates(longitude=2.444563, latitude=51.05254, altitude=8.0),
+        )
+        self.assertIsNone(parse_coordinates('2.444563,51.052540'))
+        self.assertIsNone(parse_coordinates('abc,def,ghi'))
+
+    def test_parse_datetime(self):
+        self.assertEqual(
+            parse_datetime('2024/07/21 14:30:24 UTC+01:00'),
+            datetime(
+                2024,
+                7,
+                21,
+                14,
+                30,
+                24,
+                tzinfo=timezone(timedelta(hours=1)),
+            ),
+        )
+        self.assertEqual(
+            parse_datetime('2024/07/21 14:30:24 UTC-05:30'),
+            datetime(
+                2024,
+                7,
+                21,
+                14,
+                30,
+                24,
+                tzinfo=timezone(timedelta(hours=-5, minutes=-30)),
+            ),
+        )
+
+        with self.assertLogs('for_runners.gpx_tools.kml', 'WARNING') as cm:
+            self.assertIsNone(parse_datetime('2024/07/21 14:30:24'))
+            self.assertIn('Failed to parse datetime string', '\n'.join(cm.output))
+
+        with self.assertLogs('for_runners.gpx_tools.kml', 'WARNING') as cm:
+            self.assertIsNone(parse_datetime('21/07/2024 14:30:24 UTC+01:00'))
+            self.assertIn('Failed to parse datetime string', '\n'.join(cm.output))
+
+        with self.assertLogs('for_runners.gpx_tools.kml', 'WARNING') as cm:
+            self.assertIsNone(parse_datetime('2024/07/21 14:30:24 UTC+1:00'))
+            self.assertIn('Failed to parse timezone offset', '\n'.join(cm.output))
+
+    def test_kml2gpx(self):
+        fixture_path = get_fixture_path('PentaxK1.KML')
+
+        with self.assertLogs('for_runners.gpx_tools.kml', 'WARNING'):
+            gpx = kml2gpx(fixture_path)
+        self.assertIsInstance(gpx, GPX)
+
+        first_point = gpx.tracks[0].segments[0].points[0]
+        self.assertEqual(first_point.latitude, 51.052540)
+        self.assertEqual(first_point.longitude, 2.444563)
+        self.assertEqual(first_point.elevation, 8.0)
+        self.assertEqual(first_point.time, parse_dt('2024-07-21T14:30:24+01:00'))
+
+        last_point = gpx.tracks[0].segments[0].points[-1]
+        self.assertEqual(last_point.latitude, 50.944859)
+        self.assertEqual(last_point.longitude, 1.847900)
+        self.assertEqual(last_point.elevation, 14.0)
+        self.assertEqual(last_point.time, parse_dt('2024-07-21T21:28:31+01:00'))

for_runners/gpx_tools/tests/__init__.py

@@ -16,7 +16,7 @@
 
 
 class Command(BaseCommand):
-    help = "Import GPS files (*.gpx)"
+    help = "Import GPS track files (*.gpx|*.kml) for a user"
 
     def add_arguments(self, parser):
         parser.add_argument(
@@ -27,7 +27,7 @@ def add_arguments(self, parser):
             required=True,
             help="The user to assign to the imported files",
         )
-        parser.add_argument("path", help="Path to *.gpx files")
+        parser.add_argument("path", help="Path to *.gpx/*.kml files")
 
     def handle(self, *args, **options):
         username = options.get("username")
@@ -44,15 +44,20 @@ def handle(self, *args, **options):
         path = Path(options.get("path"))
         path = path.expanduser()
         path = path.resolve()
+
+        if path.is_file():
+            self.stderr.write(f"ERROR: Given path '{path}' is a file, but must be a directory that conains the files!")
+            sys.exit(4)
+
         if not path.is_dir():
             self.stderr.write(f"ERROR: Given path '{path}' is not a existing directory!")
-            sys.exit(4)
+            sys.exit(5)
 
         self.stdout.write(f"Read directory: {path}")
         self.stdout.write("\n")
 
         new_tracks = 0
-        for no, instance in enumerate(add_from_files(gpx_files_file_path=path, user=user), 1):
+        for no, instance in enumerate(add_from_files(tracks_path=path, user=user), 1):
             self.stdout.write(self.style.SUCCESS("%i - Add new track: %s" % (no, instance)))
             new_tracks += 1
 

for_runners/gpx_tools/tests/test_kml.py

@@ -7,12 +7,14 @@
 
 from django.db import models
 
+from for_runners.gpx import Identifier
+
 
 log = logging.getLogger(__name__)
 
 
 class GpxModelQuerySet(models.QuerySet):
-    def get_by_identifier(self, identifier):
+    def get_by_identifier(self, identifier: Identifier):
         """
         :param identifier: 'Identifier' namedtuple created here: for_runners.gpx.get_identifier
         """