app.py

#!/usr/bin/env python3
"""
Gradio app to visualize ADS-B flight routes from an uploaded GeoJSON file.

Upload a GeoJSON FeatureCollection (e.g. produced by generate_adsb.py),
pick a flight, and the app draws its route on a map and shows the aircraft
info plus per-sample telemetry.

IMPORTANT: this app deliberately HIDES any anomaly metadata. It is intended
for an exercise where the user must find the anomaly themselves, so the
`properties.anomaly` field (and anything that would label a flight as
anomalous) is never displayed.
"""

import json

import gradio as gr
import plotly.graph_objects as go

# Fields in feature properties that could leak the answer. Never surface these.
HIDDEN_KEYS = {"anomaly", "anomalies", "is_anomalous", "anomaly_type",
               "anomaly_index", "label", "ground_truth"}


def parse_geojson(file_path):
    """Load a GeoJSON file and return the list of flight features."""
    with open(file_path, "r") as f:
        data = json.load(f)

    if isinstance(data, dict) and data.get("type") == "FeatureCollection":
        features = data.get("features", [])
    elif isinstance(data, dict) and data.get("type") == "Feature":
        features = [data]
    else:
        raise ValueError("File is not a GeoJSON Feature or FeatureCollection.")

    flights = [f for f in features
               if f.get("geometry", {}).get("type") == "LineString"]
    if not flights:
        raise ValueError("No LineString flight tracks found in the file.")
    return flights


def flight_label(idx, feature):
    """Human-readable dropdown label for a flight (no anomaly hints)."""
    p = feature.get("properties", {})
    icao = p.get("icao24", f"flight-{idx}")
    callsign = p.get("callsign", "?")
    route = f'{p.get("origin", "?")}->{p.get("destination", "?")}'
    return f"{idx + 1:>3}. {callsign}  [{icao}]  {route}"


def load_file(file_path):
    """Handle an uploaded file: parse it and populate the flight dropdown."""
    if not file_path:
        return (gr.update(choices=[], value=None), [],
                "Upload a GeoJSON file to begin.")
    try:
        flights = parse_geojson(file_path)
    except Exception as e:  # noqa: BLE001 - surface any parse error to the UI
        return (gr.update(choices=[], value=None), [],
                f"Could not read file: {e}")

    choices = [(flight_label(i, f), i) for i, f in enumerate(flights)]
    status = (f"Loaded **{len(flights)}** flights. "
              "Select one from the dropdown to view its route.")
    # Auto-select the first flight for convenience.
    return (gr.update(choices=choices, value=0), flights, status)


def empty_map():
    """A blank world map shown before any flight is selected."""
    fig = go.Figure(go.Scattermap(lat=[], lon=[], mode="markers"))
    fig.update_layout(
        map_style="open-street-map",
        map_zoom=1,
        map_center={"lat": 25, "lon": 10},
        margin={"l": 0, "r": 0, "t": 0, "b": 0},
        height=560,
    )
    return fig


def render_flight(flight_idx, flights):
    """Draw the selected flight on the map and build its info panel."""
    if flight_idx is None or not flights:
        return empty_map(), "_No flight selected._", None

    feature = flights[int(flight_idx)]
    coords = feature["geometry"]["coordinates"]
    lons = [c[0] for c in coords]
    lats = [c[1] for c in coords]
    p = feature.get("properties", {})

    alts = p.get("altitudes_ft", [])
    spds = p.get("ground_speeds_kt", [])
    times = p.get("timestamps", [])
    n = len(coords)

    # Per-point hover text built only from non-leaking telemetry.
    hover = []
    for i in range(n):
        parts = [f"Sample {i + 1}/{n}"]
        if i < len(times):
            parts.append(f"t: {times[i]}")
        if i < len(alts):
            parts.append(f"alt: {alts[i]:,} ft")
        if i < len(spds):
            parts.append(f"gs: {spds[i]} kt")
        hover.append("<br>".join(parts))

    fig = go.Figure()
    # The route line.
    fig.add_trace(go.Scattermap(
        lat=lats, lon=lons, mode="lines",
        line={"width": 3, "color": "#1f77b4"},
        name="route", hoverinfo="skip",
    ))
    # Sample points with telemetry hover.
    fig.add_trace(go.Scattermap(
        lat=lats, lon=lons, mode="markers",
        marker={"size": 7, "color": "#1f77b4"},
        text=hover, hoverinfo="text", name="samples",
    ))
    # Start / end markers.
    fig.add_trace(go.Scattermap(
        lat=[lats[0]], lon=[lons[0]], mode="markers",
        marker={"size": 13, "color": "#2ca02c"},
        text=[f"Origin: {p.get('origin', '?')}"], hoverinfo="text",
        name="origin",
    ))
    fig.add_trace(go.Scattermap(
        lat=[lats[-1]], lon=[lons[-1]], mode="markers",
        marker={"size": 13, "color": "#d62728"},
        text=[f"Destination: {p.get('destination', '?')}"], hoverinfo="text",
        name="destination",
    ))

    # Center the view on the track.
    center = {"lat": sum(lats) / n, "lon": sum(lons) / n}
    span = max(max(lats) - min(lats), max(lons) - min(lons), 0.5)
    zoom = max(1, min(9, 8 - (span ** 0.5)))
    fig.update_layout(
        map_style="open-street-map",
        map_zoom=zoom,
        map_center=center,
        margin={"l": 0, "r": 0, "t": 0, "b": 0},
        height=560,
        showlegend=False,
    )

    # Info panel (markdown) - anomaly fields are explicitly excluded.
    info_lines = [
        f"### {p.get('callsign', 'Unknown flight')}",
        "",
        f"- **ICAO24:** `{p.get('icao24', '?')}`",
        f"- **Aircraft type:** {p.get('aircraft_type', '?')}",
        f"- **Route:** {p.get('origin', '?')} -> {p.get('destination', '?')}",
        f"- **Samples:** {n}",
    ]
    if "sample_interval_s" in p:
        info_lines.append(f"- **Sample interval:** {p['sample_interval_s']} s")
    if alts:
        info_lines.append(f"- **Altitude range:** {min(alts):,} - "
                          f"{max(alts):,} ft")
    if spds:
        info_lines.append(f"- **Ground speed range:** {min(spds)} - "
                          f"{max(spds)} kt")
    if times:
        info_lines.append(f"- **First seen:** {times[0]}")
        info_lines.append(f"- **Last seen:** {times[-1]}")

    # Surface any extra, non-sensitive properties for transparency.
    extra = {k: v for k, v in p.items()
             if k not in HIDDEN_KEYS
             and k not in {"icao24", "callsign", "aircraft_type", "origin",
                           "destination", "sample_interval_s", "altitudes_ft",
                           "ground_speeds_kt", "track_deg", "timestamps"}}
    if extra:
        info_lines.append("")
        info_lines.append("**Other properties:**")
        for k, v in extra.items():
            info_lines.append(f"- {k}: {v}")

    info = "\n".join(info_lines)

    # Telemetry table rows for the dataframe.
    tracks = p.get("track_deg", [])
    rows = []
    for i in range(n):
        rows.append([
            i + 1,
            times[i] if i < len(times) else "",
            round(lats[i], 5),
            round(lons[i], 5),
            alts[i] if i < len(alts) else "",
            spds[i] if i < len(spds) else "",
            tracks[i] if i < len(tracks) else "",
        ])

    return fig, info, rows


with gr.Blocks(title="ADS-B Flight Route Viewer") as demo:
    gr.Markdown(
        "# ADS-B Flight Route Viewer\n"
        "Upload a GeoJSON flight dataset, pick a flight, and inspect its route "
        "and telemetry on the map.\n\n"
        "_This viewer intentionally does **not** reveal which flights are "
        "anomalous — finding the anomaly is the exercise._"
    )

    flights_state = gr.State([])

    with gr.Row():
        file_in = gr.File(
            label="Upload GeoJSON (.geojson / .json)",
            file_types=[".geojson", ".json"],
            type="filepath",
        )
        flight_dd = gr.Dropdown(label="Flight", choices=[], interactive=True)

    status = gr.Markdown("Upload a GeoJSON file to begin.")

    with gr.Row():
        with gr.Column(scale=3):
            map_plot = gr.Plot(label="Route", value=empty_map())
        with gr.Column(scale=2):
            info_md = gr.Markdown("_No flight selected._")

    telemetry = gr.Dataframe(
        headers=["#", "timestamp", "lat", "lon", "alt_ft", "gs_kt",
                 "track_deg"],
        label="Telemetry (per sample)",
        interactive=False,
        wrap=True,
    )

    file_in.change(load_file, inputs=file_in,
                   outputs=[flight_dd, flights_state, status])
    # Render when a file is loaded (auto-selected flight) or selection changes.
    flight_dd.change(render_flight, inputs=[flight_dd, flights_state],
                     outputs=[map_plot, info_md, telemetry])
    flights_state.change(render_flight, inputs=[flight_dd, flights_state],
                         outputs=[map_plot, info_md, telemetry])


if __name__ == "__main__":
    demo.launch(server_name="0.0.0.0", server_port=7860)