From a79b0a7098fec0ef968b4e240d9c7a79c4263974 Mon Sep 17 00:00:00 2001 From: Edward Betts Date: Wed, 30 Sep 2026 22:23:45 +0100 Subject: [PATCH] Make public transport routes discoverable with shareable search 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 --- README.md | 15 ++ src/osm_geojson/pt/core.py | 148 ++++++++++++++ tests/test_pt_search.py | 388 +++++++++++++++++++++++++++++++++++++ web/app.py | 40 +++- web/static/app.js | 224 +++++++++++++++++++++ web/templates/api.html | 32 ++- web/templates/index.html | 84 +++++++- 7 files changed, 919 insertions(+), 12 deletions(-) create mode 100644 tests/test_pt_search.py diff --git a/README.md b/README.md index 94f1a10..9f6afe7 100644 --- a/README.md +++ b/README.md @@ -76,6 +76,19 @@ routes as GeoJSON. #### Features +- Search worldwide for public transport routes by route number, name, + destination, operator or network, with an optional transport type filter. + Click a result to load it; search uses Overpass and shows up to 50 matches. +- Narrow a route search with a location (for example, route `24`, location + `Bristol, UK`, transport type `Bus`). Choose a matching place to centre the + map and show its search boundary, or tick **Search this map area** to use + the current view instead. Routes may extend outside the search area. +- Share or bookmark the page URL after searching. It saves the route text, + transport type, selected location and exact area; opening it automatically + restores the filters and matching route list. Back and Forward restore + previous searches. Results reflect the latest OpenStreetMap data. +- Start with one-click examples: Bristol airport bus A1, Istanbul Metro M11, + or all M11 directions. No relation ID lookup is needed to try the tool. - Enter a relation ID or OSM URL to load a route onto the map. - Click stops in the sidebar list or on the map to set start and end points. - Preview the selected segment highlighted on the map. @@ -101,6 +114,8 @@ when the server is running. | Endpoint | Description | |---|---| +| `GET /api/search?q=TEXT[&mode=TYPE][&south=S&west=W&north=N&east=E]` | Find public transport route relations, optionally within an area | +| `GET /api/locations?q=PLACE` | Find candidate locations and their search bounds | | `GET /api/route/` | Full route GeoJSON, stop list, and sibling routes | | `GET /api/segment/?from=NAME&to=NAME` | Segment between two named stops | | `GET /api/route_master/` | All member routes of a route_master | diff --git a/src/osm_geojson/pt/core.py b/src/osm_geojson/pt/core.py index 9463c8d..0b4448f 100644 --- a/src/osm_geojson/pt/core.py +++ b/src/osm_geojson/pt/core.py @@ -1,16 +1,32 @@ """Core data-fetching and processing functions for osm-pt-geojson.""" +import json +import math +import re +import threading +import time import warnings +from functools import lru_cache from typing import Any import requests OSM_API = "https://www.openstreetmap.org/api/0.6" +OVERPASS_URL = "https://overpass.atownsend.org.uk/api/interpreter" +PUBLIC_TRANSPORT_ROUTE_TYPES = { + "bus", "trolleybus", "tram", "subway", "train", + "light_rail", "monorail", "ferry", "funicular", +} +SEARCH_LIMIT = 50 +NOMINATIM_URL = "https://nominatim.openstreetmap.org/search" +_location_lock = threading.Lock() +_last_location_request = 0.0 # Type aliases Coord = list[float] # [lon, lat] OsmTags = dict[str, str] OsmElement = dict[str, Any] GeoJson = dict[str, Any] +Bounds = tuple[float, float, float, float] # south, west, north, east class OsmError(Exception): @@ -22,6 +38,138 @@ class OsmError(Exception): self.status_code = status_code +def validate_search_bounds(bounds: Bounds) -> None: + """Reject invalid, non-finite or excessively broad route search areas.""" + south, west, north, east = bounds + if not all(math.isfinite(value) for value in bounds) or not ( + -90 <= south < north <= 90 and -180 <= west < east <= 180 + ): + raise OsmError("Choose a valid search area that does not cross the date line.", 400) + if north - south > 20 or east - west > 20: + raise OsmError("Zoom in or choose a smaller location to search for routes.", 400) + + +def find_locations(query: str) -> list[dict[str, Any]]: + """Resolve a submitted place name to candidate areas, with cached results.""" + query = query.strip() + if not 2 <= len(query) <= 100 or any(ord(char) < 32 for char in query): + raise OsmError("Enter between 2 and 100 characters for the location.", 400) + # Serialize cache misses and space requests at least one second apart. + with _location_lock: + return _find_locations_cached(query) + + +@lru_cache(maxsize=256) +def _find_locations_cached(query: str) -> list[dict[str, Any]]: + """Fetch place bounds from Nominatim on explicit searches, never autocomplete.""" + global _last_location_request + delay = 1 - (time.monotonic() - _last_location_request) + if delay > 0: + time.sleep(delay) + _last_location_request = time.monotonic() + try: + response = requests.get( + NOMINATIM_URL, + params={"q": query, "format": "jsonv2", "limit": "5", "dedupe": "1"}, + headers={ + "User-Agent": "osm-pt-geojson/1.0 (https://openstreetmap.tools)", + "Accept": "application/json", + }, timeout=15, + ) + response.raise_for_status() + data = response.json() + except (requests.RequestException, ValueError) as exc: + raise OsmError("Location search is unavailable. Please try again shortly.", 502) from exc + if not isinstance(data, list): + raise OsmError("Location search returned an unexpected response.", 502) + locations = [] + for item in data: + try: + south, north, west, east = (float(value) for value in item["boundingbox"]) + bounds = (south, west, north, east) + # Broad place results remain selectable; route search explains zooming in. + if not all(math.isfinite(value) for value in bounds) or not ( + -90 <= south < north <= 90 and -180 <= west < east <= 180 + ): + continue + locations.append({"label": item["display_name"], "bounds": bounds}) + except (KeyError, TypeError, ValueError): + continue + return locations + + +def search_routes(query: str, mode: str = "", bounds: Bounds | None = None) -> dict[str, Any]: + """Search public transport relations by literal text or exact route number. + + Names, termini, operators and networks are matched case-insensitively. + Fetch one extra result to indicate when the output limit is exceeded. + """ + query = query.strip() + if not 2 <= len(query) <= 100 or any(ord(char) < 32 for char in query): + raise OsmError("Enter between 2 and 100 characters to search for a route.", 400) + if mode and mode not in PUBLIC_TRANSPORT_ROUTE_TYPES: + raise OsmError("Choose a supported transport type.", 400) + if bounds is not None: + validate_search_bounds(bounds) + # Overpass uses POSIX ERE; escape only its metacharacters, then quote for QL. + literal = re.sub(r'([.\[\]\\*+?{}()|^$])', r'\\\1', query) + text_pattern = json.dumps(literal, ensure_ascii=False) + ref_pattern = json.dumps(f"^{literal}$", ensure_ascii=False) + modes = mode or "|".join(sorted(PUBLIC_TRANSPORT_ROUTE_TYPES)) + spatial_query = "" + input_set = "" + if bounds is not None: + bbox = ",".join(f"{value:.7f}" for value in bounds) + spatial_query = ( + f'rel({bbox})["type"="route"]["route"~"^({modes})$"]->.areaRoutes;' + 'rel(br.areaRoutes)["type"="route_master"]->.areaMasters;' + '(.areaRoutes;.areaMasters;)->.inArea;' + ) + input_set = ".inArea" + selectors = [] + for kind, mode_tag in (("route", "route"), ("route_master", "route_master")): + base = f'rel{input_set}["type"="{kind}"]["{mode_tag}"~"^({modes})$"]' + selectors.append(f'{base}["ref"~{ref_pattern},i];') + for key in ("name", "from", "to", "operator", "network"): + selectors.append(f'{base}["{key}"~{text_pattern},i];') + overpass_query = "[out:json][timeout:25];" + spatial_query + "(" + "".join(selectors) + f");out tags {SEARCH_LIMIT + 1};" + try: + response = requests.post( + OVERPASS_URL, data={"data": overpass_query}, + headers={"User-Agent": "osm-pt-geojson/1.0"}, timeout=35, + ) + response.raise_for_status() + data = response.json() + except (requests.RequestException, ValueError) as exc: + raise OsmError("Route search is unavailable. Please try again shortly.", 502) from exc + if not isinstance(data, dict) or data.get("remark") or not isinstance(data.get("elements"), list): + raise OsmError("Route search could not complete. Try a more specific search.", 502) + routes = [] + for element in data["elements"]: + if not isinstance(element, dict) or element.get("type") != "relation": + continue + tags = element.get("tags", {}) + if not isinstance(tags, dict): + continue + relation_type = tags.get("type") + route_mode = tags.get("route_master" if relation_type == "route_master" else "route") + if relation_type not in ("route", "route_master") or route_mode not in PUBLIC_TRANSPORT_ROUTE_TYPES: + continue + if not isinstance(element.get("id"), int): + continue + routes.append({ + "id": element["id"], "name": tags.get("name") or tags.get("ref") or str(element["id"]), + "ref": tags.get("ref"), "from": tags.get("from"), "to": tags.get("to"), + "operator": tags.get("operator"), "network": tags.get("network"), + "route": route_mode, "type": relation_type, + }) + routes.sort(key=lambda route: ( + str(route["ref"] or "").casefold() != query.casefold(), + str(route["name"]).casefold(), route["id"], + )) + return {"routes": routes[:SEARCH_LIMIT], "truncated": len(routes) > SEARCH_LIMIT} + + def fetch_relation_full(relation_id: int) -> dict[str, Any]: """Fetch the full OSM API response for a relation, including all member ways and nodes.""" url = f"{OSM_API}/relation/{relation_id}/full.json" diff --git a/tests/test_pt_search.py b/tests/test_pt_search.py new file mode 100644 index 0000000..2e08e06 --- /dev/null +++ b/tests/test_pt_search.py @@ -0,0 +1,388 @@ +"""Offline tests for public transport relation search and its JSON endpoint.""" + +import importlib.util +import json +import sys +from pathlib import Path +from typing import Any +from urllib.parse import parse_qs + +import pytest +import requests +import responses + +from osm_geojson.pt import core + + +def relation(relation_id: int = 1, **tags: str) -> dict[str, Any]: + """Build a minimal tagged Overpass relation.""" + return {"type": "relation", "id": relation_id, "tags": {"type": "route", "route": "bus", **tags}} + + +@pytest.fixture() +def app(monkeypatch: Any) -> Any: + """Load the web application independently of Python's import path.""" + spec = importlib.util.spec_from_file_location("pt_web_app", Path(__file__).parents[1] / "web/app.py") + assert spec and spec.loader + module = importlib.util.module_from_spec(spec) + monkeypatch.setitem(sys.modules, spec.name, module) + spec.loader.exec_module(module) + module.app.config["TESTING"] = True + return module.app + + +@responses.activate +def test_search_metadata_and_query() -> None: + """Return routes and masters, exclude other relations, and prefer exact refs.""" + responses.post(core.OVERPASS_URL, json={"elements": [ + relation(2, name="Airport shuttle", ref="X1", operator="City Bus", **{"from": "Airport", "to": "Centre"}), + relation(3, name="M11", type="route_master", route_master="subway", ref="M11"), + relation(1, name="Metro M11", route="subway", ref="M11"), + relation(4, route="hiking"), + {"type": "node", "id": 5}, + ]}) + result = core.search_routes(" M11 ") + assert [item["id"] for item in result["routes"]] == [3, 1, 2] + assert result["routes"][0]["type"] == "route_master" + assert result["routes"][2]["operator"] == "City Bus" + assert result["truncated"] is False + request = responses.calls[0].request + query = parse_qs(request.body)["data"][0] + assert '["ref"~"^M11$",i]' in query + assert '["route_master"~' in query + assert 'out tags 51;' in query + assert request.headers["User-Agent"] == "osm-pt-geojson/1.0" + + +@responses.activate +def test_literal_search_and_mode_filter() -> None: + """User text stays a literal regex and cannot inject Overpass selectors.""" + responses.post(core.OVERPASS_URL, json={"elements": []}) + text = 'İstanbul [A.*] "\\);out;' + assert core.search_routes(text, "subway")["routes"] == [] + query = parse_qs(responses.calls[0].request.body)["data"][0] + escaped = 'İstanbul \\[A\\.\\*\\] "\\\\\\);out;' + assert f'["name"~{json.dumps(escaped, ensure_ascii=False)},i]' in query + assert '["route"~"^(subway)$"]' in query + assert '["route_master"~"^(subway)$"]' in query + + +@pytest.mark.parametrize("query,mode", [("", ""), ("x", ""), ("x" * 101, ""), ("a\nb", ""), ("M11", "hiking")]) +@responses.activate +def test_invalid_search(query: str, mode: str) -> None: + """Reject invalid input without contacting Overpass.""" + with pytest.raises(core.OsmError) as error: + core.search_routes(query, mode) + assert error.value.status_code == 400 + assert len(responses.calls) == 0 + + +@responses.activate +def test_search_limit() -> None: + """Tell the UI when more than fifty relations matched.""" + responses.post(core.OVERPASS_URL, json={"elements": [relation(i) for i in range(51)]}) + result = core.search_routes("Airport") + assert len(result["routes"]) == 50 + assert result["truncated"] is True + + +@pytest.mark.parametrize("payload", [{"remark": "runtime error: Query timed out", "elements": []}, {}, []]) +@responses.activate +def test_incomplete_search(payload: Any) -> None: + """A timed-out or malformed response must not look like an empty result.""" + responses.post(core.OVERPASS_URL, json=payload) + with pytest.raises(core.OsmError) as error: + core.search_routes("M11") + assert error.value.status_code == 502 + + +@pytest.mark.parametrize("failure", [requests.Timeout("timeout"), "invalid JSON", 429, 504]) +@responses.activate +def test_upstream_search_failure(failure: Any) -> None: + """Network, rate-limit and decoding failures become library errors.""" + if isinstance(failure, int): + responses.post(core.OVERPASS_URL, status=failure) + else: + responses.post(core.OVERPASS_URL, body=failure) + with pytest.raises(core.OsmError) as error: + core.search_routes("M11") + assert error.value.status_code == 502 + + +@responses.activate +def test_search_api(app: Any) -> None: + """The endpoint handles results, empty matches, invalid inputs and upstream errors.""" + responses.post(core.OVERPASS_URL, json={"elements": [relation(15083963, name="M11", ref="M11")]}) + client = app.test_client() + result = client.get("/api/search?q=M11&mode=subway") + assert result.status_code == 200 + assert result.json["routes"][0]["id"] == 15083963 + responses.post(core.OVERPASS_URL, json={"elements": []}) + assert client.get("/api/search?q=Unknown").json == {"routes": [], "truncated": False} + for path in ("/api/search", "/api/search?q=M11&mode=hiking"): + result = client.get(path) + assert result.status_code == 400 + assert result.json["error"] == "invalid_query" + responses.post(core.OVERPASS_URL, status=503) + result = client.get("/api/search?q=M11") + assert result.status_code == 502 + assert result.json["error"] == "osm_error" + + +@pytest.mark.parametrize("width", [1280, 390]) +def test_search_browser(app: Any, width: int, monkeypatch: Any) -> None: + """Search results load routes on desktop and mobile, with safe text and errors.""" + import threading + from werkzeug.serving import make_server + + browser_api = pytest.importorskip("playwright.sync_api") + monkeypatch.setitem(app.view_functions["route_page"].__globals__, "fetch_relation_name", lambda _: "Metro M11") + server = make_server("127.0.0.1", 0, app) + thread = threading.Thread(target=server.serve_forever, daemon=True) + thread.start() + try: + with browser_api.sync_playwright() as playwright: + browser = playwright.chromium.launch() + page = browser.new_page(viewport={"width": width, "height": 844}) + errors: list[str] = [] + location_requests: list[str] = [] + page.on("pageerror", lambda error: errors.append(str(error))) + page.on('request', lambda request: location_requests.append(request.url) if '/api/locations?' in request.url else None) + page.route("**/tile.openstreetmap.org/**", lambda route: route.abort()) + page.route("**/api/locations?*", lambda route: route.fulfill(json={"locations": [ + {"label": "Bristol, UK", "bounds": [51.39, -2.73, 51.55, -2.51]}, + {"label": "Bristol, USA", "bounds": [41.6, -72.9, 41.8, -72.8]}, + ]})) + page.route("**/api/search?*", lambda route: route.fulfill(json={ + "routes": [{"id": 15083963, "name": "Metro M11