Validate GPX files, convert CSVs into GPX

app.py

import io
import csv
import math
import datetime
import xml.etree.ElementTree as ET

from shiny import reactive
from shiny.express import input, render, ui
from shinyswatch import theme
from shinywidgets import render_widget
import gpxpy
import gpxpy.gpx
from ipyleaflet import (
    Map, Marker, Polyline, AwesomeIcon, basemaps,
)
from ipywidgets import HTML as IPyHTML

# ---------------------------------------------------------------------------
# Sample data
# ---------------------------------------------------------------------------
SAMPLE_GPX = """\
<?xml version="1.0" encoding="UTF-8"?>
<gpx xmlns="http://www.topografix.com/GPX/1/1"
     xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
     xsi:schemaLocation="http://www.topografix.com/GPX/1/1
       http://www.topografix.com/GPX/1/1/gpx.xsd"
     version="1.1" creator="gpx.py -- https://github.com/tkrajina/gpxpy">
  <trk>
    <trkseg>
      <trkpt lat="-25.47294167" lon="32.11901833333334">
        <time>2026-03-18T10:37:10Z</time>
      </trkpt>
      <trkpt lat="-25.47291" lon="32.11902166666667">
        <time>2026-03-18T11:37:24Z</time>
      </trkpt>
    </trkseg>
  </trk>
</gpx>
"""

SAMPLE_CSV = """\
latitude,longitude,timestamp,name
-25.47294167,32.11901833,2026-03-18T10:37:10Z,Point A
-25.47291,32.11902167,2026-03-18T11:37:24Z,Point B
-25.47285,32.11910000,2026-03-18T12:15:00Z,Point C
"""

# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------

def haversine(lat1, lon1, lat2, lon2):
    R = 6371000
    phi1, phi2 = math.radians(lat1), math.radians(lat2)
    dphi = math.radians(lat2 - lat1)
    dlam = math.radians(lon2 - lon1)
    a = math.sin(dphi / 2) ** 2 + math.cos(phi1) * math.cos(phi2) * math.sin(dlam / 2) ** 2
    return R * 2 * math.atan2(math.sqrt(a), math.sqrt(1 - a))


def validate_gpx(xml_text: str) -> dict:
    result = {"valid": True, "messages": [], "gpx": None, "points": []}
    try:
        gpx = gpxpy.parse(xml_text)
        result["gpx"] = gpx
    except Exception as exc:
        result["valid"] = False
        result["messages"].append(f"Parse error: {exc}")
        return result

    try:
        root = ET.fromstring(xml_text)
        version = root.attrib.get("version", "unknown")
        result["messages"].append(f"GPX version: {version}")
        if version not in ("1.0", "1.1"):
            result["messages"].append(f"⚠ Unusual GPX version '{version}'")
    except Exception:
        pass

    points = []
    for track in gpx.tracks:
        for seg in track.segments:
            for pt in seg.points:
                points.append(pt)
    for wpt in gpx.waypoints:
        points.append(wpt)
    for route in gpx.routes:
        for pt in route.points:
            points.append(pt)

    if not points:
        result["valid"] = False
        result["messages"].append("No geographic points found in file.")
        return result

    result["messages"].append(f"Total points: {len(points)}")
    result["messages"].append(
        f"Tracks: {len(gpx.tracks)}, Routes: {len(gpx.routes)}, Waypoints: {len(gpx.waypoints)}"
    )

    for i, pt in enumerate(points):
        if not (-90 <= pt.latitude <= 90):
            result["valid"] = False
            result["messages"].append(f"Point {i}: latitude {pt.latitude} out of range [-90,90]")
        if not (-180 <= pt.longitude <= 180):
            result["valid"] = False
            result["messages"].append(f"Point {i}: longitude {pt.longitude} out of range [-180,180]")

    if len(points) >= 2:
        total_dist = 0
        for i in range(1, len(points)):
            total_dist += haversine(
                points[i - 1].latitude, points[i - 1].longitude,
                points[i].latitude, points[i].longitude,
            )
        result["messages"].append(f"Total distance: {total_dist:.1f} m ({total_dist / 1000:.3f} km)")

    times = [pt.time for pt in points if pt.time]
    if times:
        result["messages"].append(f"Time span: {min(times).isoformat()} → {max(times).isoformat()}")
        result["messages"].append(f"Duration: {max(times) - min(times)}")

    if result["valid"]:
        result["messages"].insert(0, "✅ GPX file is valid")
    else:
        result["messages"].insert(0, "❌ GPX file has errors")

    result["points"] = points
    return result


def csv_to_gpx(csv_text: str, lat_col: str = None, lon_col: str = None) -> str:
    reader = csv.DictReader(io.StringIO(csv_text))
    fields = reader.fieldnames or []

    time_col = name_col = ele_col = None
    _auto_lat = _auto_lon = None
    for f in fields:
        fl = f.strip().lower()
        if fl in ("lat", "latitude"):
            _auto_lat = f
        elif fl in ("lon", "lng", "longitude", "long"):
            _auto_lon = f
        elif fl in ("time", "timestamp", "datetime", "date"):
            time_col = f
        elif fl in ("name", "label", "title"):
            name_col = f
        elif fl in ("ele", "elevation", "alt", "altitude"):
            ele_col = f

    if not lat_col:
        lat_col = _auto_lat
    if not lon_col:
        lon_col = _auto_lon

    if not lat_col or not lon_col:
        raise ValueError(f"CSV must contain latitude and longitude columns. Found: {fields}")

    gpx = gpxpy.gpx.GPX()
    track = gpxpy.gpx.GPXTrack()
    gpx.tracks.append(track)
    seg = gpxpy.gpx.GPXTrackSegment()
    track.segments.append(seg)

    reader = csv.DictReader(io.StringIO(csv_text))
    for row in reader:
        lat = float(row[lat_col].strip())
        lon = float(row[lon_col].strip())
        pt = gpxpy.gpx.GPXTrackPoint(latitude=lat, longitude=lon)
        if time_col and row.get(time_col, "").strip():
            ts = row[time_col].strip()
            for fmt in ("%Y-%m-%dT%H:%M:%SZ", "%Y-%m-%dT%H:%M:%S", "%Y-%m-%d %H:%M:%S", "%Y-%m-%d"):
                try:
                    pt.time = datetime.datetime.strptime(ts, fmt).replace(tzinfo=datetime.timezone.utc)
                    break
                except ValueError:
                    continue
        if ele_col and row.get(ele_col, "").strip():
            try:
                pt.elevation = float(row[ele_col].strip())
            except ValueError:
                pass
        if name_col and row.get(name_col, "").strip():
            pt.name = row[name_col].strip()
        seg.points.append(pt)

    return gpx.to_xml()


def build_map(points, gpx_obj=None):
    """Build an ipyleaflet Map widget showing tracks as continuous paths.

    If gpx_obj is provided, each track segment gets its own colored polyline
    with start/end markers. Otherwise falls back to drawing a single path
    through all points.
    """
    TRACK_COLORS = [
        "#3b82f6", "#ef4444", "#10b981", "#f59e0b",
        "#8b5cf6", "#ec4899", "#06b6d4", "#f97316",
    ]

    all_coords = [(pt.latitude, pt.longitude) for pt in points]
    center = (
        sum(c[0] for c in all_coords) / len(all_coords),
        sum(c[1] for c in all_coords) / len(all_coords),
    )

    # Calculate zoom from bounding box
    lats = [c[0] for c in all_coords]
    lons = [c[1] for c in all_coords]
    spread = max(max(lats) - min(lats), max(lons) - min(lons))
    if spread < 0.001:
        zoom = 18
    elif spread < 0.01:
        zoom = 16
    elif spread < 0.1:
        zoom = 13
    elif spread < 1:
        zoom = 10
    elif spread < 10:
        zoom = 7
    else:
        zoom = 4

    m = Map(
        center=center,
        zoom=zoom,
        basemap=basemaps.OpenStreetMap.Mapnik,
        layout={"height": "440px"},
    )

    def _add_endpoint(pt, label, marker_color, icon_name):
        time_str = pt.time.isoformat() if pt.time else "—"
        popup_html = IPyHTML(
            value=(
                f"<div>"
                f"<b>{label}</b><br>"
                f"Lat: {pt.latitude:.6f}<br>"
                f"Lon: {pt.longitude:.6f}<br>"
                f"Time: {time_str}"
                f"</div>"
            )
        )
        icon = AwesomeIcon(name=icon_name, marker_color=marker_color, icon_color="white")
        marker = Marker(
            location=(pt.latitude, pt.longitude),
            icon=icon,
            title=label,
            popup=popup_html,
        )
        m.add(marker)

    # Build segments list: either from gpx_obj structure or a single flat list
    segments = []
    if gpx_obj:
        track_idx = 0
        for track in gpx_obj.tracks:
            for seg in track.segments:
                if seg.points:
                    segments.append((track_idx, seg.points))
            track_idx += 1
        for route in gpx_obj.routes:
            if route.points:
                segments.append((track_idx, route.points))
                track_idx += 1

    if not segments and points:
        segments = [(0, points)]

    for seg_idx, (track_idx, seg_points) in enumerate(segments):
        color = TRACK_COLORS[track_idx % len(TRACK_COLORS)]
        coords = [(pt.latitude, pt.longitude) for pt in seg_points]

        # Draw the continuous track line
        if len(coords) >= 2:
            line = Polyline(
                locations=coords,
                color=color,
                weight=4,
                opacity=0.85,
            )
            m.add(line)

        # Start marker
        track_label = f"Track {track_idx + 1}" if len(segments) > 1 else "Track"
        _add_endpoint(seg_points[0], f"{track_label} — Start", "green", "play")

        # End marker (only if more than one point)
        if len(seg_points) > 1:
            _add_endpoint(seg_points[-1], f"{track_label} — End", "red", "stop")

    return m


def read_uploaded(file_info):
    if file_info is None or len(file_info) == 0:
        return None
    path = file_info[0]["datapath"]
    with open(path, "r", encoding="utf-8", errors="replace") as f:
        return f.read()


# ---------------------------------------------------------------------------
# Reactive state
# ---------------------------------------------------------------------------
gpx_validation = reactive.Value(None)
csv_gpx_xml = reactive.Value(None)
csv_gpx_obj = reactive.Value(None)
csv_points = reactive.Value(None)

# ---------------------------------------------------------------------------
# Page options
# ---------------------------------------------------------------------------
ui.page_opts(title="gpxCheck", fillable=False, theme=theme.darkly)

# ---------------------------------------------------------------------------
# Tabs
# ---------------------------------------------------------------------------
with ui.navset_tab():

    # ===================== TAB 1: VALIDATE GPX =====================
    with ui.nav_panel("Validate GPX"):
        with ui.layout_columns(col_widths=[5, 7]):

            with ui.card():
                ui.card_header("GPX Source")
                ui.input_file("gpx_file", "Upload .gpx file", accept=[".gpx", ".xml"], multiple=False)
                ui.p("— or paste GPX XML below —", class_="text-muted small")
                ui.input_text_area("gpx_text", "", value=SAMPLE_GPX, rows=14)
                ui.input_action_button("validate_btn", "Validate", class_="btn-primary mt-2")

            with ui.div():
                with ui.card():
                    ui.card_header("Validation Results")

                    @render.ui
                    def validation_output():
                        res = gpx_validation.get()
                        if res is None:
                            return ui.p("Click Validate to check the GPX data.", class_="text-muted")
                        lines = []
                        for msg in res["messages"]:
                            if msg.startswith("✅"):
                                lines.append(f'<span class="text-success">{msg}</span>')
                            elif msg.startswith("❌") or "error" in msg.lower() or "out of range" in msg:
                                lines.append(f'<span class="text-danger">{msg}</span>')
                            elif "⚠" in msg:
                                lines.append(f'<span class="text-warning">{msg}</span>')
                            else:
                                lines.append(msg)
                        return ui.HTML(f'<pre class="small overflow-auto">{"<br>".join(lines)}</pre>')

                with ui.card():
                    ui.card_header("Map Preview")

                    @render_widget
                    def gpx_map():
                        res = gpx_validation.get()
                        if res is None or not res.get("points"):
                            return Map(center=(0, 0), zoom=2, layout={"height": "440px"})
                        return build_map(res["points"], gpx_obj=res.get("gpx"))

    # ===================== TAB 2: CSV → GPX =====================
    with ui.nav_panel("CSV ➜ GPX"):
        with ui.layout_columns(col_widths=[5, 7]):

            with ui.card():
                ui.card_header("CSV Source")
                ui.input_file("csv_file", "Upload .csv file", accept=[".csv", ".txt"], multiple=False)
                ui.p("— or paste CSV below —", class_="text-muted small")
                ui.input_text_area("csv_text", "", value=SAMPLE_CSV, rows=10)
                ui.p(
                    ui.HTML(
                        "Required columns: <b>latitude</b> (or lat) &amp; <b>longitude</b> (or lon/lng). "
                        "Optional: timestamp, name, elevation."
                    ),
                    class_="small text-muted",
                )
                with ui.layout_columns(col_widths=[6, 6]):
                    ui.input_select("csv_lat_col", "Latitude column", choices=[], size=None)
                    ui.input_select("csv_lon_col", "Longitude column", choices=[], size=None)
                ui.input_action_button("convert_btn", "Convert to GPX", class_="btn-primary mt-2")

            with ui.div():
                with ui.card():
                    ui.card_header("Generated GPX")

                    @render.ui
                    def csv_result():
                        xml = csv_gpx_xml.get()
                        if xml is None:
                            return ui.p("Click Convert to generate GPX from CSV.", class_="text-muted")
                        if xml.startswith("ERROR:"):
                            return ui.HTML(f'<pre class="small text-danger overflow-auto">{xml}</pre>')
                        escaped = xml.replace("&", "&amp;").replace("<", "&lt;").replace(">", "&gt;")
                        return ui.HTML(f'<pre class="small overflow-auto">{escaped}</pre>')

                    @render.download(filename="export.gpx", media_type="application/gpx+xml")
                    def download_gpx():
                        xml = csv_gpx_xml.get()
                        if xml and not xml.startswith("ERROR:"):
                            yield xml

                with ui.card():
                    ui.card_header("Map Preview")

                    @render_widget
                    def csv_map():
                        pts = csv_points.get()
                        if not pts:
                            return Map(center=(0, 0), zoom=2, layout={"height": "440px"})
                        return build_map(pts, gpx_obj=csv_gpx_obj.get())


# ---------------------------------------------------------------------------
# Reactive effects (button handlers)
# ---------------------------------------------------------------------------

@reactive.effect
@reactive.event(input.validate_btn)
def _do_validate():
    uploaded = read_uploaded(input.gpx_file())
    text = uploaded if uploaded else input.gpx_text()
    if not text or not text.strip():
        gpx_validation.set({"valid": False, "messages": ["No GPX data provided."], "points": []})
        return
    gpx_validation.set(validate_gpx(text.strip()))


@reactive.calc
def _csv_fields():
    uploaded = read_uploaded(input.csv_file())
    text = (uploaded if uploaded else input.csv_text() or "").strip()
    if not text:
        return []
    try:
        reader = csv.DictReader(io.StringIO(text))
        return list(reader.fieldnames or [])
    except Exception:
        return []


@reactive.effect
def _update_col_selects():
    fields = _csv_fields()
    if not fields:
        ui.update_select("csv_lat_col", choices={"": "(no columns)"}, selected="")
        ui.update_select("csv_lon_col", choices={"": "(no columns)"}, selected="")
        return
    choices = {f: f for f in fields}
    lat_default = next(
        (f for f in fields if f.strip().lower() in ("lat", "latitude")),
        fields[0],
    )
    lon_default = next(
        (f for f in fields if f.strip().lower() in ("lon", "lng", "longitude", "long")),
        fields[1] if len(fields) > 1 else fields[0],
    )
    ui.update_select("csv_lat_col", choices=choices, selected=lat_default)
    ui.update_select("csv_lon_col", choices=choices, selected=lon_default)


@reactive.effect
@reactive.event(input.convert_btn)
def _do_convert():
    uploaded = read_uploaded(input.csv_file())
    text = uploaded if uploaded else input.csv_text()
    if not text or not text.strip():
        csv_gpx_xml.set(None)
        csv_gpx_obj.set(None)
        csv_points.set(None)
        return
    lat_col = input.csv_lat_col() or None
    lon_col = input.csv_lon_col() or None
    try:
        gpx_xml = csv_to_gpx(text.strip(), lat_col=lat_col, lon_col=lon_col)
        csv_gpx_xml.set(gpx_xml)
        gpx_parsed = gpxpy.parse(gpx_xml)
        pts = []
        for t in gpx_parsed.tracks:
            for s in t.segments:
                pts.extend(s.points)
        csv_gpx_obj.set(gpx_parsed)
        csv_points.set(pts)
    except Exception as exc:
        csv_gpx_xml.set(f"ERROR: {exc}")
        csv_gpx_obj.set(None)
        csv_points.set(None)

requirements.txt

shiny 
shinywidgets 
ipyleaflet 
gpxpy
shinyswatch