Script to merge a GPX file into a KML file.

import_gpx.py

"""Imports a GPX file into the canonical KML file."""

import os
import sys
import traceback
import gpxpy
import simplekml
from datetime import datetime, timezone
from dateutil.parser import parse
from lxml import etree

# This script will generate both a KML file (to act as the canonical
# storage for driving data in a human readible format) and a KMZ file
# (to have a smaller filesize for loading over a network). The KML file
# will be read when merging new data.
KML_PATH = "Driving.kml"
KMZ_PATH = "Driving.kmz"

def main(argv):
    """Merges a GPX file into existing KML data."""
    try:
        input_gpx_file = argv[1]
    except IndexError:
        raise SystemExit(f"Usage: {argv[0]} <tracks.gpx>")
    
    existing_tracks = kml_to_dict(KML_PATH)
    new_tracks = gpx_to_dict(input_gpx_file)
    
    # Merge existing tracks with gpx tracks. Existing tracks should
    # override new tracks with same timestamp.

    print("Merging new tracks into existing tracks …")
    merged = {**new_tracks, **existing_tracks}

    print("Creating KML/KMZ files …")

    print(f"{len(new_tracks)} new tracks")
    print(f"{len(merged)} merged tracks")

    kml = simplekml.Kml()
    kml.document.name = "Driving"

    track_style = simplekml.Style()
    track_style.linestyle.color = simplekml.Color.red
    track_style.linestyle.width = 4

    for timestamp, values in sorted(merged.items()):
        line = kml.newlinestring(
            name=timestamp.strftime("%Y-%m-%d %H:%M:%SZ"),
            tessellate=1,
            description=values['description'],
            coords=values['coords']
        )
        line.style = track_style
        line.timestamp = simplekml.TimeStamp(when=timestamp.isoformat())

    kml.save(KML_PATH)
    print(f"Saved KML to {KML_PATH}!")
    kml.savekmz(KMZ_PATH)
    print(f"Saved KMZ to {KMZ_PATH}!")


def kml_to_dict(kml_file):
    """
    Reads the supplied KML file and returns a dictionary with datetime
    keys and descriptions/coordinates (lists of (lon,lat,ele) tuples) as
    values.
    """
    print(f"Reading KML from {kml_file} …")

    track_dict = {}

    try:
        root = etree.parse(KML_PATH).getroot()
        nsmap = {None: "http://www.opengis.net/kml/2.2"}

        for e in root.findall("./Document/Placemark", nsmap):
            timestamp = parse(e.find("TimeStamp/when", nsmap).text)
            timestamp = timestamp.astimezone(timezone.utc)
            raw_coords = e.find("LineString/coordinates", nsmap).text.strip()
            coords = list(
                tuple(
                    float(n) for n in c.split(",")
                ) for c in raw_coords.split(" ")
            )
            desc = e.find("description", nsmap)
            if desc is not None:
                print("found description")
                desc = desc.text.strip()
            track_dict[timestamp] = dict(coords=coords, description=desc)

    except OSError:
        print(f"Could not open {KML_PATH}.")
        os.system("pause")

    print(f"Found {len(track_dict)} tracks!")
    return track_dict

def gpx_to_dict(gpx_file):
    """
    Reads the supplied GPX file and returns a dictionary with datetime
    keys and descriptions/coordinates (lists of (lon,lat,ele) tuples) as
    values.
    """
    print(f"Reading GPX from {gpx_file} …")

    gpx_file = open(gpx_file)
    gpx = gpxpy.parse(gpx_file)

    track_dict = {}

    for track in gpx.tracks:
        print(f"Converting {track.name}")
        desc = track.description
        for segment in track.segments:
            try:
                timestamp = segment.points[0].time.astimezone(timezone.utc)
            except AttributeError:
                timestamp = parse(track.name).astimezone(timezone.utc)

            coords = list(
                (p.longitude, p.latitude, p.elevation) for p in segment.points
            )
            track_dict[timestamp] = dict(coords=coords, description=desc)
    
    return track_dict


if __name__ == "__main__":
    try:
        main(sys.argv)
    except BaseException:
        print(sys.exc_info()[0])
        print(traceback.format_exc())
    finally:
        os.system("pause")