haptic_skin.routing

  1"""Routing — turn the trip into a list of waypoints the navigator can follow.
  2
  3A :class:`Waypoint` is a maneuver point with the bearing to follow *after* it.
  4Sources:
  5  * Google Directions API (live, needs an API key) — :func:`fetch_route_google`
  6  * a saved Directions JSON / a simple waypoint JSON — :func:`load_route`
  7
  8Both produce a :class:`Route`. Keeping a file loader means the whole nav stack
  9is testable offline (and gives us the indoor "replay" fallback for the demo).
 10"""
 11
 12from __future__ import annotations
 13
 14import json
 15import re
 16from dataclasses import dataclass
 17from typing import List
 18
 19from .geo import LatLon, bearing_deg
 20
 21_TAG_RE = re.compile(r"<[^>]+>")
 22
 23
 24@dataclass
 25class Waypoint:
 26    lat: float
 27    lon: float
 28    next_bearing: float          # compass bearing to follow after this point
 29    instruction: str = ""
 30
 31    @property
 32    def point(self) -> LatLon:
 33        return (self.lat, self.lon)
 34
 35
 36@dataclass
 37class Route:
 38    waypoints: List[Waypoint]
 39
 40    def __len__(self) -> int:
 41        return len(self.waypoints)
 42
 43    def to_json(self) -> str:
 44        return json.dumps(
 45            {"waypoints": [vars(w) for w in self.waypoints]}, indent=2
 46        )
 47
 48
 49def _strip_html(text: str) -> str:
 50    return _TAG_RE.sub("", text or "").strip()
 51
 52
 53def route_from_directions(payload: dict) -> Route:
 54    """Build a Route from a Google Directions API JSON response."""
 55    routes = payload.get("routes") or []
 56    if not routes:
 57        raise ValueError("no route in Directions payload")
 58    steps = routes[0]["legs"][0]["steps"]
 59    wps: List[Waypoint] = []
 60    for step in steps:
 61        start = (step["start_location"]["lat"], step["start_location"]["lng"])
 62        end = (step["end_location"]["lat"], step["end_location"]["lng"])
 63        wps.append(
 64            Waypoint(start[0], start[1], bearing_deg(start, end),
 65                     _strip_html(step.get("html_instructions", "")))
 66        )
 67    # final destination as a terminal waypoint (bearing irrelevant)
 68    last_end = steps[-1]["end_location"]
 69    wps.append(Waypoint(last_end["lat"], last_end["lng"], 0.0, "Arrivée"))
 70    return Route(wps)
 71
 72
 73def route_from_waypoints(payload: dict) -> Route:
 74    """Build a Route from a simple ``{"waypoints": [{lat,lon,next_bearing,instruction}]}``."""
 75    return Route([Waypoint(**w) for w in payload["waypoints"]])
 76
 77
 78def load_route(path: str) -> Route:
 79    """Load a route from JSON — accepts both formats above."""
 80    with open(path, "r", encoding="utf-8") as fh:
 81        data = json.load(fh)
 82    if "routes" in data:
 83        return route_from_directions(data)
 84    return route_from_waypoints(data)
 85
 86
 87def fetch_route_google(origin: LatLon, dest: LatLon, api_key: str,
 88                       mode: str = "walking") -> Route:
 89    """Call the Google Directions API and return a Route (uses stdlib only)."""
 90    import urllib.parse
 91    import urllib.request
 92
 93    params = urllib.parse.urlencode({
 94        "origin": f"{origin[0]},{origin[1]}",
 95        "destination": f"{dest[0]},{dest[1]}",
 96        "mode": mode,
 97        "key": api_key,
 98    })
 99    url = f"https://maps.googleapis.com/maps/api/directions/json?{params}"
100    with urllib.request.urlopen(url, timeout=10) as resp:  # noqa: S310
101        payload = json.loads(resp.read().decode("utf-8"))
102    if payload.get("status") != "OK":
103        raise RuntimeError(f"Directions error: {payload.get('status')}")
104    return route_from_directions(payload)
105
106
107def route_to_trace(route: Route, step_m: float = 3.0) -> List[dict]:
108    """Densify a route into a list of fixes ``{lat, lon, heading}``.
109
110    Used offline to feed the navigator without a browser/GPS (tests + replay).
111    """
112    from .geo import haversine_m
113
114    fixes: List[dict] = []
115    pts = route.waypoints
116    if not pts:
117        return fixes
118    for i in range(len(pts) - 1):
119        a, b = pts[i].point, pts[i + 1].point
120        seg = haversine_m(a, b)
121        heading = bearing_deg(a, b)
122        n = max(1, int(seg / step_m))
123        for k in range(n):
124            t = k / n
125            fixes.append({
126                "lat": a[0] + (b[0] - a[0]) * t,
127                "lon": a[1] + (b[1] - a[1]) * t,
128                "heading": heading,
129            })
130    last = pts[-1].point
131    fixes.append({"lat": last[0], "lon": last[1], "heading": 0.0})
132    return fixes