openstreetmap-tools/web/app.py

"""Flask web frontend for osm-pt-geojson."""
import re
from typing import Any

from flask import Flask, jsonify, redirect, render_template, request, url_for
from flask.typing import ResponseReturnValue

from osm_geojson.pt.core import (
    OsmError,
    build_route_coords,
    fetch_relation_full,
    fetch_route_master_routes,
    fetch_sibling_routes,
    make_geojson,
    nearest_coord_index,
    node_name,
    parse_elements,
)

app = Flask(__name__)
app.secret_key = "dev"

PUBLIC_TRANSPORT_ROUTE_TYPES = {
    "bus", "trolleybus", "tram", "subway", "train",
    "light_rail", "monorail", "ferry", "funicular",
}


def parse_relation_id(text: str) -> int | None:
    """Extract a relation ID from a bare integer, partial path, or full OSM URL."""
    text = text.strip()
    m = re.search(r"relation/(\d+)", text)
    if m:
        return int(str(m.group(1)))
    if re.fullmatch(r"\d+", text):
        return int(text)
    return None


def check_route_tags(
    tags: dict[str, str], relation_id: int
) -> tuple[dict[str, Any], int] | None:
    """Return a (error_dict, http_status) if the relation is not a supported public transport route."""
    if tags.get("type") == "route_master":
        return {
            "error": "is_route_master",
            "message": (
                f"Relation {relation_id} is a route_master relation. "
                "Select one of its individual routes."
            ),
        }, 422
    if tags.get("type") != "route":
        kind = tags.get("type", "unknown")
        return {
            "error": "not_public_transport",
            "message": (
                f"Relation {relation_id} is a {kind!r} relation, not a route relation. "
                "This tool only supports public transport route relations."
            ),
        }, 422
    route = tags.get("route")
    if not route:
        return {
            "error": "not_public_transport",
            "message": (
                f"Relation {relation_id} is a route relation but has no 'route' tag. "
                "Cannot determine the route type."
            ),
        }, 422
    if route not in PUBLIC_TRANSPORT_ROUTE_TYPES:
        supported = ", ".join(sorted(PUBLIC_TRANSPORT_ROUTE_TYPES))
        return {
            "error": "not_public_transport",
            "route_type": route,
            "message": (
                f"Relation {relation_id} is a {route!r} route, not a public transport route. "
                f"This tool supports: {supported}."
            ),
        }, 422
    return None


@app.route("/docs")
def docs() -> ResponseReturnValue:
    """Render the API documentation page."""
    return render_template("api.html")


@app.route("/")
def index() -> ResponseReturnValue:
    """Render the landing page with no relation loaded."""
    error = request.args.get("error")
    return render_template("index.html", relation_id=None, error=error)


@app.route("/<int:relation_id>")
def route_page(relation_id: int) -> ResponseReturnValue:
    """Render the page with a relation pre-loaded."""
    return render_template("index.html", relation_id=relation_id, error=None)


@app.route("/load", methods=["POST"])
def load() -> ResponseReturnValue:
    """Parse user input and redirect to the relation page."""
    raw = request.form.get("relation", "").strip()
    relation_id = parse_relation_id(raw)
    if relation_id is None:
        return redirect(url_for("index", error="Could not find a relation ID in that input."))
    return redirect(url_for("route_page", relation_id=relation_id))


@app.route("/api/route/<int:relation_id>")
def api_route(relation_id: int) -> ResponseReturnValue:
    """Return the full route as GeoJSON plus a list of stops."""
    try:
        data = fetch_relation_full(relation_id)
        nodes, ways, stop_ids, way_ids, tags = parse_elements(data, relation_id)
    except OsmError as e:
        return jsonify({"error": "osm_error", "message": str(e)}), e.status_code

    err = check_route_tags(tags, relation_id)
    if err:
        return jsonify(err[0]), err[1]

    route_coords = build_route_coords(way_ids, ways, nodes)
    geojson = make_geojson(route_coords, stop_ids, nodes, tags)

    stops = []
    for sid in stop_ids:
        if sid in nodes:
            n = nodes[sid]
            stops.append({"name": node_name(n), "lat": n["lat"], "lon": n["lon"]})

    other_directions = fetch_sibling_routes(relation_id)

    return jsonify({
        "name": tags.get("name", str(relation_id)),
        "ref": tags.get("ref"),
        "stops": stops,
        "geojson": geojson,
        "other_directions": other_directions,
    })


@app.route("/api/segment/<int:relation_id>")
def api_segment(relation_id: int) -> ResponseReturnValue:
    """Return GeoJSON for the segment between two named stops."""
    from_name = request.args.get("from", "").strip()
    to_name = request.args.get("to", "").strip()
    include_stops = request.args.get("stops", "1") != "0"

    if not from_name or not to_name:
        return jsonify({
            "error": "missing_params",
            "message": "Both 'from' and 'to' parameters are required.",
        }), 400

    try:
        data = fetch_relation_full(relation_id)
        nodes, ways, stop_ids, way_ids, tags = parse_elements(data, relation_id)
    except OsmError as e:
        return jsonify({"error": "osm_error", "message": str(e)}), e.status_code

    err = check_route_tags(tags, relation_id)
    if err:
        return jsonify(err[0]), err[1]

    route_coords = build_route_coords(way_ids, ways, nodes)

    def find_stop(name: str) -> int | None:
        """Return the node ID of the stop matching name (case-insensitive), or None."""
        for sid in stop_ids:
            if sid in nodes and node_name(nodes[sid]).lower() == name.lower():
                return sid
        return None

    sid_from = find_stop(from_name)
    sid_to = find_stop(to_name)

    missing = [n for n, s in ((from_name, sid_from), (to_name, sid_to)) if s is None]
    if missing:
        available = [node_name(nodes[s]) for s in stop_ids if s in nodes]
        return jsonify({
            "error": "station_not_found",
            "message": f"Station(s) not found: {', '.join(repr(m) for m in missing)}",
            "available": available,
        }), 404

    assert sid_from is not None and sid_to is not None
    idx_from = nearest_coord_index(nodes[sid_from]["lon"], nodes[sid_from]["lat"], route_coords)
    idx_to = nearest_coord_index(nodes[sid_to]["lon"], nodes[sid_to]["lat"], route_coords)

    geojson = make_geojson(
        route_coords, stop_ids, nodes, tags,
        idx_from=idx_from, idx_to=idx_to,
        no_stops=not include_stops,
    )
    return jsonify(geojson)


@app.route("/api/route_master/<int:relation_id>")
def api_route_master(relation_id: int) -> ResponseReturnValue:
    """Return all member routes of a route_master with their GeoJSON geometries."""
    try:
        rm_tags, routes = fetch_route_master_routes(relation_id)
    except OsmError as e:
        return jsonify({"error": "osm_error", "message": str(e)}), e.status_code
    return jsonify({
        "name": rm_tags.get("name", str(relation_id)),
        "ref": rm_tags.get("ref"),
        "routes": routes,
    })


if __name__ == "__main__":
    app.run(debug=True)