edward/openstreetmap-tools
1
0
Fork 0
Code Issues 1 Pull requests Projects Releases Packages Wiki Activity Actions

Simplify bus stop finder code

Browse source
This commit is contained in:
Edward Betts 2026-08-15 15:51:58 +01:00
parent 7823349f8f
commit e8d0ff5c2c
3 changed files with 75 additions and 73 deletions
Download patch file Download diff file Expand all files Collapse all files

81
src/uk_bus_stops/app.py
View file

@ -9,6 +9,7 @@ from flask import Flask, Response, jsonify, render_template, request
from flask.typing import ResponseReturnValue
from uk_bus_stops.core import (
OSM_ELEMENT_SELECTORS,
UpstreamError,
fetch_stop,
find_atco_code,
@ -56,6 +57,11 @@ def create_app(test_config: dict[str, Any] | None = None) -> Flask:
if test_config:
app.config.update(test_config)
@app.errorhandler(UpstreamError)
def handle_upstream_error(exc: UpstreamError) -> tuple[Response, int]:
"""Return the common JSON response for upstream OSM service failures."""
return _error("upstream_error", str(exc), 502)
@app.get("/")
def index() -> ResponseReturnValue:
"""Render the bus stop finder."""
@ -76,25 +82,22 @@ def create_app(test_config: dict[str, Any] | None = None) -> Flask:
viewbox = parse_viewbox()
except ValueError as exc:
return _error("invalid_viewbox", str(exc), 400)
try:
coordinates = parse_coordinates(query)
if coordinates is not None:
lat, lon = coordinates
coordinate_location = {"lat": lat, "lon": lon, "label": f"{lat}, {lon}"}
return jsonify({
"kind": "location",
"location": coordinate_location,
"stops": nearby_stops(lat, lon),
})
if ATCO_PATTERN.fullmatch(query):
stops = find_atco_code(query)
return jsonify({"kind": "atco", "label": query.upper(), "stops": stops})
locations = geocode(query, viewbox=viewbox)
if not locations:
return _error("not_found", "No UK location matched that search.", 404)
return jsonify({"kind": "geocode", "query": query, "locations": locations})
except UpstreamError as exc:
return _error("upstream_error", str(exc), 502)
coordinates = parse_coordinates(query)
if coordinates is not None:
lat, lon = coordinates
coordinate_location = {"lat": lat, "lon": lon, "label": f"{lat}, {lon}"}
return jsonify({
"kind": "location",
"location": coordinate_location,
"stops": nearby_stops(lat, lon),
})
if ATCO_PATTERN.fullmatch(query):
stops = find_atco_code(query)
return jsonify({"kind": "atco", "label": query.upper(), "stops": stops})
locations = geocode(query, viewbox=viewbox)
if not locations:
return _error("not_found", "No UK location matched that search.", 404)
return jsonify({"kind": "geocode", "query": query, "locations": locations})
@app.get("/api/stops")
def stops() -> ResponseReturnValue:
@ -106,38 +109,29 @@ def create_app(test_config: dict[str, Any] | None = None) -> Flask:
return _error("invalid_location", "Valid latitude and longitude are required.", 400)
if not -90 <= lat <= 90 or not -180 <= lon <= 180:
return _error("invalid_location", "Latitude or longitude is out of range.", 400)
try:
label = request.args.get("label", "").strip() or f"{lat}, {lon}"
return jsonify({
"kind": "location",
"location": {"lat": lat, "lon": lon, "label": label},
"stops": nearby_stops(lat, lon),
})
except UpstreamError as exc:
return _error("upstream_error", str(exc), 502)
label = request.args.get("label", "").strip() or f"{lat}, {lon}"
return jsonify({
"kind": "location",
"location": {"lat": lat, "lon": lon, "label": label},
"stops": nearby_stops(lat, lon),
})
@app.get("/api/stop/<element_type>/<int:element_id>/routes")
def stop_routes(element_type: str, element_id: int) -> ResponseReturnValue:
"""Return stop-area-aware OSM bus routes for one stop object."""
if element_type not in {"node", "way", "relation"}:
if element_type not in OSM_ELEMENT_SELECTORS:
return _error("invalid_stop", "Unsupported OpenStreetMap object type.", 404)
try:
return jsonify({"routes": routes_for_stop(element_type, element_id)})
except UpstreamError as exc:
return _error("upstream_error", str(exc), 502)
return jsonify({"routes": routes_for_stop(element_type, element_id)})
@app.get("/api/stop/<element_type>/<int:element_id>")
def stop_detail(element_type: str, element_id: int) -> ResponseReturnValue:
"""Return one OSM stop for a shareable stop URL."""
if element_type not in {"node", "way", "relation"}:
if element_type not in OSM_ELEMENT_SELECTORS:
return _error("invalid_stop", "Unsupported OpenStreetMap object type.", 404)
try:
stop = fetch_stop(element_type, element_id)
if stop is None:
return _error("stop_not_found", "That OpenStreetMap stop was not found.", 404)
return jsonify({"stop": stop})
except UpstreamError as exc:
return _error("upstream_error", str(exc), 502)
stop = fetch_stop(element_type, element_id)
if stop is None:
return _error("stop_not_found", "That OpenStreetMap stop was not found.", 404)
return jsonify({"stop": stop})
@app.get("/api/stops/in-bounds")
def bounded_stops() -> ResponseReturnValue:
@ -155,10 +149,7 @@ def create_app(test_config: dict[str, Any] | None = None) -> Flask:
lon_span = east - west
if lat_span > 0.15 or lon_span > 0.2 or lat_span * lon_span > 0.015:
return _error("area_too_large", "Zoom in further to load transport stops.", 400)
try:
return jsonify({"kind": "map", "stops": stops_in_bounds(south, west, north, east)})
except UpstreamError as exc:
return _error("upstream_error", str(exc), 502)
return jsonify({"kind": "map", "stops": stops_in_bounds(south, west, north, east)})
return app

43
src/uk_bus_stops/core.py
View file

@ -11,6 +11,11 @@ OVERPASS_URL = "https://overpass.atownsend.org.uk/api/interpreter"
OVERPASS_FALLBACK_URLS = ("https://overpass.private.coffee/api/interpreter",)
USER_AGENT = "uk-bus-stops/0.1 (https://openstreetmap.tools/uk-bus-stops; edward@4angle.com)"
TIMEOUT = 30
OSM_ELEMENT_SELECTORS = {
"node": ("node", "bn"),
"way": ("way", "bw"),
"relation": ("rel", "br"),
}
class UpstreamError(Exception):
@ -41,7 +46,6 @@ def geocode(
"format": "jsonv2",
"limit": 20,
"countrycodes": "gb",
"addressdetails": 1,
"dedupe": 1,
"bounded": 0,
}
@ -104,6 +108,16 @@ def _normalise_stop(element: dict[str, Any]) -> dict[str, Any] | None:
}
def _normalise_stops(elements: list[dict[str, Any]]) -> list[dict[str, Any]]:
"""Normalise and deduplicate Overpass stop elements by OSM object reference."""
found: dict[tuple[str, int], dict[str, Any]] = {}
for element in elements:
stop = _normalise_stop(element)
if stop is not None:
found[(stop["type"], stop["id"])] = stop
return list(found.values())
def transport_type(tags: dict[str, str]) -> str:
"""Return a human-readable transport type inferred from OSM tags."""
if tags.get("amenity") == "ferry_terminal" or tags.get("ferry") == "yes":
@ -138,17 +152,13 @@ def nearby_stops(lat: float, lon: float, radius: int = 1000) -> list[dict[str, A
nwr(around:{radius},{lat:.7f},{lon:.7f})[\"public_transport\"=\"platform\"][\"bus\"=\"yes\"];
);
out center tags;"""
found: dict[tuple[str, int], dict[str, Any]] = {}
for element in _overpass(query):
stop = _normalise_stop(element)
if stop is not None:
found[(stop["type"], stop["id"])] = stop
found = _normalise_stops(_overpass(query))
def distance_squared(stop: dict[str, Any]) -> float:
"""Give a cheap local-distance ordering without another dependency."""
return (float(stop["lat"]) - lat) ** 2 + (float(stop["lon"]) - lon) ** 2
return sorted(found.values(), key=distance_squared)
return sorted(found, key=distance_squared)
def stops_in_bounds(
@ -163,12 +173,8 @@ def stops_in_bounds(
nwr({bbox})[\"public_transport\"=\"platform\"][\"bus\"=\"yes\"];
);
out center tags 1000;"""
found: dict[tuple[str, int], dict[str, Any]] = {}
for element in _overpass(query):
stop = _normalise_stop(element)
if stop is not None:
found[(stop["type"], stop["id"])] = stop
return sorted(found.values(), key=lambda stop: (stop["name"], stop["id"]))
found = _normalise_stops(_overpass(query))
return sorted(found, key=lambda stop: (stop["name"], stop["id"]))
def find_atco_code(code: str) -> list[dict[str, Any]]:
@ -182,9 +188,10 @@ out center tags;"""
def fetch_stop(element_type: str, element_id: int) -> dict[str, Any] | None:
"""Fetch and normalise one OSM transport stop by object type and ID."""
selector = {"node": "node", "way": "way", "relation": "rel"}.get(element_type)
if selector is None:
selectors = OSM_ELEMENT_SELECTORS.get(element_type)
if selectors is None:
return None
selector, _ = selectors
query = f"""[out:json][timeout:25];
{selector}({element_id});
out center tags;"""
@ -197,10 +204,10 @@ out center tags;"""
def routes_for_stop(element_type: str, element_id: int) -> list[dict[str, Any]]:
"""Find bus routes using a stop or any member of its stop area."""
selector = {"node": "node", "way": "way", "relation": "rel"}.get(element_type)
membership = {"node": "bn", "way": "bw", "relation": "br"}.get(element_type)
if selector is None or membership is None:
selectors = OSM_ELEMENT_SELECTORS.get(element_type)
if selectors is None:
return []
selector, membership = selectors
query = f"""[out:json][timeout:25];
{selector}({element_id})->.selected;
rel({membership}.selected)[\"public_transport\"=\"stop_area\"]->.parent_areas;

24
src/uk_bus_stops/static/app.js
View file

@ -236,6 +236,14 @@ function distanceLabel(distance) {
return `${(distance / 1000).toFixed(1)} km`;
}
/** Replace the marker identifying the location around which stops were found. */
function setSearchMarker(locationResult, label) {
if (searchMarker) searchMarker.remove();
searchMarker = L.circleMarker([locationResult.lat, locationResult.lon], {
radius: 7, color: '#0d6efd', fillColor: '#fff', fillOpacity: 1, weight: 3,
}).bindTooltip(label).addTo(map);
}
function renderStops(data, fitMap = true, sortOrigin = null, preserveSelection = false) {
const origin = data.location || sortOrigin;
stops = [...data.stops]
@ -267,9 +275,7 @@ function renderStops(data, fitMap = true, sortOrigin = null, preserveSelection =
if (searchMarker) { searchMarker.remove(); searchMarker = null; }
if (data.location) {
searchMarker = L.circleMarker([data.location.lat, data.location.lon], {
radius: 7, color: '#0d6efd', fillColor: '#fff', fillOpacity: 1, weight: 3,
}).bindTooltip(data.location.label).addTo(map);
setSearchMarker(data.location, data.location.label);
}
if (isMobile() && data.kind !== 'map') setPanel(true);
@ -338,7 +344,7 @@ function renderGeocodeChoices(data) {
for (const locationResult of data.locations) {
const button = document.createElement('button');
button.type = 'button';
button.className = 'stop-item location-item';
button.className = 'stop-item';
const title = document.createElement('span');
title.className = 'stop-title';
title.textContent = locationResult.label;
@ -366,10 +372,7 @@ function chooseLocation(locationResult) {
/** Mark and zoom to known coordinates without waiting for the stop query. */
function showLocationImmediately(locationResult) {
if (searchMarker) searchMarker.remove();
searchMarker = L.circleMarker([locationResult.lat, locationResult.lon], {
radius: 7, color: '#0d6efd', fillColor: '#fff', fillOpacity: 1, weight: 3,
}).bindTooltip(locationResult.label || 'Search location').addTo(map);
setSearchMarker(locationResult, locationResult.label || 'Search location');
if (isMobile()) setPanel(true);
moveMap(() => centrePointInVisibleMap(locationResult.lat, locationResult.lon, 16));
}
@ -497,8 +500,9 @@ async function selectStop(stop, updateUrl = true) {
show('stop-detail');
byId('stop-name').textContent = stop.name;
byId('stop-type').textContent = stop.transport_type || 'Transport stop';
byId('stop-bearing').textContent = bearingLabel(stop) || '';
show('stop-bearing', Boolean(bearingLabel(stop)));
const bearing = bearingLabel(stop);
byId('stop-bearing').textContent = bearing || '';
show('stop-bearing', Boolean(bearing));
byId('stop-indicator').textContent = stop.indicator ? `Stop ${stop.indicator}` : '';
show('stop-indicator', Boolean(stop.indicator));
byId('coordinates').textContent = `${Number(stop.lat).toFixed(6)}, ${Number(stop.lon).toFixed(6)}`;