Edward Betts
a79b0a7098
Add one-click examples and route search by number or metadata, with transport type, location and map-area filters. Preserve search filters and bounds in shareable URLs and restore searches on reload and browser navigation. Validate with 91 passing tests, including desktop and mobile browser coverage, and strict mypy checks. Closes #2
246 lines
8.5 KiB
Python
246 lines
8.5 KiB
Python
"""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,
|
|
PUBLIC_TRANSPORT_ROUTE_TYPES,
|
|
Bounds,
|
|
build_route_coords,
|
|
fetch_relation_full,
|
|
fetch_relation_name,
|
|
fetch_route_master_routes,
|
|
fetch_sibling_routes,
|
|
find_locations,
|
|
make_geojson,
|
|
nearest_coord_index,
|
|
node_name,
|
|
parse_elements,
|
|
search_routes,
|
|
)
|
|
|
|
app = Flask(__name__)
|
|
app.secret_key = "dev"
|
|
|
|
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."""
|
|
name = fetch_relation_name(relation_id)
|
|
return render_template("index.html", relation_id=relation_id, error=None, route_name=name)
|
|
|
|
|
|
@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.get("/api/search")
|
|
def api_search() -> ResponseReturnValue:
|
|
"""Find public transport routes by name, terminus or route number."""
|
|
try:
|
|
bounds: Bounds | None = None
|
|
keys = ("south", "west", "north", "east")
|
|
if any(key in request.args for key in keys):
|
|
try:
|
|
bounds = (
|
|
float(request.args["south"]), float(request.args["west"]),
|
|
float(request.args["north"]), float(request.args["east"]),
|
|
)
|
|
except (KeyError, ValueError) as exc:
|
|
raise OsmError("All four search area coordinates are required and must be numbers.", 400) from exc
|
|
return jsonify(search_routes(request.args.get("q", ""), request.args.get("mode", ""), bounds))
|
|
except OsmError as exc:
|
|
code = "invalid_query" if exc.status_code == 400 else "osm_error"
|
|
return jsonify({"error": code, "message": str(exc)}), exc.status_code
|
|
|
|
|
|
@app.get("/api/locations")
|
|
def api_locations() -> ResponseReturnValue:
|
|
"""Find candidate locations and their search bounds for a place name."""
|
|
try:
|
|
return jsonify({"locations": find_locations(request.args.get("q", ""))})
|
|
except OsmError as exc:
|
|
code = "invalid_query" if exc.status_code == 400 else "osm_error"
|
|
return jsonify({"error": code, "message": str(exc)}), exc.status_code
|
|
|
|
|
|
@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)
|