Tool for converting CSV files exported from TSI LogView into GPX format.

tsicsv2gpx.py

#!/usr/bin/env python3

"""
This is a tool for converting CSV files exported from TSI LogView into GPX format.

While LogView can export to GPX natively, the produced file has several downsides:

- It uses GPX version 1.1, which cannot be imported to Garmin BaseCamp program.
- It lacks waypoints.
- It lacks some other recorded data like speed or heading.

LogView also supports KML export, which includes the waypoints, but it has issues of its own:

- Like the GPX export, it lacks speed data. But while BaseCamp can calculate it for GPX imports, it reportedly does not do it for KML.
- The waypoint names are very verbose.

CSV files appear to contain the most information (aside from maybe the bin export).

This was tested with GL-770 logger.
"""

from dataclasses import dataclass, replace
from datetime import datetime
from decimal import Decimal
from enum import Enum
from math import ceil, log10
from pathlib import Path
from xml.dom.minidom import Document, Element, getDOMImplementation
import argparse
import csv
import os

GPX_1_0_NS = "http://www.topografix.com/GPX/1/0"


class PointType(Enum):
    WayPoint = "wpt"
    TrackPoint = "trkpt"


class SourceType(Enum):
    Sps = "sps"
    DeadReckoning = "dr"
    NoFix = "none"


@dataclass
class Event:
    label: str
    utc_datetime: datetime
    latitude: Decimal
    longitude: Decimal
    elevation: Decimal
    speed: Decimal
    heading: Decimal
    source: SourceType


def fill_point_info(
    doc: Document,
    point: Element,
    event: Event,
    point_type: PointType,
) -> None:
    point.setAttribute("lat", str(event.latitude))
    point.setAttribute("lon", str(event.longitude))

    ele = doc.createElement("ele")
    ele.appendChild(doc.createTextNode(str(event.elevation)))
    point.appendChild(ele)

    time = doc.createElement("time")
    time.appendChild(doc.createTextNode(event.utc_datetime.isoformat()))
    point.appendChild(time)

    match point_type:
        case PointType.TrackPoint:
            course = doc.createElement("course")
            course.appendChild(doc.createTextNode(str(event.heading)))
            point.appendChild(course)

            speed = doc.createElement("speed")
            speed.appendChild(doc.createTextNode(str(event.speed)))
            point.appendChild(speed)

    match point_type:
        case PointType.WayPoint:
            name = doc.createElement("name")
            name.appendChild(doc.createTextNode(str(event.label)))
            point.appendChild(name)
        case PointType.TrackPoint:
            desc = doc.createElement("desc")
            desc.appendChild(doc.createTextNode(str(event.label)))
            point.appendChild(desc)

    match event.source:
        case SourceType.NoFix:
            fix = doc.createElement("fix")
            fix.appendChild(doc.createTextNode("none"))
            point.appendChild(fix)


def make_gpx(trackpoints: list[Event], waypoints: list[Event]) -> Document:
    dom = getDOMImplementation()
    doc = dom.createDocument(
        GPX_1_0_NS,
        "gpx",
        None,
    )

    gpx = doc.documentElement
    gpx.setAttribute("version", "1.0")
    gpx.setAttribute("xmlns", GPX_1_0_NS)
    gpx.setAttribute("creator", "tsicsv2gpx")
    gpx.setAttribute("xmlns:xsi", "http://www.w3.org/2001/XMLSchema-instance")
    gpx.setAttribute(
        "xsi:schemaLocation",
        "http://www.topografix.com/GPX/1/0 http://www.topografix.com/GPX/1/0/gpx.xsd",
    )

    waypoint_count = len(waypoints)
    waypoint_name_width = ceil(log10(waypoint_count)) if waypoint_count > 0 else 0
    for i, event in enumerate(waypoints, start=1):
        wpt = doc.createElement("wpt")
        gpx.appendChild(wpt)
        event = replace(event, label=str(i).zfill(waypoint_name_width))
        fill_point_info(doc, wpt, event, PointType.WayPoint)

    if trackpoints:
        trk = doc.createElement("trk")
        gpx.appendChild(trk)

        name = doc.createElement("name")
        name.appendChild(
            doc.createTextNode(trackpoints[0].utc_datetime.strftime("%Y_%m_%d"))
        )
        trk.appendChild(name)

        trkseg = doc.createElement("trkseg")
        trk.appendChild(trkseg)

        for event in trackpoints:
            trkpt = doc.createElement("trkpt")
            trkseg.appendChild(trkpt)

            fill_point_info(doc, trkpt, event, PointType.TrackPoint)

    return doc


def process(input_path: Path, force: bool) -> None:
    output_path = input_path.with_suffix(".gpx")
    if output_path.exists() and not force:
        raise RuntimeError(
            f"{output_path} already exists, use --force flag to overwrite it."
        )

    with open(input_path, newline="", encoding="utf-8-sig") as csvfile:
        reader = csv.DictReader(csvfile)

        trackpoints: list[Event] = []
        waypoints: list[Event] = []

        for record in reader:
            index = int(record["INDEX"])
            utc_datetime = datetime.strptime(
                record["UTC DATE"] + "T" + record["UTC TIME"] + "+00:00",
                "%Y/%m/%dT%H:%M:%S%z",
            )

            match record["VALID"]:
                case "SPS":
                    # Standard Positioning Service
                    source = SourceType.Sps
                case "Estimated (dead reckoning)":
                    source = SourceType.DeadReckoning
                case "NO FIX":
                    source = SourceType.NoFix
                case _:
                    raise RuntimeError(
                        f"Record {index} might not be valid, “{record['VALID']}” given."
                    )

            track_point = False
            button_point = False
            match record["RCR"]:
                case "T":
                    track_point = True
                case "TB":
                    track_point = True
                    button_point = True
                case "B":
                    button_point = True
                case rcr:
                    raise RuntimeError(
                        f"Unknown recording reason {rcr} for record {index}."
                    )

            latitude = Decimal(record["LATITUDE"])
            longitude = Decimal(record["LONGITUDE"])
            elevation = Decimal(record["HEIGHT"])
            speed = Decimal(record["SPEED"])
            heading = Decimal(record["HEADING"])

            match record["N/S"]:
                case "N":
                    pass
                case "S":
                    latitude *= -1
                case ns:
                    raise RuntimeError(f"Unknown N/S value {ns} for record {index}.")

            match record["E/W"]:
                case "E":
                    pass
                case "W":
                    longitude *= -1
                case ew:
                    raise RuntimeError(f"Unknown E/W value {ew} for record {index}.")

            event = Event(
                label=str(index),
                utc_datetime=utc_datetime,
                longitude=longitude,
                latitude=latitude,
                elevation=elevation,
                speed=speed,
                heading=heading,
                source=source,
            )

            if track_point:
                trackpoints.append(event)

            if button_point:
                waypoints.append(event)

    with open(output_path, "w") as gpxfile:
        gpx = make_gpx(trackpoints, waypoints)
        gpxfile.write(gpx.toprettyxml())


def main() -> None:
    parser = argparse.ArgumentParser(
        prog="tsicsv2gpx",
        description="Converts CSV file exported from TSV LogView to GPX file",
    )

    parser.add_argument(
        "path",
        type=Path,
        nargs="+",
    )
    parser.add_argument(
        "-f",
        "--force",
        action="store_true",
        help="Owerwrite GPX file if it already exists",
    )

    args = parser.parse_args()

    for path in args.path:
        process(path, args.force)


if __name__ == "__main__":
    main()