diff --git a/.gitignore b/.gitignore index b66a79b..bce175a 100644 --- a/.gitignore +++ b/.gitignore @@ -6,3 +6,8 @@ build/ dist/ .venv/ *.geojson + +# Local NaPTAN downloads and generated spatial database +/Stops.csv +/NaPTAN.xml +/data/ diff --git a/README.md b/README.md index 9a33fa7..94f1a10 100644 --- a/README.md +++ b/README.md @@ -109,33 +109,50 @@ 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. Postcode, street, and place searches show up to 20 -UK-only Nominatim matches for the user to choose from. 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. -Postcode searches with exactly one Nominatim match skip the choice screen. -The current map bounds bias Nominatim's ranking without excluding UK matches -outside the visible area. +A mobile-friendly map for finding ATCO codes using the Department for Transport's +[NaPTAN dataset](https://beta-naptan.dft.gov.uk/download). It covers active public +transport stops in England, Scotland and Wales, including bus, rail, tram, +ferry and airport access points; Northern Ireland is not included. -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. +Search by postcode, street, place, exact ATCO code, coordinates or browser +location. Place searches use Nominatim, with the visible map as a ranking bias. +The map loads stops at zoom level 15 or closer. Selecting a stop shows its +NaPTAN fields and looks up bus routes in OpenStreetMap by matching its ATCO code. +OSM route coverage may be incomplete; NaPTAN itself does not contain routes or +live departures. An Overpass outage does not prevent NaPTAN stop searches. -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. +Searches have shareable `?q=...` and `?lat=...&lon=...` URLs; a selected NaPTAN +stop uses `?naptan=010000056`. Existing `?node=...`, `?way=...` and +`?relation=...` OSM links still work. -Run it locally with: +Download the national CSV from NaPTAN, then import it and start the app: +```sh +python3 -m venv .venv +.venv/bin/pip install -e ".[web]" +.venv/bin/uk-bus-stops-import-naptan Stops.csv +.venv/bin/flask --app uk_bus_stops.app run ``` -pip install -e ".[web]" -flask --app uk_bus_stops.app run -``` + +The importer uses SQLite's built-in R-tree spatial index; no database server or +SpatiaLite extension is required. It preserves leading zeros in ATCO codes, +converts British National Grid coordinates when latitude/longitude are absent, +and reports imported, inactive/pending, invalid and converted row counts. +Only active records with usable positions are included. The original CSV fields +remain available in the detail panel. XML is not needed. + +The default database is `data/naptan.sqlite3` in the project directory. Set +`NAPTAN_DATABASE=/absolute/path/naptan.sqlite3` for both import and serving, or +use `--database PATH` for the importer. The serving process needs read access. +To refresh, download a new CSV and rerun the same command. It builds a temporary +database alongside the destination and replaces the old snapshot atomically; +a failed import leaves the old snapshot intact. Downloads and databases are +excluded from git. The app returns a JSON 503 with import instructions if its +database is missing or unreadable. + +NaPTAN is the default stop source. Set `UK_BUS_STOPS_SOURCE=osm` to retain +Overpass stop searches instead. OSM route queries use `overpass.atownsend.org.uk`, +with `overpass.private.coffee` as a fallback. The production URL is `https://openstreetmap.tools/uk-bus-stops/`. The app's `/about` page documents shareable URL parameters, map behaviour, diff --git a/pyproject.toml b/pyproject.toml index d8d0deb..4cf5eef 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -17,6 +17,7 @@ dependencies = [ [project.optional-dependencies] web = [ "flask>=3.0", + "pyproj>=3.4", ] dev = [ "mypy", @@ -25,10 +26,12 @@ dev = [ "responses", "types-requests", "flask>=3.0", + "pyproj>=3.4", ] [project.scripts] osm-pt-geojson = "osm_geojson.pt.cli:cli" +uk-bus-stops-import-naptan = "uk_bus_stops.naptan:main" uk-bus-stops = "uk_bus_stops.app:main" [tool.setuptools.packages.find] diff --git a/src/uk_bus_stops/__init__.py b/src/uk_bus_stops/__init__.py index 3bfd7c7..f023ec0 100644 --- a/src/uk_bus_stops/__init__.py +++ b/src/uk_bus_stops/__init__.py @@ -1,4 +1,4 @@ -"""Find UK bus stop ATCO codes using OpenStreetMap data.""" +"""Find UK bus stop ATCO codes using NaPTAN and OpenStreetMap data.""" from uk_bus_stops.app import create_app diff --git a/src/uk_bus_stops/app.py b/src/uk_bus_stops/app.py index 870e2f3..bd256ce 100644 --- a/src/uk_bus_stops/app.py +++ b/src/uk_bus_stops/app.py @@ -2,7 +2,9 @@ from __future__ import annotations +import os import re +from pathlib import Path from typing import Any from flask import Flask, Response, jsonify, render_template, request @@ -18,6 +20,7 @@ from uk_bus_stops.core import ( routes_for_stop, stops_in_bounds, ) +from uk_bus_stops.naptan import NaptanError, NaptanStore, database_path ATCO_PATTERN = re.compile(r"^(?=[A-Za-z0-9]{9,16}$)(?=.*\d)[A-Za-z0-9]+$") COORDINATE_PATTERN = re.compile( @@ -54,9 +57,26 @@ def parse_viewbox() -> tuple[float, float, float, float] | None: def create_app(test_config: dict[str, Any] | None = None) -> Flask: """Create and configure the bus stop finder application.""" app = Flask(__name__) + app.config.from_mapping( + STOP_SOURCE=os.environ.get("UK_BUS_STOPS_SOURCE", "naptan"), + NAPTAN_DATABASE=str(database_path()), + ) if test_config: app.config.update(test_config) + if app.config["STOP_SOURCE"] not in ("naptan", "osm"): + raise ValueError("UK_BUS_STOPS_SOURCE must be naptan or osm") + store = NaptanStore(Path(app.config["NAPTAN_DATABASE"])) + use_naptan = app.config["STOP_SOURCE"] == "naptan" + find_nearby = store.nearby_stops if use_naptan else nearby_stops + find_bounds = store.stops_in_bounds if use_naptan else stops_in_bounds + find_code = store.find_atco_code if use_naptan else find_atco_code + + @app.errorhandler(NaptanError) + def handle_naptan_error(exc: NaptanError) -> tuple[Response, int]: + """Explain an unavailable local dataset using the standard error shape.""" + return _error("naptan_unavailable", str(exc), 503) + @app.errorhandler(UpstreamError) def handle_upstream_error(exc: UpstreamError) -> tuple[Response, int]: """Return the common JSON response for upstream OSM service failures.""" @@ -89,10 +109,10 @@ def create_app(test_config: dict[str, Any] | None = None) -> Flask: return jsonify({ "kind": "location", "location": coordinate_location, - "stops": nearby_stops(lat, lon), + "stops": find_nearby(lat, lon), }) if ATCO_PATTERN.fullmatch(query): - stops = find_atco_code(query) + stops = find_code(query) return jsonify({"kind": "atco", "label": query.upper(), "stops": stops}) locations = geocode(query, viewbox=viewbox) if not locations: @@ -113,7 +133,31 @@ def create_app(test_config: dict[str, Any] | None = None) -> Flask: return jsonify({ "kind": "location", "location": {"lat": lat, "lon": lon, "label": label}, - "stops": nearby_stops(lat, lon), + "stops": find_nearby(lat, lon), + }) + + @app.get("/api/stop/naptan/") + def naptan_detail(code: str) -> ResponseReturnValue: + """Resolve a shareable NaPTAN stop by ATCO code.""" + found = store.find_atco_code(code) + if not found: + return _error("stop_not_found", "That active NaPTAN stop was not found.", 404) + return jsonify({"stop": found[0]}) + + @app.get("/api/stop/naptan//routes") + def naptan_routes(code: str) -> ResponseReturnValue: + """Use exact ATCO matches to retrieve OSM routes for a NaPTAN stop.""" + if not ATCO_PATTERN.fullmatch(code) or not store.find_atco_code(code): + return _error("stop_not_found", "That active NaPTAN stop was not found.", 404) + matches = find_atco_code(code) + routes = {} + for stop in matches: + for route in routes_for_stop(stop["type"], stop["id"]): + routes[route["id"]] = route + return jsonify({ + "routes": sorted(routes.values(), key=lambda route: (route.get("ref") or "", route["id"])), + "source": "OpenStreetMap", + "matched_stops": len(matches), }) @app.get("/api/stop///routes") @@ -149,7 +193,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) - return jsonify({"kind": "map", "stops": stops_in_bounds(south, west, north, east)}) + return jsonify({"kind": "map", "stops": find_bounds(south, west, north, east)}) return app diff --git a/src/uk_bus_stops/naptan.py b/src/uk_bus_stops/naptan.py new file mode 100644 index 0000000..0664970 --- /dev/null +++ b/src/uk_bus_stops/naptan.py @@ -0,0 +1,173 @@ +"""Import NaPTAN CSV snapshots and search a local SQLite spatial index.""" + +from __future__ import annotations + +import csv +import json +import math +import os +from pathlib import Path +import sqlite3 +import tempfile +from contextlib import closing +from typing import Any + +import click +from pyproj import Transformer + +DEFAULT_DATABASE = Path(__file__).resolve().parents[2] / "data" / "naptan.sqlite3" +STOP_TYPES = { + "BCT": "Bus stop", "BCS": "Bus / coach station bay", "BCQ": "Bus / coach station (variable bay)", + "BCE": "Bus / coach station entrance", "BST": "Bus / coach station", + "RSE": "Railway station entrance", "RLY": "Railway station", + "RPL": "Railway platform", "TMU": "Tram / metro entrance", + "MET": "Tram / metro stop", "PLT": "Tram / metro platform", + "FER": "Ferry terminal", "FTD": "Ferry terminal entrance", "FBT": "Ferry berth", + "AIR": "Airport entrance", "GAT": "Airport interchange", "STR": "Shared taxi rank", + "TXR": "Taxi rank", "SDA": "Pick-up / set-down point", + "LSE": "Cable car entrance", "LCB": "Cable car station", "LPL": "Cable car platform", +} + + +class NaptanError(Exception): + """Report an unavailable or invalid local NaPTAN dataset.""" + + +def database_path() -> Path: + """Return the configured database path, shared by the app and importer.""" + return Path(os.environ.get("NAPTAN_DATABASE", str(DEFAULT_DATABASE))) + + +def import_csv(source: Path, destination: Path) -> dict[str, int]: + """Build a complete snapshot beside the old database and replace it atomically.""" + destination = destination.resolve() + destination.parent.mkdir(parents=True, exist_ok=True) + transformer = Transformer.from_crs(27700, 4326, always_xy=True) + counts = {"imported": 0, "inactive": 0, "invalid": 0, "converted": 0} + fd, filename = tempfile.mkstemp(prefix=".naptan-", suffix=".sqlite3", dir=destination.parent) + os.close(fd) + temporary = Path(filename) + try: + with source.open(encoding="utf-8-sig", newline="") as stream, closing(sqlite3.connect(temporary)) as db: + reader = csv.DictReader(stream) + required = {"ATCOCode", "CommonName", "StopType", "Status", "Latitude", "Longitude", "Easting", "Northing"} + if not required.issubset(reader.fieldnames or []): + raise NaptanError("The file is not a NaPTAN Stops.csv export (missing required columns).") + db.executescript(""" + CREATE TABLE stops ( + id INTEGER PRIMARY KEY, code TEXT UNIQUE COLLATE NOCASE, + lat REAL NOT NULL, lon REAL NOT NULL, fields TEXT NOT NULL + ); + CREATE VIRTUAL TABLE positions USING rtree(id, min_lon, max_lon, min_lat, max_lat); + """) + with db: + for row in reader: + if row["Status"].strip().lower() != "active": + counts["inactive"] += 1 + continue + converted = False + try: + if not row["ATCOCode"].strip(): + raise ValueError("Missing ATCO code") + if row["Latitude"] and row["Longitude"]: + lat, lon = float(row["Latitude"]), float(row["Longitude"]) + else: + if row.get("GridType", "") not in ("", "UKOS"): + raise ValueError("Unsupported grid") + east, north = float(row["Easting"]), float(row["Northing"]) + if not (0 < east <= 700000 and 0 < north <= 1300000): + raise ValueError("Invalid British National Grid position") + lon, lat = transformer.transform(east, north) + converted = True + if not (49 <= lat <= 61.5 and -9 <= lon <= 2.5): + raise ValueError("Position outside Great Britain") + except (ValueError, TypeError): + counts["invalid"] += 1 + continue + cursor = db.execute( + "INSERT INTO stops(code, lat, lon, fields) VALUES (?, ?, ?, ?)", + (row["ATCOCode"].strip(), lat, lon, json.dumps({key: value for key, value in row.items() if value}, ensure_ascii=False, separators=(",", ":"))), + ) + db.execute("INSERT INTO positions VALUES (?, ?, ?, ?, ?)", (cursor.lastrowid, lon, lon, lat, lat)) + counts["imported"] += 1 + counts["converted"] += int(converted) + if not counts["imported"]: + raise NaptanError("No active stops with valid coordinates were found; database unchanged.") + temporary.chmod(0o644) + os.replace(temporary, destination) + except (OSError, sqlite3.Error, csv.Error) as exc: + raise NaptanError(f"Could not import NaPTAN: {exc}") from exc + finally: + temporary.unlink(missing_ok=True) + return counts + + +class NaptanStore: + """Read one indexed snapshot with a separate read-only connection per query.""" + + def __init__(self, path: Path) -> None: + """Store the database location without opening it during app startup.""" + self.path = path + + def _query(self, sql: str, parameters: tuple[Any, ...]) -> list[dict[str, Any]]: + """Execute a query and convert records to the shared stop response format.""" + try: + with closing(sqlite3.connect(self.path.resolve().as_uri() + "?mode=ro", uri=True)) as db: + rows = db.execute(sql, parameters).fetchall() + except sqlite3.Error as exc: + raise NaptanError("NaPTAN data is unavailable. Import Stops.csv with uk-bus-stops-import-naptan.") from exc + stops = [] + for code, lat, lon, fields in rows: + tags = json.loads(fields) + stops.append({ + "type": "naptan", "source": "NaPTAN", "id": code, "atco_code": code, + "lat": lat, "lon": lon, "name": tags.get("CommonName") or "Unnamed stop", + "transport_type": STOP_TYPES.get(tags.get("StopType"), "Transport stop"), + "indicator": tags.get("Indicator"), "bearing": tags.get("Bearing"), + "tags": {key: value for key, value in tags.items() if value}, + }) + return stops + + def find_atco_code(self, code: str) -> list[dict[str, Any]]: + """Look up an exact ATCO code without changing its leading zeros.""" + return self._query("SELECT code, lat, lon, fields FROM stops WHERE code = ?", (code,)) + + def stops_in_bounds(self, south: float, west: float, north: float, east: float) -> list[dict[str, Any]]: + """Search the R-tree, then filter rounding at the exact viewport edges.""" + return self._query(""" + SELECT s.code, s.lat, s.lon, s.fields FROM positions p + JOIN stops s ON s.id = p.id + WHERE p.max_lat >= ? AND p.min_lat <= ? AND p.max_lon >= ? AND p.min_lon <= ? + AND s.lat BETWEEN ? AND ? AND s.lon BETWEEN ? AND ? + ORDER BY s.code + """, (south, north, west, east, south, north, west, east)) + + def nearby_stops(self, lat: float, lon: float, radius: int = 1000) -> list[dict[str, Any]]: + """Filter indexed candidates by great-circle distance and sort nearest first.""" + delta_lat = math.degrees(radius / 6371000) + delta_lon = delta_lat / max(math.cos(math.radians(lat)), 0.000001) + candidates = self.stops_in_bounds(lat - delta_lat, lon - delta_lon, lat + delta_lat, lon + delta_lon) + + def distance(stop: dict[str, Any]) -> float: + """Return spherical distance in metres from the requested coordinate.""" + a = math.sin(math.radians(stop["lat"] - lat) / 2) ** 2 + a += math.cos(math.radians(lat)) * math.cos(math.radians(stop["lat"])) * math.sin(math.radians(stop["lon"] - lon) / 2) ** 2 + return 2 * 6371000 * math.asin(min(1, math.sqrt(a))) + + return sorted((stop for stop in candidates if distance(stop) <= radius), key=distance) + + +@click.command() +@click.argument("source", type=click.Path(exists=True, dir_okay=False, path_type=Path)) +@click.option("--database", type=click.Path(dir_okay=False, path_type=Path), default=database_path, show_default="NAPTAN_DATABASE or data/naptan.sqlite3") +def main(source: Path, database: Path) -> None: + """Import a downloaded Stops.csv, replacing the previous NaPTAN snapshot.""" + try: + counts = import_csv(source, database) + except NaptanError as exc: + raise click.ClickException(str(exc)) from exc + click.echo(json.dumps(counts)) + + +if __name__ == "__main__": + main() diff --git a/src/uk_bus_stops/static/app.js b/src/uk_bus_stops/static/app.js index 7f49426..e6ef725 100644 --- a/src/uk_bus_stops/static/app.js +++ b/src/uk_bus_stops/static/app.js @@ -145,8 +145,10 @@ function restoreSearchFromUrl() { } } -/** Parse shareable OSM node, way, or relation parameters. */ +/** Parse shareable NaPTAN codes or OSM node, way and relation IDs. */ function parseStopReference(params) { + const atco = params.get('naptan'); + if (atco && /^[a-z0-9]+$/i.test(atco)) return {type: 'naptan', id: atco}; for (const type of ['node', 'way', 'relation']) { const id = params.get(type); if (id && /^\d+$/.test(id)) return {type, id: Number(id)}; @@ -154,9 +156,9 @@ function parseStopReference(params) { return null; } -/** Build a mounted-path-safe API URL for a particular OSM stop. */ +/** Build a mounted-path-safe API URL for a particular stop. */ function stopApiUrl(template, type, id) { - return template.replace('/TYPE/', `/${type}/`).replace('/0', `/${id}`); + return template.replace('/TYPE/', `/${type}/`).replace('/0', `/${encodeURIComponent(id)}`); } /** Fetch and display the stop named by a shareable URL. */ @@ -294,6 +296,7 @@ function renderStops(data, fitMap = true, sortOrigin = null, preserveSelection = empty.className = 'small text-muted'; empty.textContent = data.kind === 'map' ? 'No transport stops were found in this map area.' + : data.kind === 'atco' ? 'No stop was found with that ATCO code.' : 'No transport stops were found within 1 km.'; list.appendChild(empty); if (fitMap && data.location) { @@ -364,10 +367,15 @@ function showLocationImmediately(locationResult) { function markerColour(stop) { const type = stop.transport_type || ''; if (type.includes('Railway')) return '#6f42c1'; - if (type.includes('Tram')) return '#dc3545'; + if (type.includes('Tram') || type.includes('Light rail')) return '#dc3545'; if (type.includes('Underground')) return '#0d6efd'; if (type.includes('Ferry')) return '#0dcaf0'; + if (type.includes('Airport')) return '#334155'; + if (/taxi/i.test(type)) return '#ca8a04'; + if (type.includes('Cable car')) return '#d63384'; + if (type.includes('Pick-up')) return '#795548'; if (type.includes('Bus')) return '#198754'; + if (stop.type === 'naptan') return '#6c757d'; return stop.atco_code ? '#198754' : '#fd7e14'; } @@ -489,8 +497,12 @@ async function selectStop(stop, updateUrl = true) { 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); + const isNaptan = stop.type === 'naptan'; + byId('stop-source').textContent = isNaptan ? 'Source: NaPTAN · Department for Transport' : 'Source: OpenStreetMap'; + byId('tags-heading').textContent = isNaptan ? 'NaPTAN stop data' : 'Other OpenStreetMap tags'; + show('osm-link', !isNaptan); + byId('osm-link').href = isNaptan ? '#' : osmUrl(stop); + byId('edit-link').href = isNaptan ? '#' : osmUrl(stop, true); byId('atco-code').textContent = stop.atco_code || ''; show('atco-present', Boolean(stop.atco_code)); show('atco-missing', !stop.atco_code); @@ -513,11 +525,17 @@ async function selectStop(stop, updateUrl = true) { const response = await fetch(url); const data = await response.json(); if (!response.ok) throw new Error(data.message || 'Could not load routes.'); + if (selectedStop !== stop) return; renderRoutes(data.routes); + if (isNaptan && !data.routes.length) { + byId('route-list').textContent = data.matched_stops + ? 'No bus routes are recorded for the matching OpenStreetMap stop. Route coverage may be incomplete.' + : 'No OpenStreetMap stop matches this ATCO code, so route information is unavailable.'; + } } catch (error) { - byId('route-list').textContent = error.message; + if (selectedStop === stop) byId('route-list').textContent = error.message; } finally { - show('route-loading', false); + if (selectedStop === stop) show('route-loading', false); } } diff --git a/src/uk_bus_stops/templates/about.html b/src/uk_bus_stops/templates/about.html index 095b952..23f7986 100644 --- a/src/uk_bus_stops/templates/about.html +++ b/src/uk_bus_stops/templates/about.html @@ -22,18 +22,17 @@

About the UK Bus Stop Finder

This tool makes it easier to find the ATCO code recorded for a UK public - transport stop in OpenStreetMap. ATCO codes are used by journey-planning + transport stop in NaPTAN, the Department for Transport stop dataset. ATCO codes are used by journey-planning services and integrations such as Home Assistant bus departure displays.

Search by postcode, street, place, ATCO code, or coordinates. Select a - result to see its code, direction, coordinates, OSM tags, and associated - bus route relations. Stops without an ATCO code are still shown and link - to the OpenStreetMap editor. + result to see its code, direction, coordinates and source fields. Bus routes + come from OpenStreetMap, matched by ATCO code, and may be incomplete.

Shareable URL parameters

-

Use one search form at a time; selecting a stop replaces search parameters with its OSM object ID.

+

Use one search form at a time; selecting a stop replaces search parameters with its NaPTAN ATCO code (or OSM object ID).

@@ -48,6 +47,11 @@ + + + + + @@ -78,24 +82,28 @@
  • After moving the map, results are sorted from the new map centre.
  • Text searches use the visible map as a Nominatim ranking preference, but matches outside the view remain eligible.
  • Stops in the visible area load automatically at zoom level 15 or closer.
    • -
  • Wider views do not query Overpass, to avoid requesting an excessively large area.
    • +
  • Wider views do not load stops, to keep the map readable.
  • Selecting a stop highlights it and refreshes surrounding markers without closing its details.

    • Data and availability

    - Stop and route data comes from - OpenStreetMap contributors. - Place searches use Nominatim. Transport - queries use the Britain-and-Ireland Overpass instance operated at - overpass.atownsend.org.uk, - with Private.coffee as a fallback. + Stops come from a locally imported NaPTAN download + published by the Department for Transport. Only active stops with usable coordinates + are shown, covering England, Scotland and Wales, not Northern Ireland. + The snapshot is refreshed by the site operator; it is not a live departure feed. + Contains public sector information licensed under the + Open Government Licence v3.0.

    - Public Overpass services can occasionally be busy. OSM data may be incomplete - or out of date, and this finder is not an official source of NaPTAN data. - If a code or stop is missing, use the edit link on its detail panel and follow - OpenStreetMap's editing and source requirements. + Maps and bus routes use OpenStreetMap contributors' data. + Place searches use Nominatim. + Selecting a NaPTAN stop looks up OSM stops with the same ATCO code, then their bus route relations. + A missing match or empty route list does not mean no buses serve the stop. + Route queries use overpass.atownsend.org.uk, + with Private.coffee as a fallback. + These public services may occasionally be busy. Operators can also configure OSM as the stop source; + existing OSM stop links continue to work.

    Location and privacy

    @@ -122,6 +130,8 @@ + + diff --git a/src/uk_bus_stops/templates/index.html b/src/uk_bus_stops/templates/index.html index 5dc0e96..9c77615 100644 --- a/src/uk_bus_stops/templates/index.html +++ b/src/uk_bus_stops/templates/index.html @@ -3,7 +3,7 @@ - + UK Bus Stop ATCO Code Finder @@ -61,6 +61,7 @@

    +

    @@ -85,13 +86,13 @@

    View stop on OpenStreetMap ↗

    -

    Bus routes serving this stop

    +

    Bus routes in OpenStreetMap

    Loading routes…
    - Other OpenStreetMap tags + Other OpenStreetMap tags
    diff --git a/tests/test_naptan.py b/tests/test_naptan.py new file mode 100644 index 0000000..07d34cf --- /dev/null +++ b/tests/test_naptan.py @@ -0,0 +1,147 @@ +"""Offline NaPTAN import, spatial lookup and API regression tests.""" + +from __future__ import annotations + +import csv +from pathlib import Path + +import pytest +import responses + +from uk_bus_stops import core +from uk_bus_stops.app import create_app +from uk_bus_stops.naptan import NaptanError, NaptanStore, import_csv + + +@pytest.fixture() +def csv_file(tmp_path: Path) -> Path: + """Write representative stop records, including grid-only and invalid data.""" + path = tmp_path / "Stops.csv" + fields = ["ATCOCode", "CommonName", "StopType", "Status", "Latitude", "Longitude", "Easting", "Northing", "GridType", "Bearing", "Indicator"] + records = [ + ["0100BRP90317", "Temple Meads Stn", "BCT", "active", "51.44827", "-2.58302", "359581", "172304", "UKOS", "SE", "T1"], + ["0100GRID001", "Gare café", "RSE", "active", "", "", "359581", "172304", "", "", ""], + ["0100INACTIVE", "Closed", "BCT", "inactive", "51.448", "-2.583", "", "", "", "", ""], + ["0100INVALID", "Invalid", "BCT", "active", "nan", "0", "", "", "", "", ""], + ["0100CORNER", "Outside radius", "FER", "active", "51.456", "-2.571", "", "", "", "", ""], + ["0100EAST001", "East", "BCT", "active", "51.44827", "-2.57302", "", "", "", "", ""], + ["0100NORTH01", "North", "BCT", "active", "51.45627", "-2.58302", "", "", "", "", ""], + ] + with path.open("w", encoding="utf-8-sig", newline="") as stream: + writer = csv.writer(stream) + writer.writerow(fields) + writer.writerows(records) + return path + + +@pytest.fixture() +def database(csv_file: Path, tmp_path: Path) -> Path: + """Import an isolated database shared by query and API tests.""" + path = tmp_path / "naptan.sqlite3" + assert import_csv(csv_file, path) == {"imported": 5, "inactive": 1, "invalid": 1, "converted": 1} + return path + + +def test_import_and_exact_lookup(database: Path) -> None: + """Preserve codes, accents, transport metadata and converted grid positions.""" + store = NaptanStore(database) + stop = store.find_atco_code("0100brp90317")[0] + assert stop["id"] == "0100BRP90317" + assert stop["source"] == "NaPTAN" + assert stop["bearing"] == "SE" + assert stop["indicator"] == "T1" + grid = store.find_atco_code("0100GRID001")[0] + assert grid["name"] == "Gare café" + assert grid["transport_type"] == "Railway station entrance" + assert grid["lat"] == pytest.approx(stop["lat"], abs=0.0001) + assert grid["lon"] == pytest.approx(stop["lon"], abs=0.0001) + assert store.find_atco_code("0100INACTIVE") == [] + assert store.find_atco_code("' OR 1=1 --") == [] + + +def test_spatial_search(database: Path) -> None: + """Radius excludes box corners and orders longitude distances correctly.""" + store = NaptanStore(database) + nearby = store.nearby_stops(51.44827, -2.58302) + assert [stop["id"] for stop in nearby][-2:] == ["0100EAST001", "0100NORTH01"] + assert "0100CORNER" not in {stop["id"] for stop in nearby} + assert len(store.stops_in_bounds(51.448, -2.584, 51.449, -2.582)) == 2 + assert store.stops_in_bounds(51.44828, -2.583021, 51.44829, -2.583019) == [] + assert store.nearby_stops(0, 0) == [] + + +def test_atomic_refresh_and_failed_import(database: Path, csv_file: Path) -> None: + """Invalid refreshes preserve the old database; valid refreshes replace it.""" + original = csv_file.read_text(encoding="utf-8-sig") + csv_file.write_text("invalid\n") + with pytest.raises(NaptanError): + import_csv(csv_file, database) + store = NaptanStore(database) + assert store.find_atco_code("0100BRP90317")[0]["name"] == "Temple Meads Stn" + csv_file.write_text(original.replace("Temple Meads Stn", "Updated name")) + import_csv(csv_file, database) + assert store.find_atco_code("0100BRP90317")[0]["name"] == "Updated name" + assert not list(database.parent.glob(".naptan-*")) + + +@responses.activate +def test_naptan_api_needs_no_network(database: Path) -> None: + """Coordinate, viewport, code and shared stop requests all use local data.""" + client = create_app({"TESTING": True, "STOP_SOURCE": "naptan", "NAPTAN_DATABASE": str(database)}).test_client() + for url in ( + "/api/search?q=0100brp90317", + "/api/search?q=51.44827,-2.58302", + "/api/stops?lat=51.44827&lon=-2.58302", + "/api/stops/in-bounds?south=51.448&west=-2.584&north=51.449&east=-2.582", + ): + response = client.get(url) + assert response.status_code == 200 + assert response.json["stops"][0]["source"] == "NaPTAN" + response = client.get("/api/stop/naptan/0100brp90317") + assert response.status_code == 200 + assert response.json["stop"]["id"] == "0100BRP90317" + assert client.get("/api/stop/naptan/missing").status_code == 404 + assert len(responses.calls) == 0 + + +def test_missing_database(tmp_path: Path) -> None: + """An unconfigured installation explains how to import instead of creating a DB.""" + database = tmp_path / "missing.sqlite3" + client = create_app({"TESTING": True, "NAPTAN_DATABASE": str(database), "STOP_SOURCE": "naptan"}).test_client() + response = client.get("/api/stops?lat=51.45&lon=-2.58") + assert response.status_code == 503 + assert response.json["error"] == "naptan_unavailable" + assert not database.exists() + + +@responses.activate +def test_naptan_routes_match_atco_and_deduplicate(database: Path) -> None: + """Multiple matching OSM objects contribute one combined route list.""" + responses.get(core.OVERPASS_URL, json={"elements": [ + {"type": "node", "id": 1, "lat": 51.448, "lon": -2.583}, + {"type": "way", "id": 2, "center": {"lat": 51.448, "lon": -2.583}}, + ]}) + for _ in range(2): + responses.get(core.OVERPASS_URL, json={"elements": [ + {"id": 12, "tags": {"ref": "A1", "route": "bus"}}, + ]}) + client = create_app({"TESTING": True, "NAPTAN_DATABASE": str(database)}).test_client() + response = client.get("/api/stop/naptan/0100BRP90317/routes") + assert response.status_code == 200 + assert response.json["matched_stops"] == 2 + assert [route["ref"] for route in response.json["routes"]] == ["A1"] + assert "0100BRP90317" in responses.calls[0].request.url + + +@responses.activate +def test_naptan_routes_without_osm_match(database: Path) -> None: + """A missing OSM match has a distinct result from a service failure.""" + responses.get(core.OVERPASS_URL, json={"elements": []}) + client = create_app({"TESTING": True, "NAPTAN_DATABASE": str(database)}).test_client() + assert client.get("/api/stop/naptan/0100BRP90317/routes").json == { + "routes": [], "source": "OpenStreetMap", "matched_stops": 0, + } + responses.replace(responses.GET, core.OVERPASS_URL, status=503) + responses.get(core.OVERPASS_FALLBACK_URLS[0], status=503) + assert client.get("/api/stop/naptan/0100BRP90317/routes").status_code == 502 + assert client.get("/api/stop/naptan/0100BRP90317").status_code == 200 diff --git a/tests/test_uk_bus_stops.py b/tests/test_uk_bus_stops.py index bd61f99..67e3f66 100644 --- a/tests/test_uk_bus_stops.py +++ b/tests/test_uk_bus_stops.py @@ -14,7 +14,7 @@ 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() + return create_app({"TESTING": True, "STOP_SOURCE": "osm"}).test_client() @responses.activate diff --git a/tests/test_uk_bus_stops_playwright.py b/tests/test_uk_bus_stops_playwright.py index a6e7091..5650625 100644 --- a/tests/test_uk_bus_stops_playwright.py +++ b/tests/test_uk_bus_stops_playwright.py @@ -320,3 +320,33 @@ def test_mobile_postcode_is_fitted_above_results_sheet( }""") assert 0 < position["markerY"] < position["visibleBottom"] page.close() + + +def test_naptan_shared_stop_and_osm_routes(chromium_browser: Any, flask_url: str) -> None: + """NaPTAN links survive reload and show source fields plus matched OSM routes.""" + page = chromium_browser.new_page(viewport={"width": 1280, "height": 800}) + errors: list[str] = [] + page.on("pageerror", lambda error: errors.append(str(error))) + stop = { + "type": "naptan", "source": "NaPTAN", "id": "0100BRP90317", + "lat": 51.44827, "lon": -2.58302, "name": "Temple Meads Stn", + "atco_code": "0100BRP90317", "indicator": "T1", "bearing": "SE", + "transport_type": "Bus stop", "tags": {"CommonName": "Temple Meads Stn", "Status": "active"}, + } + page.route("**/api/stop/naptan/0100BRP90317", lambda route: route.fulfill(json={"stop": stop})) + page.route("**/api/stop/naptan/0100BRP90317/routes", lambda route: route.fulfill(json={ + "matched_stops": 1, "routes": [{"id": 12, "ref": "A1", "name": "Airport bus"}], + })) + page.route("**/api/stops/in-bounds?*", lambda route: route.fulfill(json={"kind": "map", "stops": [stop]})) + page.goto(f"{flask_url}/?naptan=0100BRP90317") + expect = playwright_api.expect + for _ in range(2): + expect(page.locator("#stop-name")).to_have_text("Temple Meads Stn") + expect(page.locator("#stop-source")).to_contain_text("NaPTAN") + expect(page.locator("#tags-heading")).to_have_text("NaPTAN stop data") + expect(page.locator("#osm-link")).not_to_be_visible() + expect(page.locator("#route-list")).to_contain_text("A1") + assert "naptan=0100BRP90317" in page.url + page.reload() + assert errors == [] + page.close()