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