File size: 3,777 Bytes
e42dd8b f867b96 c62beab e42dd8b c62beab e42dd8b c62beab e42dd8b f867b96 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 | """
Snap an ordered list of stop coordinates to the actual road network so a bus
leg can be drawn as a road-following polyline instead of straight hops.
Uses the public OSRM demo server (no API key). Routing waypoint-to-waypoint
through every stop on the leg keeps the geometry close to the real bus path,
since consecutive stops are usually close enough that the road between them is
unambiguous. Results are cached on disk; any failure falls back to None so the
caller can draw straight lines.
"""
import hashlib
import json
import os
import subprocess
import threading
DATA_DIR = os.path.join(os.path.dirname(os.path.abspath(__file__)), "data")
CACHE = os.path.join(DATA_DIR, "roads_cache.json")
OSRM = "https://router.project-osrm.org/route/v1/driving/"
# FOSSGIS public OSRM, foot profile — for the "walk to your stop" path. Reached
# server-side (the container can resolve it even where a phone's network can't).
FOOT = "https://routing.openstreetmap.de/routed-foot/route/v1/foot/"
UA = (
"MauritiusBusPlanner/1.0 (personal route-planning demo; "
"contact: syadone@gmail.com)"
)
# Stock macOS Python links against an old LibreSSL that the OSRM demo server
# rejects at the TLS handshake, so we fetch via the system `curl`, whose TLS
# stack negotiates fine. Any failure falls back to None (straight lines).
def _fetch(url):
try:
out = subprocess.run(
["curl", "-s", "-m", "25", "-A", UA, url],
capture_output=True,
timeout=30,
)
if out.returncode == 0 and out.stdout:
return json.loads(out.stdout.decode("utf-8", "ignore"))
except Exception:
pass
return None
_lock = threading.Lock()
_cache = None
def _load():
global _cache
if _cache is None:
if os.path.exists(CACHE):
with open(CACHE, encoding="utf-8") as fh:
_cache = json.load(fh)
else:
_cache = {}
return _cache
def _save():
os.makedirs(DATA_DIR, exist_ok=True)
tmp = CACHE + ".tmp"
with open(tmp, "w", encoding="utf-8") as fh:
json.dump(_cache, fh)
os.replace(tmp, CACHE)
def _key(coords):
raw = ";".join(f"{round(la, 5)},{round(lo, 5)}" for la, lo in coords)
return hashlib.sha1(raw.encode()).hexdigest()
def snap_route(coords):
"""coords: ordered [[lat, lon], ...]. Returns road geometry [[lat, lon], ...]
following the streets between the stops, or None if it can't be computed."""
coords = [c for c in coords if c]
if len(coords) < 2:
return None
cache = _load()
key = _key(coords)
if key in cache:
return cache[key] or None
# OSRM wants lon,lat pairs.
path = ";".join(f"{lo},{la}" for la, lo in coords)
url = OSRM + path + "?overview=full&geometries=geojson"
geom = None
data = _fetch(url)
if data and data.get("code") == "Ok" and data.get("routes"):
# GeoJSON coordinates are [lon, lat]; flip for Leaflet.
geom = [[lat, lon] for lon, lat in data["routes"][0]["geometry"]["coordinates"]]
with _lock:
cache[key] = geom
_save()
return geom
def walk_route(a, b):
"""a, b: [lat, lon]. Returns {geometry: [[lat,lon],...], distance, duration}
following pedestrian ways, or None. Not cached — GPS origins are always new."""
if not a or not b:
return None
url = f"{FOOT}{a[1]},{a[0]};{b[1]},{b[0]}?overview=full&geometries=geojson"
data = _fetch(url)
if data and data.get("code") == "Ok" and data.get("routes"):
r = data["routes"][0]
return {
"geometry": [[lat, lon] for lon, lat in r["geometry"]["coordinates"]],
"distance": r.get("distance"),
"duration": r.get("duration"),
}
return None
|