roads.py

"""
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