diff --git a/README.md b/README.md index c25e469..0d288ec 100644 --- a/README.md +++ b/README.md @@ -107,6 +107,44 @@ when the server is running. --- +### UK Bus Stop Finder + +A mobile-friendly map for finding the `naptan:AtcoCode` recorded on UK bus +stops in OpenStreetMap. Search by postcode, street, place, exact ATCO code, or +browser location. Searches are preserved in shareable `?q=...` or +`?lat=...&lon=...` URLs, and `latitude, longitude` can be entered directly in +the search box. Selecting a stop shows its tags and stop-area-aware bus route +relations. Stop results identify their transport mode (such as bus, rail or +tram), and moving the map at zoom level 15 or closer loads stops in the visible +area. Wider map views deliberately do not query Overpass. + +Selecting a stop replaces the search parameters with a shareable stop URL such +as `?node=485403163` (or `?way=...` / `?relation=...` for other OSM object +types). Refreshing that URL fetches and reopens the same stop. + +UK stop queries use the Britain-and-Ireland Overpass instance at +`overpass.atownsend.org.uk`, with `overpass.private.coffee` as an automatic +fallback when the primary service is unavailable. + +Run it locally with: + +``` +pip install -e ".[web]" +flask --app uk_bus_stops.app run +``` + +The production URL is `https://openstreetmap.tools/uk-bus-stops/`. + +Browser tests use Python Playwright. After installing the development extras, +install Chromium once and run the suite: + +``` +playwright install chromium +pytest tests/ +``` + +--- + ## Licence MIT License. Copyright (c) 2026 Edward Betts. diff --git a/homepage/index.html b/homepage/index.html index c4866ef..59f41c5 100644 --- a/homepage/index.html +++ b/homepage/index.html @@ -325,6 +325,18 @@ +
+
+

UK Bus Stop Finder

+ ATCO +
+

+ Find the ATCO code for a UK bus stop by postcode, place, street or + current location. See nearby stops and the bus routes serving them. +

+ Find a bus stop → +
+

OWL Places

diff --git a/pyproject.toml b/pyproject.toml index cb66129..d8d0deb 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -20,6 +20,7 @@ web = [ ] dev = [ "mypy", + "playwright", "pytest", "responses", "types-requests", @@ -28,6 +29,10 @@ dev = [ [project.scripts] osm-pt-geojson = "osm_geojson.pt.cli:cli" +uk-bus-stops = "uk_bus_stops.app:main" [tool.setuptools.packages.find] where = ["src"] + +[tool.setuptools.package-data] +uk_bus_stops = ["templates/*.html", "static/*.css", "static/*.js"] diff --git a/src/uk_bus_stops/__init__.py b/src/uk_bus_stops/__init__.py new file mode 100644 index 0000000..3bfd7c7 --- /dev/null +++ b/src/uk_bus_stops/__init__.py @@ -0,0 +1,5 @@ +"""Find UK bus stop ATCO codes using OpenStreetMap data.""" + +from uk_bus_stops.app import create_app + +__all__ = ["create_app"] diff --git a/src/uk_bus_stops/app.py b/src/uk_bus_stops/app.py new file mode 100644 index 0000000..e98bd67 --- /dev/null +++ b/src/uk_bus_stops/app.py @@ -0,0 +1,158 @@ +"""Flask application for finding UK bus stop ATCO codes.""" + +from __future__ import annotations + +import re +from typing import Any + +from flask import Flask, Response, jsonify, render_template, request +from flask.typing import ResponseReturnValue + +from uk_bus_stops.core import ( + UpstreamError, + fetch_stop, + find_atco_code, + geocode, + nearby_stops, + routes_for_stop, + stops_in_bounds, +) + +ATCO_PATTERN = re.compile(r"^(?=[A-Za-z0-9]{10,16}$)(?=.*\d)[A-Za-z0-9]+$") +COORDINATE_PATTERN = re.compile( + r"^\s*([+-]?(?:\d+(?:\.\d*)?|\.\d+))\s*,\s*" + r"([+-]?(?:\d+(?:\.\d*)?|\.\d+))\s*$" +) + + +def parse_coordinates(value: str) -> tuple[float, float] | None: + """Parse a valid ``latitude, longitude`` pair from user input.""" + match = COORDINATE_PATTERN.fullmatch(value) + if match is None: + return None + lat, lon = float(match.group(1)), float(match.group(2)) + if not -90 <= lat <= 90 or not -180 <= lon <= 180: + return None + return lat, lon + + +def create_app(test_config: dict[str, Any] | None = None) -> Flask: + """Create and configure the bus stop finder application.""" + app = Flask(__name__) + if test_config: + app.config.update(test_config) + + @app.get("/") + def index() -> ResponseReturnValue: + """Render the bus stop finder.""" + return render_template("index.html") + + @app.get("/api/search") + def search() -> ResponseReturnValue: + """Search directly by ATCO code or geocode a UK location.""" + query = request.args.get("q", "").strip() + if not query: + return _error("missing_query", "Enter a postcode, place, street or ATCO code.", 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}) + location = geocode(query) + if location is None: + return _error("not_found", "No UK location matched that search.", 404) + stops = nearby_stops(location["lat"], location["lon"]) + return jsonify({"kind": "location", "location": location, "stops": stops}) + except UpstreamError as exc: + return _error("upstream_error", str(exc), 502) + + @app.get("/api/stops") + def stops() -> ResponseReturnValue: + """Find stops around coordinates, primarily for browser geolocation.""" + try: + lat = float(request.args["lat"]) + lon = float(request.args["lon"]) + except (KeyError, ValueError): + 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) + + @app.get("/api/stop///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"}: + 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) + + @app.get("/api/stop//") + 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"}: + 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) + + @app.get("/api/stops/in-bounds") + def bounded_stops() -> ResponseReturnValue: + """Find stops in a safely limited visible map area.""" + try: + south = float(request.args["south"]) + west = float(request.args["west"]) + north = float(request.args["north"]) + east = float(request.args["east"]) + except (KeyError, ValueError): + return _error("invalid_bounds", "Valid map bounds are required.", 400) + if not (-90 <= south < north <= 90 and -180 <= west < east <= 180): + return _error("invalid_bounds", "The map bounds are invalid.", 400) + lat_span = north - south + 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 app + + +def _error(code: str, message: str, status: int) -> tuple[Response, int]: + """Build the common JSON error shape.""" + return jsonify({"error": code, "message": message}), status + + +app = create_app() + + +def main() -> None: + """Run the development server from the console script.""" + app.run(debug=True) + + +if __name__ == "__main__": + main() diff --git a/src/uk_bus_stops/core.py b/src/uk_bus_stops/core.py new file mode 100644 index 0000000..84de7fe --- /dev/null +++ b/src/uk_bus_stops/core.py @@ -0,0 +1,223 @@ +"""Nominatim and Overpass clients for the UK bus stop finder.""" + +from __future__ import annotations + +from typing import Any, cast + +import requests + +NOMINATIM_URL = "https://nominatim.openstreetmap.org/search" +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 + + +class UpstreamError(Exception): + """Report a failure while talking to an OpenStreetMap service.""" + + +def _get_json(url: str, *, params: dict[str, Any]) -> Any: + """Fetch JSON from an OSM service with the application's identity.""" + try: + response = requests.get( + url, + params=params, + headers={"User-Agent": USER_AGENT, "Accept": "application/json"}, + timeout=TIMEOUT, + ) + response.raise_for_status() + return response.json() + except (requests.RequestException, ValueError) as exc: + raise UpstreamError("The OpenStreetMap service is temporarily unavailable.") from exc + + +def geocode(query: str) -> dict[str, Any] | None: + """Geocode a UK postcode, street, or place name with Nominatim.""" + results = _get_json( + NOMINATIM_URL, + params={ + "q": query, + "format": "jsonv2", + "limit": 1, + "countrycodes": "gb", + "addressdetails": 1, + }, + ) + if not isinstance(results, list) or not results: + return None + result = results[0] + return { + "lat": float(result["lat"]), + "lon": float(result["lon"]), + "label": result.get("display_name", query), + } + + +def _overpass(query: str) -> list[dict[str, Any]]: + """Run an Overpass query, trying a second public instance on failure.""" + last_error: UpstreamError | None = None + for url in (OVERPASS_URL, *OVERPASS_FALLBACK_URLS): + try: + data = _get_json(url, params={"data": query}) + if not isinstance(data, dict) or not isinstance(data.get("elements"), list): + raise UpstreamError("Overpass returned an unexpected response.") + return cast(list[dict[str, Any]], data["elements"]) + except UpstreamError as exc: + last_error = exc + assert last_error is not None + raise last_error + + +def _normalise_stop(element: dict[str, Any]) -> dict[str, Any] | None: + """Convert an Overpass element with a position into frontend stop data.""" + center = element.get("center", {}) + lat = element.get("lat", center.get("lat")) + lon = element.get("lon", center.get("lon")) + if lat is None or lon is None: + return None + tags = element.get("tags", {}) + return { + "type": element["type"], + "id": element["id"], + "lat": lat, + "lon": lon, + "name": tags.get("name") or tags.get("naptan:CommonName") or "Unnamed bus stop", + "transport_type": transport_type(tags), + "atco_code": tags.get("naptan:AtcoCode"), + "indicator": tags.get("naptan:Indicator"), + "bearing": tags.get("naptan:Bearing") or tags.get("bearing") or tags.get("direction"), + "tags": tags, + } + + +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": + return "Ferry terminal" + if tags.get("station") == "subway" or tags.get("subway") == "yes": + return "Underground station" + if tags.get("railway") == "tram_stop" or tags.get("tram") == "yes": + return "Tram stop" + if tags.get("light_rail") == "yes": + return "Light rail stop" + if tags.get("railway") == "station" or tags.get("train") == "yes": + return "Railway station" + if tags.get("railway") == "halt": + return "Railway halt" + if tags.get("highway") == "bus_stop" or tags.get("bus") == "yes": + return "Bus stop" + if tags.get("public_transport") == "station": + return "Public transport station" + if tags.get("public_transport") == "platform": + return "Public transport platform" + if tags.get("public_transport") == "stop_position": + return "Public transport stop" + return "Transport stop" + + +def nearby_stops(lat: float, lon: float, radius: int = 1000) -> list[dict[str, Any]]: + """Return bus stop objects near a coordinate, nearest first.""" + query = f"""[out:json][timeout:25]; +( + nwr(around:{radius},{lat:.7f},{lon:.7f})[\"naptan:AtcoCode\"]; + nwr(around:{radius},{lat:.7f},{lon:.7f})[\"highway\"=\"bus_stop\"]; + 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 + + 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) + + +def stops_in_bounds( + south: float, west: float, north: float, east: float +) -> list[dict[str, Any]]: + """Return transport stops inside a small map bounding box.""" + bbox = f"{south:.7f},{west:.7f},{north:.7f},{east:.7f}" + query = f"""[out:json][timeout:25]; +( + nwr({bbox})[\"naptan:AtcoCode\"]; + nwr({bbox})[\"highway\"=\"bus_stop\"]; + 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"])) + + +def find_atco_code(code: str) -> list[dict[str, Any]]: + """Find stops whose ATCO code exactly matches the supplied value.""" + escaped = code.replace("\\", "\\\\").replace('"', '\\"') + query = f"""[out:json][timeout:25]; +nwr[\"naptan:AtcoCode\"~\"^{escaped}$\",i]; +out center tags;""" + return [stop for item in _overpass(query) if (stop := _normalise_stop(item))] + + +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: + return None + query = f"""[out:json][timeout:25]; +{selector}({element_id}); +out center tags;""" + for element in _overpass(query): + stop = _normalise_stop(element) + if stop is not None: + return stop + return None + + +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: + return [] + query = f"""[out:json][timeout:25]; +{selector}({element_id})->.selected; +rel({membership}.selected)[\"public_transport\"=\"stop_area\"]->.parent_areas; +(.selected;.parent_areas;)->.areas; +nwr(r.areas)->.related; +rel(bn.selected)[\"route\"=\"bus\"]->.selected_node_routes; +rel(bw.selected)[\"route\"=\"bus\"]->.selected_way_routes; +rel(bn.related)[\"route\"=\"bus\"]->.related_node_routes; +rel(bw.related)[\"route\"=\"bus\"]->.related_way_routes; +( + .selected_node_routes; + .selected_way_routes; + .related_node_routes; + .related_way_routes; +); +out tags;""" + routes: list[dict[str, Any]] = [] + seen: set[int] = set() + for item in _overpass(query): + route_id = int(item["id"]) + if route_id in seen: + continue + seen.add(route_id) + tags = item.get("tags", {}) + routes.append({ + "id": route_id, + "ref": tags.get("ref"), + "name": tags.get("name"), + "operator": tags.get("operator"), + "from": tags.get("from"), + "to": tags.get("to"), + }) + return sorted(routes, key=lambda route: (route.get("ref") or "", route.get("name") or "")) diff --git a/src/uk_bus_stops/static/app.js b/src/uk_bus_stops/static/app.js new file mode 100644 index 0000000..ade616d --- /dev/null +++ b/src/uk_bus_stops/static/app.js @@ -0,0 +1,559 @@ +/** Interactive map and accessible results UI for the UK bus stop finder. */ +'use strict'; + +const map = L.map('map').setView([54.5, -3], 6); +L.tileLayer('https://tile.openstreetmap.org/{z}/{x}/{y}.png', { + attribution: '© OpenStreetMap contributors', + maxZoom: 19, +}).addTo(map); + +let stops = []; +let selectedStop = null; +// FeatureGroup provides getBounds(), which is used to frame search results. +let markers = L.featureGroup().addTo(map); +let searchMarker = null; +let selectedMarkerHalo = null; +let userMapInteractionPending = false; +let mapMoveTimer = null; +let mapRequestController = null; +let returnToResults = false; +let returnViaHistory = false; +const MIN_MAP_STOP_ZOOM = 15; + +const byId = id => document.getElementById(id); +const isMobile = () => window.innerWidth < 768; +const show = (id, visible = true) => byId(id).classList.toggle('d-none', !visible); + +function setPanel(open) { + byId('sidebar').classList.toggle('panel-open', open); + show('panel-button', !open); + setTimeout(() => map.invalidateSize(), 260); +} + +function showError(message) { + byId('alert').textContent = message; + show('alert'); +} + +function setLoading(loading, message = 'Finding stops…') { + byId('loading-text').textContent = message; + show('loading', loading); + byId('search-input').disabled = loading; +} + +/** Parse latitude, longitude input while rejecting out-of-range coordinates. */ +function parseCoordinates(value) { + const match = value.match(/^\s*([+-]?(?:\d+(?:\.\d*)?|\.\d+))\s*,\s*([+-]?(?:\d+(?:\.\d*)?|\.\d+))\s*$/); + if (!match) return null; + const lat = Number(match[1]); + const lon = Number(match[2]); + if (lat < -90 || lat > 90 || lon < -180 || lon > 180) return null; + return {lat, lon}; +} + +/** Put the current search in the address bar without reloading the page. */ +function setSearchUrl(params, replace = false) { + const query = params.toString(); + const url = query ? `${location.pathname}?${query}` : location.pathname; + history[replace ? 'replaceState' : 'pushState'](null, '', url); +} + +/** Perform a map movement without treating it as a user pan. */ +function moveMap(callback, onComplete = null) { + userMapInteractionPending = false; + map.stop(); + if (!onComplete) { + callback(); + return; + } + let finished = false; + let fallbackTimer = null; + const finish = () => { + if (finished) return; + finished = true; + map.off('moveend', finish); + clearTimeout(fallbackTimer); + onComplete(); + }; + map.once('moveend', finish); + callback(); + if (!finished) fallbackTimer = setTimeout(finish, 500); +} + +/** Run the search encoded in q or lat/lon URL parameters. */ +function restoreSearchFromUrl() { + const params = new URLSearchParams(location.search); + const stopReference = parseStopReference(params); + if (stopReference) { + returnToResults = false; + returnViaHistory = false; + byId('search-input').value = ''; + loadSharedStop(stopReference.type, stopReference.id); + return; + } + const query = params.get('q'); + if (query) { + byId('search-input').value = query; + loadUrl(`${API_URLS.search}?${new URLSearchParams({q: query})}`); + return; + } + + const latText = params.get('lat'); + const lonText = params.get('lon'); + const coordinates = latText !== null && lonText !== null + ? parseCoordinates(`${latText},${lonText}`) + : null; + if (coordinates) { + byId('search-input').value = `${coordinates.lat}, ${coordinates.lon}`; + loadUrl(`${API_URLS.stops}?${new URLSearchParams(coordinates)}`); + return; + } + + byId('search-input').value = ''; + markers.clearLayers(); + if (searchMarker) { searchMarker.remove(); searchMarker = null; } + if (selectedMarkerHalo) { selectedMarkerHalo.remove(); selectedMarkerHalo = null; } + selectedStop = null; + show('results-panel', false); + show('stop-detail', false); + const center = map.getCenter(); + if (map.getZoom() !== 6 || Math.abs(center.lat - 54.5) > 0.0001 || Math.abs(center.lng + 3) > 0.0001) { + moveMap(() => map.setView([54.5, -3], 6)); + } +} + +/** Parse shareable OSM node, way, or relation parameters. */ +function parseStopReference(params) { + for (const type of ['node', 'way', 'relation']) { + const id = params.get(type); + if (id && /^\d+$/.test(id)) return {type, id: Number(id)}; + } + return null; +} + +/** Build a mounted-path-safe API URL for a particular OSM stop. */ +function stopApiUrl(template, type, id) { + return template.replace('/TYPE/', `/${type}/`).replace('/0', `/${id}`); +} + +/** Fetch and display the stop named by a shareable URL. */ +async function loadSharedStop(type, id) { + show('alert', false); + setLoading(true, 'Loading stop…'); + try { + const response = await fetch(stopApiUrl(API_URLS.stopTemplate, type, id)); + const data = await response.json(); + if (!response.ok) throw new Error(data.message || 'Could not load that stop.'); + let nearbyData = { + kind: 'location', + location: {lat: data.stop.lat, lon: data.stop.lon, label: data.stop.name}, + stops: [data.stop], + }; + try { + const nearbyParams = new URLSearchParams({lat: data.stop.lat, lon: data.stop.lon}); + const nearbyResponse = await fetch(`${API_URLS.stops}?${nearbyParams}`); + const nearbyResult = await nearbyResponse.json(); + if (nearbyResponse.ok) nearbyData = nearbyResult; + } catch (_) { + // The selected stop is still useful if the optional nearby lookup fails. + } + const selectedIndex = nearbyData.stops.findIndex( + stop => stop.type === data.stop.type && stop.id === data.stop.id + ); + if (selectedIndex === -1) nearbyData.stops.unshift(data.stop); + else nearbyData.stops[selectedIndex] = data.stop; + renderStops(nearbyData, true); + returnToResults = true; + returnViaHistory = false; + await selectStop(data.stop, false); + } catch (error) { + showError(error.message || 'Network error. Please try again.'); + } finally { + setLoading(false); + } +} + +function stopSubtitle(stop) { + const parts = [stop.transport_type || 'Transport stop']; + const bearing = bearingLabel(stop); + if (bearing) parts.push(bearing); + if (stop.indicator && !parts.includes(stop.indicator)) parts.push(stop.indicator); + if (stop.atco_code) parts.push(stop.atco_code); + else parts.push('No ATCO code recorded'); + return parts.join(' · '); +} + +/** Format a compass bearing for concise marker hover text. */ +function bearingLabel(stop) { + if (!stop.bearing) return null; + const bearing = String(stop.bearing).trim(); + if (!bearing) return null; + return /-bound$/i.test(bearing) ? bearing : `${bearing.toUpperCase()}-bound`; +} + +/** Calculate great-circle distance between a stop and a reference point. */ +function distanceMetres(stop, origin) { + if (!origin) return null; + const originLon = origin.lon ?? origin.lng; + if (!Number.isFinite(Number(origin.lat)) || !Number.isFinite(Number(originLon))) return null; + const radians = degrees => degrees * Math.PI / 180; + const lat1 = radians(Number(origin.lat)); + const lat2 = radians(Number(stop.lat)); + const deltaLat = lat2 - lat1; + const deltaLon = radians(Number(stop.lon) - Number(originLon)); + const a = Math.sin(deltaLat / 2) ** 2 + + Math.cos(lat1) * Math.cos(lat2) * Math.sin(deltaLon / 2) ** 2; + return 6371000 * 2 * Math.atan2(Math.sqrt(a), Math.sqrt(1 - a)); +} + +/** Format a stop distance compactly for the result list. */ +function distanceLabel(distance) { + if (distance === null) return null; + if (distance < 1000) return `${Math.round(distance / 10) * 10} m`; + return `${(distance / 1000).toFixed(1)} km`; +} + +function renderStops(data, fitMap = true, sortOrigin = null, preserveSelection = false) { + const origin = data.location || sortOrigin; + stops = [...data.stops] + .map(stop => ({...stop, distance_metres: distanceMetres(stop, origin)})) + .sort((first, second) => { + if (first.distance_metres === null && second.distance_metres === null) return 0; + if (first.distance_metres === null) return 1; + if (second.distance_metres === null) return -1; + return first.distance_metres - second.distance_metres; + }); + if (!preserveSelection) { + selectedStop = null; + if (selectedMarkerHalo) { selectedMarkerHalo.remove(); selectedMarkerHalo = null; } + } + if (fitMap) show('map-status', false); + markers.clearLayers(); + if (!preserveSelection) { + show('stop-detail', false); + show('results-panel'); + } + byId('results-heading').textContent = data.kind === 'atco' + ? `ATCO code ${data.label}` + : data.kind === 'map' ? 'Stops in this map area' : 'Nearby stops'; + byId('stop-count').textContent = `${stops.length} found`; + const list = byId('stop-list'); + list.replaceChildren(); + + 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); + } + + for (const stop of stops) { + const button = document.createElement('button'); + button.type = 'button'; + button.className = 'stop-item'; + const title = document.createElement('span'); + title.className = 'stop-title'; + title.textContent = stop.name; + const meta = document.createElement('span'); + meta.className = 'stop-meta'; + meta.textContent = [distanceLabel(stop.distance_metres), stopSubtitle(stop)] + .filter(Boolean).join(' · '); + button.append(title, meta); + button.addEventListener('click', () => selectStop(stop)); + list.appendChild(button); + + const tooltipParts = [stop.name, bearingLabel(stop), stop.transport_type || 'Transport stop']; + const marker = L.circleMarker([stop.lat, stop.lon], { + radius: 7, color: '#fff', weight: 2, fillColor: markerColour(stop), fillOpacity: .95, + }).bindTooltip(tooltipParts.filter(Boolean).join(' · ')); + marker.on('click', () => selectStop(stop)); + markers.addLayer(marker); + } + + if (!stops.length) { + const empty = document.createElement('p'); + empty.className = 'small text-muted'; + empty.textContent = data.kind === 'map' + ? 'No transport stops were found in this map area.' + : 'No transport stops were found within 1 km.'; + list.appendChild(empty); + if (fitMap && data.location) moveMap(() => map.setView([data.location.lat, data.location.lon], 16)); + } else if (fitMap) { + const bounds = markers.getBounds(); + if (searchMarker) bounds.extend(searchMarker.getLatLng()); + moveMap(() => map.fitBounds(bounds, {padding: [30, 30], maxZoom: 17})); + if (data.kind === 'atco' && stops.length === 1) selectStop(stops[0]); + } + if (isMobile() && data.kind !== 'map') setPanel(true); +} + +/** Choose marker colours that make different transport modes easy to scan. */ +function markerColour(stop) { + const type = stop.transport_type || ''; + if (type.includes('Railway')) return '#6f42c1'; + if (type.includes('Tram')) return '#dc3545'; + if (type.includes('Underground')) return '#0d6efd'; + if (type.includes('Ferry')) return '#0dcaf0'; + if (type.includes('Bus')) return '#198754'; + return stop.atco_code ? '#198754' : '#fd7e14'; +} + +/** Draw a persistent high-contrast ring around the active stop. */ +function highlightSelectedStop(stop) { + if (selectedMarkerHalo) selectedMarkerHalo.remove(); + selectedMarkerHalo = L.circleMarker([stop.lat, stop.lon], { + radius: 15, + color: '#ffc107', + fillColor: '#ffc107', + fillOpacity: 0.22, + weight: 5, + opacity: 1, + interactive: false, + className: 'selected-stop-marker', + }).addTo(map); + selectedMarkerHalo.bringToFront(); +} + +function osmUrl(stop, edit = false) { + return `https://www.openstreetmap.org/${edit ? 'edit?' : ''}${edit ? `${stop.type}=${stop.id}` : `${stop.type}/${stop.id}`}`; +} + +function renderTags(tags) { + const list = byId('tag-list'); + list.replaceChildren(); + for (const [key, value] of Object.entries(tags).sort(([a], [b]) => a.localeCompare(b))) { + const term = document.createElement('dt'); + term.textContent = key; + const detail = document.createElement('dd'); + detail.textContent = value; + list.append(term, detail); + } +} + +async function selectStop(stop, updateUrl = true) { + selectedStop = stop; + highlightSelectedStop(stop); + userMapInteractionPending = false; + clearTimeout(mapMoveTimer); + if (mapRequestController) { + mapRequestController.abort(); + mapRequestController = null; + } + if (updateUrl) { + const currentParams = new URLSearchParams(location.search); + const alreadyShowingStop = Boolean(parseStopReference(currentParams)); + if (!alreadyShowingStop) { + returnToResults = true; + returnViaHistory = true; + } + setSearchUrl(new URLSearchParams({[stop.type]: String(stop.id)}), alreadyShowingStop); + } + show('back-button', returnToResults); + show('results-panel', false); + 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))); + 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)}`; + byId('osm-link').href = osmUrl(stop); + byId('edit-link').href = osmUrl(stop, true); + byId('atco-code').textContent = stop.atco_code || ''; + show('atco-present', Boolean(stop.atco_code)); + show('atco-missing', !stop.atco_code); + byId('copy-status').textContent = ''; + renderTags(stop.tags); + moveMap( + () => map.setView([stop.lat, stop.lon], Math.max(map.getZoom(), 17)), + () => { + if (selectedStop && selectedStop.type === stop.type && selectedStop.id === stop.id) { + loadVisibleMapStops(); + } + } + ); + if (isMobile()) setPanel(true); + + byId('route-list').replaceChildren(); + show('route-loading'); + try { + const url = stopApiUrl(API_URLS.routeTemplate, stop.type, stop.id); + const response = await fetch(url); + const data = await response.json(); + if (!response.ok) throw new Error(data.message || 'Could not load routes.'); + renderRoutes(data.routes); + } catch (error) { + byId('route-list').textContent = error.message; + } finally { + show('route-loading', false); + } +} + +function renderRoutes(routes) { + const list = byId('route-list'); + list.replaceChildren(); + if (!routes.length) { + list.textContent = 'No bus route relations were found in OpenStreetMap.'; + list.className = 'small text-muted'; + return; + } + list.className = ''; + for (const route of routes) { + const card = document.createElement('div'); + card.className = 'route-card small'; + const line = document.createElement('div'); + const link = document.createElement('a'); + link.href = `https://www.openstreetmap.org/relation/${route.id}`; + link.target = '_blank'; + link.rel = 'noopener'; + link.textContent = route.ref || route.name || `Route relation ${route.id}`; + link.className = 'route-ref'; + line.appendChild(link); + if (route.name && route.name !== route.ref) line.append(` ${route.name}`); + card.appendChild(line); + const details = [route.from && route.to ? `${route.from} → ${route.to}` : null, route.operator].filter(Boolean); + if (details.length) { + const meta = document.createElement('div'); + meta.className = 'text-muted'; + meta.textContent = details.join(' · '); + card.appendChild(meta); + } + list.appendChild(card); + } +} + +async function loadUrl(url) { + show('alert', false); + setLoading(true); + try { + const response = await fetch(url); + const data = await response.json(); + if (!response.ok) throw new Error(data.message || 'Search failed.'); + renderStops(data); + } catch (error) { + showError(error.message || 'Network error. Please try again.'); + } finally { + setLoading(false); + } +} + +/** Load stops for the current viewport after a user pan or zoom. */ +async function loadVisibleMapStops() { + if (map.getZoom() < MIN_MAP_STOP_ZOOM) { + markers.clearLayers(); + byId('map-status').textContent = 'Zoom in to load transport stops'; + show('map-status'); + return; + } + const bounds = map.getBounds(); + const params = new URLSearchParams({ + south: bounds.getSouth().toFixed(7), + west: bounds.getWest().toFixed(7), + north: bounds.getNorth().toFixed(7), + east: bounds.getEast().toFixed(7), + }); + const center = map.getCenter(); + if (!selectedStop) { + setSearchUrl(new URLSearchParams({ + lat: center.lat.toFixed(6), lon: center.lng.toFixed(6), + }), true); + } + byId('map-status').textContent = 'Loading stops in this area…'; + show('map-status'); + if (mapRequestController) mapRequestController.abort(); + mapRequestController = new AbortController(); + try { + const response = await fetch(`${API_URLS.boundedStops}?${params}`, { + signal: mapRequestController.signal, + }); + const data = await response.json(); + if (!response.ok) throw new Error(data.message || 'Could not load this map area.'); + renderStops(data, false, map.getCenter(), Boolean(selectedStop)); + show('map-status', false); + } catch (error) { + if (error.name === 'AbortError') return; + byId('map-status').textContent = error.message || 'Could not load this map area.'; + } +} + +byId('search-form').addEventListener('submit', event => { + event.preventDefault(); + const query = byId('search-input').value.trim(); + if (!query) return; + const coordinates = parseCoordinates(query); + if (coordinates) { + const params = new URLSearchParams(coordinates); + setSearchUrl(params); + loadUrl(`${API_URLS.stops}?${params}`); + } else { + const params = new URLSearchParams({q: query}); + setSearchUrl(params); + loadUrl(`${API_URLS.search}?${params}`); + } +}); + +byId('locate-button').addEventListener('click', () => { + show('alert', false); + if (!navigator.geolocation) return showError('Your browser does not support location access.'); + setLoading(true, 'Getting your location…'); + navigator.geolocation.getCurrentPosition( + position => { + const coordinates = {lat: position.coords.latitude, lon: position.coords.longitude}; + const params = new URLSearchParams(coordinates); + setSearchUrl(params); + byId('search-input').value = `${coordinates.lat}, ${coordinates.lon}`; + const apiParams = new URLSearchParams({...coordinates, label: 'Your location'}); + loadUrl(`${API_URLS.stops}?${apiParams}`); + }, + () => { setLoading(false); showError('Location access was denied or unavailable.'); }, + {enableHighAccuracy: true, timeout: 10000} + ); +}); + +byId('back-button').addEventListener('click', () => { + if (returnToResults && returnViaHistory) { + history.back(); + return; + } + selectedStop = null; + if (selectedMarkerHalo) { + selectedMarkerHalo.remove(); + selectedMarkerHalo = null; + } + show('stop-detail', false); + show('results-panel'); + if (returnToResults && stops.length) { + const center = map.getCenter(); + setSearchUrl(new URLSearchParams({ + lat: center.lat.toFixed(6), lon: center.lng.toFixed(6), + }), true); + } +}); + +byId('copy-button').addEventListener('click', async () => { + if (!selectedStop || !selectedStop.atco_code) return; + try { + await navigator.clipboard.writeText(selectedStop.atco_code); + byId('copy-status').textContent = 'Copied'; + } catch (_) { + byId('copy-status').textContent = 'Copy failed'; + } +}); + +byId('panel-button').addEventListener('click', () => setPanel(true)); +byId('mobile-handle').addEventListener('click', () => setPanel(false)); +map.on('moveend', () => { + if (!userMapInteractionPending) return; + userMapInteractionPending = false; + clearTimeout(mapMoveTimer); + mapMoveTimer = setTimeout(loadVisibleMapStops, 350); +}); +for (const eventName of ['pointerdown', 'wheel', 'keydown']) { + map.getContainer().addEventListener(eventName, () => { + userMapInteractionPending = true; + }, {passive: true}); +} +if (isMobile()) setPanel(true); +window.addEventListener('popstate', restoreSearchFromUrl); +restoreSearchFromUrl(); diff --git a/src/uk_bus_stops/static/style.css b/src/uk_bus_stops/static/style.css new file mode 100644 index 0000000..dc2115c --- /dev/null +++ b/src/uk_bus_stops/static/style.css @@ -0,0 +1,43 @@ +html, body { height: 100%; overflow: hidden; } +body { color: #1c2333; } +.navbar { height: 56px; } +.navbar-brand { font-size: 1.05rem; } +#main-row { display: flex; height: calc(100dvh - 56px); } +#sidebar { width: 380px; flex: 0 0 380px; border-right: 1px solid #dee2e6; background: #fff; z-index: 1000; } +#sidebar-inner { height: 100%; overflow-y: auto; padding: 1rem; } +#map { flex: 1; min-width: 0; height: 100%; position: relative; } +#stop-list { padding-bottom: 1rem; } +.stop-item { width: 100%; text-align: left; border: 0; border-bottom: 1px solid #edf0f2; background: #fff; padding: .7rem .5rem; } +.stop-item:hover, .stop-item:focus { background: #f0f5ff; } +.stop-title { display: block; font-weight: 600; } +.stop-meta { display: block; color: #6c757d; font-size: .8rem; margin-top: .1rem; } +.transport-type { display: inline-block; border-radius: 1rem; padding: .15rem .55rem; background: #e9ecef; color: #343a40; font-size: .78rem; font-weight: 600; } +.atco-card { border-radius: .5rem; color: #fff; background: #0f1c2e; padding: .85rem 1rem; } +.atco-card code { color: #b9edac; font-size: 1.35rem; font-weight: 700; letter-spacing: .03em; } +.atco-card .text-uppercase { color: #b9c3cf; font-size: .7rem; letter-spacing: .08em; } +.route-card { border-left: 3px solid #0d6efd; padding: .45rem .6rem; margin: .45rem 0; background: #f8f9fa; } +.route-ref { font-weight: 700; margin-right: .35rem; } +.tag-list { display: grid; grid-template-columns: minmax(7rem, auto) 1fr; gap: .25rem .6rem; overflow-wrap: anywhere; } +.tag-list dt, .tag-list dd { margin: 0; } +.tag-list dt { font-family: monospace; color: #495057; } +.leaflet-container { font-family: inherit; } +.selected-stop-marker { animation: selected-stop-pulse 1.5s ease-in-out infinite; } +@keyframes selected-stop-pulse { + 0%, 100% { stroke-opacity: 1; fill-opacity: .25; stroke-width: 5; } + 50% { stroke-opacity: .55; fill-opacity: .08; stroke-width: 8; } +} +@media (prefers-reduced-motion: reduce) { + .selected-stop-marker { animation: none; } +} +#panel-button { position: absolute; z-index: 900; right: 12px; bottom: 42px; border: 0; border-radius: 24px; padding: .65rem 1rem; background: #fff; box-shadow: 0 2px 9px #0005; font-weight: 600; } +#map-status { position: absolute; z-index: 900; top: 12px; left: 50%; transform: translateX(-50%); max-width: calc(100% - 100px); border-radius: 5px; padding: .4rem .7rem; background: rgba(255,255,255,.94); box-shadow: 0 1px 5px #0004; color: #495057; font-size: .8rem; text-align: center; } +#mobile-handle { height: 25px; align-items: center; justify-content: center; } +#mobile-handle span { width: 38px; height: 4px; border-radius: 2px; background: #ced4da; } + +@media (max-width: 767.98px) { + #sidebar { position: fixed; width: 100%; height: 68dvh; left: 0; right: 0; bottom: 0; border: 0; border-radius: 16px 16px 0 0; box-shadow: 0 -4px 18px #0003; transform: translateY(100%); transition: transform .25s ease; padding-bottom: env(safe-area-inset-bottom); } + #sidebar.panel-open { transform: translateY(0); } + #mobile-handle { display: flex !important; cursor: pointer; } + #sidebar-inner { height: calc(100% - 25px); padding-top: .4rem; } + #map { width: 100%; flex: 0 0 100%; } +} diff --git a/src/uk_bus_stops/templates/index.html b/src/uk_bus_stops/templates/index.html new file mode 100644 index 0000000..e02289a --- /dev/null +++ b/src/uk_bus_stops/templates/index.html @@ -0,0 +1,110 @@ + + + + + + + UK Bus Stop ATCO Code Finder + + + + + + + +
+ + +
+
+ +
+
+ + + + + + diff --git a/src/uk_bus_stops/wsgi.py b/src/uk_bus_stops/wsgi.py new file mode 100644 index 0000000..8579bff --- /dev/null +++ b/src/uk_bus_stops/wsgi.py @@ -0,0 +1,9 @@ +"""Production WSGI entry point for the UK bus stop finder.""" + +from werkzeug.middleware.proxy_fix import ProxyFix + +from uk_bus_stops.app import app + +app.wsgi_app = ProxyFix( # type: ignore[method-assign] + app.wsgi_app, x_for=1, x_proto=1, x_host=1, x_prefix=1 +) diff --git a/tests/test_uk_bus_stops.py b/tests/test_uk_bus_stops.py new file mode 100644 index 0000000..83c7b95 --- /dev/null +++ b/tests/test_uk_bus_stops.py @@ -0,0 +1,213 @@ +"""Tests for the UK bus stop finder backend.""" + +from __future__ import annotations + +from typing import Any + +import pytest +import responses + +from uk_bus_stops import core +from uk_bus_stops.app import ATCO_PATTERN, create_app, parse_coordinates + + +@pytest.fixture() +def client() -> Any: + """Return a Flask test client.""" + return create_app({"TESTING": True}).test_client() + + +@responses.activate +def test_geocode_limits_search_to_great_britain_and_identifies_app() -> None: + """Nominatim receives the UK restriction and identifying User-Agent.""" + responses.get( + core.NOMINATIM_URL, + json=[{"lat": "51.45", "lon": "-2.59", "display_name": "Bristol, England"}], + ) + result = core.geocode("Bristol") + assert result == {"lat": 51.45, "lon": -2.59, "label": "Bristol, England"} + request = responses.calls[0].request + assert "countrycodes=gb" in request.url + assert request.headers["User-Agent"].startswith("uk-bus-stops/") + assert "edward@4angle.com" in request.headers["User-Agent"] + + +@responses.activate +def test_nearby_stops_normalises_deduplicates_and_orders() -> None: + """Nearby Overpass objects become nearest-first stop records without duplicates.""" + responses.get( + core.OVERPASS_URL, + json={"elements": [ + {"type": "node", "id": 2, "lat": 51.46, "lon": -2.58, + "tags": {"highway": "bus_stop", "name": "Far Stop"}}, + {"type": "node", "id": 1, "lat": 51.4501, "lon": -2.5901, + "tags": {"naptan:CommonName": "Near Stop", "naptan:AtcoCode": "0100BRP90314", + "naptan:Bearing": "NW"}}, + {"type": "node", "id": 1, "lat": 51.4501, "lon": -2.5901, + "tags": {"name": "Near Stop", "naptan:AtcoCode": "0100BRP90314", + "naptan:Bearing": "NW"}}, + ]}, + ) + result = core.nearby_stops(51.45, -2.59) + assert [stop["id"] for stop in result] == [1, 2] + assert result[0]["atco_code"] == "0100BRP90314" + assert result[0]["transport_type"] == "Transport stop" + assert result[0]["bearing"] == "NW" + assert result[1]["atco_code"] is None + assert "highway%22%3D%22bus_stop" in responses.calls[0].request.url + + +@responses.activate +def test_find_atco_code_is_case_insensitive() -> None: + """Direct code lookup emits an anchored, case-insensitive Overpass query.""" + responses.get( + core.OVERPASS_URL, + json={"elements": [{ + "type": "node", "id": 123, "lat": 51.45, "lon": -2.59, + "tags": {"name": "Central", "naptan:AtcoCode": "0100BRP90314"}, + }]}, + ) + result = core.find_atco_code("0100brp90314") + assert result[0]["name"] == "Central" + assert "%5E0100brp90314%24%22%2Ci" in responses.calls[0].request.url + + +@responses.activate +def test_fetch_stop_by_osm_reference() -> None: + """A shared stop reference is resolved to normalised stop details.""" + responses.get(core.OVERPASS_URL, json={"elements": [{ + "type": "node", "id": 123, "lat": 51.45, "lon": -2.59, + "tags": {"name": "Central", "highway": "bus_stop"}, + }]}) + stop = core.fetch_stop("node", 123) + assert stop is not None + assert stop["id"] == 123 + assert stop["transport_type"] == "Bus stop" + + +@responses.activate +def test_overpass_uses_fallback_instance() -> None: + """An unavailable UK Overpass host is retried on the global fallback.""" + responses.get(core.OVERPASS_URL, status=503) + responses.get(core.OVERPASS_FALLBACK_URLS[0], json={"elements": []}) + assert core.find_atco_code("0100BRP90314") == [] + assert len(responses.calls) == 2 + assert responses.calls[1].request.url.startswith(core.OVERPASS_FALLBACK_URLS[0]) + + +@responses.activate +def test_routes_for_stop_returns_useful_route_fields() -> None: + """Route lookup uses stop-area members and returns human-readable metadata.""" + responses.get( + core.OVERPASS_URL, + json={"elements": [ + {"type": "relation", "id": 9, "tags": { + "ref": "A1", "name": "Airport bus", "operator": "Example Bus", + "from": "Airport", "to": "City Centre", + }}, + ]}, + ) + routes = core.routes_for_stop("node", 123) + assert routes == [{ + "id": 9, "ref": "A1", "name": "Airport bus", "operator": "Example Bus", + "from": "Airport", "to": "City Centre", + }] + query_url = responses.calls[0].request.url + assert "stop_area" in query_url + assert "route" in query_url + + +@pytest.mark.parametrize( + ("tags", "expected"), + [ + ({"highway": "bus_stop"}, "Bus stop"), + ({"railway": "station"}, "Railway station"), + ({"railway": "tram_stop"}, "Tram stop"), + ({"station": "subway"}, "Underground station"), + ({"amenity": "ferry_terminal"}, "Ferry terminal"), + ], +) +def test_transport_type_classifies_osm_tags(tags: dict[str, str], expected: str) -> None: + """Common OSM public transport tagging receives a clear display type.""" + assert core.transport_type(tags) == expected + + +def test_search_rejects_empty_query(client: Any) -> None: + """The search endpoint returns the standard JSON error shape.""" + response = client.get("/api/search") + assert response.status_code == 400 + assert response.get_json()["error"] == "missing_query" + + +def test_place_name_is_not_mistaken_for_atco_code() -> None: + """Long alphabetic place names still go through Nominatim geocoding.""" + assert ATCO_PATTERN.fullmatch("Manchester") is None + assert ATCO_PATTERN.fullmatch("0100BRP90314") is not None + + +def test_parse_coordinates_accepts_valid_pair_and_rejects_invalid_pair() -> None: + """Coordinate searches require a comma and valid latitude/longitude ranges.""" + assert parse_coordinates(" 51.4545, -2.5879 ") == (51.4545, -2.5879) + assert parse_coordinates("51.4545 -2.5879") is None + assert parse_coordinates("91, -2") is None + + +@responses.activate +def test_search_coordinates_bypasses_nominatim(client: Any) -> None: + """Latitude/longitude entered in the search endpoint goes straight to Overpass.""" + responses.get(core.OVERPASS_URL, json={"elements": []}) + response = client.get("/api/search?q=51.4545%2C+-2.5879") + assert response.status_code == 200 + data = response.get_json() + assert data["location"]["lat"] == 51.4545 + assert data["location"]["lon"] == -2.5879 + assert len(responses.calls) == 1 + assert responses.calls[0].request.url.startswith(core.OVERPASS_URL) + + +def test_location_coordinates_are_validated(client: Any) -> None: + """Latitude and longitude must be numeric and within geographic ranges.""" + assert client.get("/api/stops?lat=hello&lon=1").status_code == 400 + assert client.get("/api/stops?lat=91&lon=1").status_code == 400 + + +def test_map_bounds_reject_large_overpass_area(client: Any) -> None: + """The server refuses viewport queries that are too large for Overpass.""" + response = client.get("/api/stops/in-bounds?south=50&west=-3&north=51&east=-2") + assert response.status_code == 400 + assert response.get_json()["error"] == "area_too_large" + + +@responses.activate +def test_shared_stop_api(client: Any) -> None: + """The stop detail endpoint supports refreshing a shared stop URL.""" + responses.get(core.OVERPASS_URL, json={"elements": [{ + "type": "node", "id": 123, "lat": 51.45, "lon": -2.59, + "tags": {"name": "Central", "highway": "bus_stop"}, + }]}) + response = client.get("/api/stop/node/123") + assert response.status_code == 200 + assert response.get_json()["stop"]["name"] == "Central" + + +@responses.activate +def test_map_bounds_load_small_area(client: Any) -> None: + """A small viewport returns classified stops from Overpass.""" + responses.get(core.OVERPASS_URL, json={"elements": [{ + "type": "node", "id": 4, "lat": 51.45, "lon": -2.59, + "tags": {"railway": "station", "name": "Temple Meads"}, + }]}) + response = client.get( + "/api/stops/in-bounds?south=51.44&west=-2.60&north=51.46&east=-2.58" + ) + assert response.status_code == 200 + assert response.get_json()["stops"][0]["transport_type"] == "Railway station" + + +def test_index_contains_search_and_map(client: Any) -> None: + """The landing page includes the primary search and map controls.""" + response = client.get("/") + assert response.status_code == 200 + assert b"Find a UK bus stop code" in response.data + assert b'id="search-input"' in response.data + assert b'id="map"' in response.data diff --git a/tests/test_uk_bus_stops_playwright.py b/tests/test_uk_bus_stops_playwright.py new file mode 100644 index 0000000..d258d8c --- /dev/null +++ b/tests/test_uk_bus_stops_playwright.py @@ -0,0 +1,194 @@ +"""Browser tests for the UK bus stop finder using Python Playwright.""" + +from __future__ import annotations + +import re +import threading +from collections.abc import Iterator +from typing import Any + +import pytest +from werkzeug.serving import make_server + +from uk_bus_stops.app import create_app + +playwright_api = pytest.importorskip("playwright.sync_api") + + +@pytest.fixture(scope="module") +def flask_url() -> Iterator[str]: + """Run the Flask application on a temporary local port.""" + server = make_server("127.0.0.1", 0, create_app({"TESTING": True})) + thread = threading.Thread(target=server.serve_forever, daemon=True) + thread.start() + yield f"http://127.0.0.1:{server.server_port}" + server.shutdown() + thread.join() + + +@pytest.fixture(scope="module") +def chromium_browser() -> Iterator[Any]: + """Launch the installed Chromium browser through Python Playwright.""" + with playwright_api.sync_playwright() as playwright: + instance = playwright.chromium.launch(headless=True) + yield instance + instance.close() + + +def test_search_and_stop_details_in_browser(chromium_browser: Any, flask_url: str) -> None: + """A location search renders stops, an ATCO code, tags, and route metadata.""" + page = chromium_browser.new_page(viewport={"width": 1280, "height": 800}) + page_errors: list[str] = [] + page.on("pageerror", lambda error: page_errors.append(str(error))) + stop = { + "type": "node", "id": 123, "lat": 51.451, "lon": -2.591, + "name": "Central Stop", "atco_code": "0100BRP90314", "indicator": "N", + "bearing": "NW", + "transport_type": "Bus stop", + "tags": {"highway": "bus_stop", "shelter": "yes"}, + } + page.route("**/api/search?*", lambda route: route.fulfill(json={ + "kind": "location", + "location": {"lat": 51.45, "lon": -2.59, "label": "Bristol"}, + "stops": [{ + "type": "node", "id": 125, "lat": 51.47, "lon": -2.61, + "name": "Far Stop", "atco_code": "0100BRP90316", "indicator": None, + "bearing": "E", "transport_type": "Bus stop", + "tags": {"highway": "bus_stop"}, + }, stop], + })) + page.route("**/api/stops?*", lambda route: route.fulfill(json={ + "kind": "location", + "location": {"lat": 51.451, "lon": -2.591, "label": "Central Stop"}, + "stops": [stop, { + "type": "node", "id": 124, "lat": 51.452, "lon": -2.592, + "name": "Nearby Stop", "atco_code": "0100BRP90315", "indicator": "S", + "bearing": "SE", "transport_type": "Bus stop", + "tags": {"highway": "bus_stop"}, + }], + })) + page.route("**/api/stops/in-bounds?*", lambda route: route.fulfill(json={ + "kind": "map", + "stops": [stop, { + "type": "node", "id": 124, "lat": 51.452, "lon": -2.592, + "name": "Nearby Stop", "atco_code": "0100BRP90315", "indicator": "S", + "bearing": "SE", "transport_type": "Bus stop", + "tags": {"highway": "bus_stop"}, + }], + })) + page.route("**/api/stop/node/123", lambda route: route.fulfill(json={"stop": stop})) + page.route("**/api/stop/node/123/routes", lambda route: route.fulfill(json={"routes": [{ + "id": 9, "ref": "A1", "name": "Airport bus", "operator": "Example Bus", + "from": "Airport", "to": "City Centre", + }]})) + page.goto(flask_url, wait_until="networkidle") + page.get_by_label("Location or ATCO code").fill("Bristol") + page.get_by_role("button", name="Search").click() + playwright_api.expect(page).to_have_url(re.compile(r"\?q=Bristol$")) + playwright_api.expect(page.get_by_text("Central Stop", exact=True)).to_be_visible() + assert page.locator("#stop-list .stop-title").all_inner_texts() == ["Central Stop", "Far Stop"] + playwright_api.expect(page.locator("#stop-list .stop-meta").first).to_contain_text("m · Bus stop") + playwright_api.expect(page.locator("#stop-list .stop-meta").first).to_contain_text("NW-bound") + assert page.evaluate("map.getZoom()") > 6 + page.evaluate("markers.getLayers()[0].openTooltip()") + playwright_api.expect(page.get_by_text("Central Stop · NW-bound · Bus stop", exact=True)).to_be_visible() + page.get_by_text("Central Stop", exact=True).click() + playwright_api.expect(page).to_have_url(re.compile(r"\?node=123$")) + playwright_api.expect(page.locator("#atco-code")).to_have_text("0100BRP90314") + playwright_api.expect(page.locator("#stop-type")).to_have_text("Bus stop") + playwright_api.expect(page.locator("#stop-bearing")).to_have_text("NW-bound") + playwright_api.expect(page.locator(".selected-stop-marker")).to_be_visible() + playwright_api.expect(page.get_by_text("Airport → City Centre · Example Bus", exact=True)).to_be_visible() + assert page_errors == [] + page.reload(wait_until="networkidle") + playwright_api.expect(page.get_by_label("Location or ATCO code")).to_have_value("") + playwright_api.expect(page.locator("#stop-name")).to_have_text("Central Stop") + playwright_api.expect(page.locator("#atco-code")).to_have_text("0100BRP90314") + playwright_api.expect(page.locator("#stop-bearing")).to_have_text("NW-bound") + playwright_api.expect(page.get_by_role("button", name="All nearby stops")).to_be_visible() + page.get_by_role("button", name="All nearby stops").click() + playwright_api.expect(page.get_by_text("Nearby Stop", exact=True)).to_be_visible() + playwright_api.expect(page).to_have_url(re.compile(r"\?lat=.*&lon=.*$")) + page.close() + + +def test_moving_map_loads_only_when_zoomed_in(chromium_browser: Any, flask_url: str) -> None: + """Panning a close map loads visible stops while a wide map makes no query.""" + page = chromium_browser.new_page(viewport={"width": 1280, "height": 800}) + requests: list[str] = [] + + def fulfil_bounds(route: Any) -> None: + """Record viewport calls and return a railway station.""" + requests.append(route.request.url) + route.fulfill(json={"kind": "map", "stops": [{ + "type": "node", "id": 456, "lat": 51.45, "lon": -2.59, + "name": "Temple Meads", "atco_code": "9100BRSTLTM", "indicator": None, + "transport_type": "Railway station", "tags": {"railway": "station"}, + }]}) + + page.route("**/api/stops/in-bounds?*", fulfil_bounds) + page.route("**/api/stop/node/456/routes", lambda route: route.fulfill(json={"routes": []})) + page.goto(flask_url, wait_until="networkidle") + page.evaluate("userMapInteractionPending = true; map.setView([51.45, -2.59], 15)") + playwright_api.expect(page.get_by_text("Temple Meads", exact=True)).to_be_visible() + assert len(requests) == 1 + page.evaluate("userMapInteractionPending = true; map.setZoom(10)") + playwright_api.expect(page.locator("#map-status")).to_have_text( + "Zoom in to load transport stops" + ) + page.wait_for_timeout(500) + assert len(requests) == 1 + page.evaluate("userMapInteractionPending = true; map.setZoom(15)") + page.wait_for_timeout(700) + assert len(requests) == 2 + page.get_by_text("Temple Meads", exact=True).click() + playwright_api.expect(page.locator("#stop-name")).to_have_text("Temple Meads") + page.wait_for_timeout(700) + playwright_api.expect(page.locator("#stop-detail")).to_be_visible() + assert len(requests) == 3 + page.evaluate("userMapInteractionPending = true; map.panBy([80, 0])") + page.wait_for_timeout(700) + playwright_api.expect(page.locator("#stop-name")).to_have_text("Temple Meads") + playwright_api.expect(page.locator("#stop-detail")).to_be_visible() + playwright_api.expect(page).to_have_url(re.compile(r"\?node=456$")) + assert len(requests) == 4 + page.close() + + +def test_coordinate_url_and_search_input(chromium_browser: Any, flask_url: str) -> None: + """lat/lon URLs load on refresh and coordinate input creates a shareable URL.""" + page = chromium_browser.new_page(viewport={"width": 1280, "height": 800}) + requested_urls: list[str] = [] + + def fulfil_stops(route: Any) -> None: + """Record coordinate API calls and return an empty nearby result.""" + requested_urls.append(route.request.url) + route.fulfill(json={ + "kind": "location", + "location": {"lat": 51.4545, "lon": -2.5879, "label": "Coordinates"}, + "stops": [], + }) + + page.route("**/api/stops?*", fulfil_stops) + page.goto(f"{flask_url}?lat=51.4545&lon=-2.5879", wait_until="networkidle") + playwright_api.expect(page.get_by_label("Location or ATCO code")).to_have_value( + "51.4545, -2.5879" + ) + playwright_api.expect(page.locator("#stop-count")).to_have_text("0 found") + assert page.evaluate("map.getZoom()") == 16 + assert "lat=51.4545" in requested_urls[-1] + + page.get_by_label("Location or ATCO code").fill("51.46, -2.58") + page.get_by_role("button", name="Search").click() + playwright_api.expect(page).to_have_url(re.compile(r"\?lat=51.46&lon=-2.58$")) + playwright_api.expect(page.locator("#stop-count")).to_have_text("0 found") + page.close() + + +def test_mobile_layout_opens_results_sheet(chromium_browser: Any, flask_url: str) -> None: + """The finder exposes its search panel as an open mobile bottom sheet.""" + page = chromium_browser.new_page(viewport={"width": 390, "height": 844}) + page.goto(flask_url, wait_until="domcontentloaded") + playwright_api.expect(page.locator("#sidebar")).to_have_class(re.compile("panel-open")) + playwright_api.expect(page.get_by_label("Location or ATCO code")).to_be_visible() + page.close()