"""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("/") 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/") 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/") 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/") 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)