MVUM Layer 3c: surface-change transition candidates (#27)

Extract surface-category boundaries along the winning single-mode polyline as
additional multi-modal-Auto transition candidates, so Auto can suggest "pull off
where the pavement turns to dirt and switch vehicles" trips even with no MVUM
trailhead nearby. Candidates share the trailhead record shape, so _try_hybrid_auto
consumes them with no restructuring.

Backend-only:
- mvum_surface_change.py: get_surface_change_candidates(coords, valhalla_url) walks
  the polyline through Valhalla trace_attributes (action=include, costing=auto,
  edge.surface/road_class/use/begin_shape_index/end_shape_index). classify_surface
  buckets each edge into PAVED/UNPAVED/TRACK/TRAIL; adjacent edges are grouped into
  runs, runs shorter than MIN_STRETCH_M (100 m, measured by haversine along the input
  coords) are collapsed to suppress noise, and each surviving category boundary emits
  {lat, lon, name: "Surface change: <from>-><to>", road_class}. Capped at 10. Adds an
  encode_polyline6 helper (the inverse of the router _decode_polyline method).
- router.py: _try_hybrid_auto concatenates trailheads + surface-change candidates,
  then re-sorts by distance to the route and applies the existing
  HYBRID_MAX_TRAILHEADS cap. Probing logic unchanged.

Verified trace_attributes on the live Valhalla before coding (returns the requested
edge fields). Two empirically-driven deviations from the spec, flagged:
1. This Valhalla normalizes OSM surface tags into its own enum (paved_smooth/paved/
   paved_rough/compacted/dirt/gravel/path/impassable); classify_surface keys on that
   enum AND the raw OSM names for robustness.
2. Urban alleys come back as road_class=service_other with surface=paved_smooth, so
   the service_other->TRACK rule is gated on a non-paved surface to avoid classifying
   paved alleys as tracks.

Tests: test_mvum_surface_change.py (6) -- classify spot-check, paved->unpaved boundary,
sub-100 m noise suppression, uniform-surface empty, encoder round-trip vs the router
decoder, and hybrid integration (both trailhead + surface candidates probed). Full
offroute suite: 77 passed.

Co-authored-by: Matt <mj@k7zvx.com>
Co-authored-by: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

backend/services/navi_offroute/mvum_surface_change.py

@@ -0,0 +1,215 @@
+"""
+MVUM Layer 3c: surface-change transition candidates for multi-modal Auto.
+
+Walks the winning single-mode polyline through Valhalla's ``trace_attributes`` and
+emits a transition candidate wherever the road surface category changes (e.g.
+pavement -> dirt, dirt -> track, track -> trail). These let Auto suggest "pull off
+where the pavement turns to dirt and switch vehicles" hybrid trips even when no MVUM
+trailhead sits nearby. Candidates share the trailhead record shape
+(``{lat, lon, name, road_class}``) so ``_try_hybrid_auto`` consumes them unchanged.
+
+Surface categories use Valhalla's normalized surface enum (paved_smooth / paved /
+paved_rough / compacted / dirt / gravel / path / impassable) as observed on the live
+instance, plus the raw OSM surface names for robustness.
+"""
+import logging
+import math
+
+import requests
+
+logger = logging.getLogger("navi_offroute.mvum_surface_change")
+
+# Surface-change tuning.
+MIN_STRETCH_M = 100.0     # a new category must persist this far, else it is noise
+MAX_BOUNDARIES = 10       # cap candidates emitted per route
+TRACE_TIMEOUT_S = 20
+
+# Categories.
+PAVED = "PAVED"
+UNPAVED = "UNPAVED"
+TRACK = "TRACK"
+TRAIL = "TRAIL"
+
+# Valhalla normalized surface enum + raw OSM surface names (defensive).
+PAVED_SURFACES = frozenset({
+    "paved_smooth", "paved", "paved_rough",
+    "asphalt", "concrete", "concrete:lanes", "concrete:plates",
+    "paving_stones", "sett", "cobblestone", "metal", "wood",
+})
+UNPAVED_SURFACES = frozenset({
+    "compacted", "dirt", "gravel",
+    "fine_gravel", "ground", "sand", "earth", "mud", "grass", "unpaved",
+})
+DIRT_TRACK_SURFACES = frozenset({"dirt", "compacted"})
+TRAIL_USES = frozenset({
+    "path", "footway", "cycleway", "bridleway", "steps", "pedestrian",
+    "mountain_bike", "sidewalk",
+})
+
+
+def classify_surface(edge):
+    """Classify one trace_attributes edge into PAVED / UNPAVED / TRACK / TRAIL, or
+    None when it cannot be categorised (unknown surfaces break runs, not boundaries).
+
+    Precedence: trail-like ``use`` first, then track (``use==track``; or an
+    *unpaved* service road; or a dirt/compacted ``use==road``), then plain
+    unpaved/paved surface.
+    """
+    use = (edge.get("use") or "").lower()
+    road_class = (edge.get("road_class") or "").lower()
+    surface = (edge.get("surface") or "").lower()
+
+    if use in TRAIL_USES:
+        return TRAIL
+    # A paved alley/driveway is road_class==service_other on this Valhalla, so only
+    # treat service_other as a track when its surface is NOT paved.
+    if use == "track" or (road_class == "service_other" and surface not in PAVED_SURFACES):
+        return TRACK
+    if surface in DIRT_TRACK_SURFACES and use == "road":
+        return TRACK
+    if surface in UNPAVED_SURFACES:
+        return UNPAVED
+    if surface in PAVED_SURFACES:
+        return PAVED
+    return None
+
+
+def _haversine_m(a, b):
+    (lat1, lon1), (lat2, lon2) = a, b
+    r = 6371000.0
+    p1, p2 = math.radians(lat1), math.radians(lat2)
+    dp = math.radians(lat2 - lat1)
+    dl = math.radians(lon2 - lon1)
+    h = math.sin(dp / 2) ** 2 + math.cos(p1) * math.cos(p2) * math.sin(dl / 2) ** 2
+    return 2 * r * math.asin(min(1.0, math.sqrt(h)))
+
+
+def _cumulative_m(coords):
+    """Cumulative geodesic distance (m) along the (lat, lon) polyline; len == len(coords)."""
+    cum = [0.0]
+    for i in range(1, len(coords)):
+        cum.append(cum[-1] + _haversine_m(coords[i - 1], coords[i]))
+    return cum
+
+
+def encode_polyline6(coords, precision=6):
+    """Encode (lat, lon) coords as a Valhalla polyline6 string (the inverse of the
+    router's _decode_polyline)."""
+    factor = 10 ** precision
+    out = []
+    prev_lat = prev_lon = 0
+    for lat, lon in coords:
+        lat_i = int(round(lat * factor))
+        lon_i = int(round(lon * factor))
+        for delta in (lat_i - prev_lat, lon_i - prev_lon):
+            v = ~(delta << 1) if delta < 0 else (delta << 1)
+            while v >= 0x20:
+                out.append(chr((0x20 | (v & 0x1f)) + 63))
+                v >>= 5
+            out.append(chr(v + 63))
+        prev_lat, prev_lon = lat_i, lon_i
+    return "".join(out)
+
+
+def _collapse_short_runs(runs, cum, min_m):
+    """Drop category runs shorter than ``min_m`` (merging each into a neighbour) and
+    re-merge adjacent same-category runs, so a brief reversion does not raise a
+    spurious boundary. Each run is ``{cat, start_v, end_v, last_edge}``."""
+    runs = [dict(r) for r in runs]
+    changed = True
+    while changed and len(runs) > 1:
+        changed = False
+        for i, r in enumerate(runs):
+            end_v = min(r["end_v"], len(cum) - 1)
+            start_v = min(r["start_v"], len(cum) - 1)
+            if cum[end_v] - cum[start_v] >= min_m:
+                continue
+            if i > 0:
+                runs[i - 1]["end_v"] = r["end_v"]
+                runs[i - 1]["last_edge"] = r["last_edge"]
+                del runs[i]
+            else:
+                runs[i + 1]["start_v"] = r["start_v"]
+                del runs[i]
+            changed = True
+            break
+        j = 0
+        while j < len(runs) - 1:
+            if runs[j]["cat"] == runs[j + 1]["cat"]:
+                runs[j]["end_v"] = runs[j + 1]["end_v"]
+                runs[j]["last_edge"] = runs[j + 1]["last_edge"]
+                del runs[j + 1]
+            else:
+                j += 1
+    return runs
+
+
+def _edges_to_candidates(edges, coords, max_boundaries=MAX_BOUNDARIES,
+                         min_stretch_m=MIN_STRETCH_M):
+    """Pure boundary extraction: trace_attributes ``edges`` (in order) + the input
+    (lat, lon) ``coords`` -> surface-change candidate records."""
+    if not edges or len(coords) < 2:
+        return []
+    cum = _cumulative_m(coords)
+
+    # Group consecutive edges of the same category into runs.
+    runs = []
+    for edge in edges:
+        cat = classify_surface(edge)
+        bi = edge.get("begin_shape_index")
+        ei = edge.get("end_shape_index")
+        if bi is None or ei is None:
+            continue
+        if runs and runs[-1]["cat"] == cat:
+            runs[-1]["end_v"] = ei
+            runs[-1]["last_edge"] = edge
+        else:
+            runs.append({"cat": cat, "start_v": bi, "end_v": ei, "last_edge": edge})
+
+    runs = _collapse_short_runs(runs, cum, min_stretch_m)
+
+    candidates = []
+    for prev, cur in zip(runs, runs[1:]):
+        from_cat, to_cat = prev["cat"], cur["cat"]
+        if from_cat is None or to_cat is None or from_cat == to_cat:
+            continue
+        v = cur["start_v"]
+        if v < 0 or v >= len(coords):
+            continue
+        lat, lon = coords[v]
+        candidates.append({
+            "lat": lat,
+            "lon": lon,
+            "name": f"Surface change: {from_cat.lower()}→{to_cat.lower()}",
+            "road_class": prev["last_edge"].get("road_class"),
+        })
+        if len(candidates) >= max_boundaries:
+            break
+    return candidates
+
+
+def get_surface_change_candidates(coords, valhalla_url):
+    """Transition candidates at surface-category boundaries along the (lat, lon)
+    ``coords`` polyline. Returns [] on any trace failure (best-effort augmentation)."""
+    if not coords or len(coords) < 2:
+        return []
+    payload = {
+        "encoded_polyline": encode_polyline6(coords),
+        "costing": "auto",
+        "filters": {
+            "attributes": [
+                "edge.surface", "edge.road_class", "edge.use",
+                "edge.begin_shape_index", "edge.end_shape_index",
+            ],
+            "action": "include",
+        },
+    }
+    try:
+        resp = requests.post(f"{valhalla_url}/trace_attributes",
+                             json=payload, timeout=TRACE_TIMEOUT_S)
+        resp.raise_for_status()
+        edges = resp.json().get("edges", [])
+    except Exception as e:
+        logger.warning("trace_attributes failed; no surface-change candidates: %s", e)
+        return []
+    return _edges_to_candidates(edges, coords)

backend/services/navi_offroute/router.py

@@ -34,6 +34,7 @@ import psycopg2
 import psycopg2.extras
 from shapely.geometry import LineString, Point
 from .astar import astar_multigoal, inflate_cost_multiplier
+from .mvum_surface_change import get_surface_change_candidates
 
 from shared.dem import DEMReader, dem_path
 from .cost import compute_cost_grid, compute_cost_multiplier_grid, MODE_PROFILES
@@ -932,9 +933,13 @@ class OffrouteRouter:
             return None
         candidates = idx.query_trailheads_near_line(
             coords, buffer_m=HYBRID_TRAILHEAD_BUFFER_M)
+        # Layer 3c: also treat surface-category boundaries along the winning polyline
+        # (e.g. pavement -> dirt) as transition candidates. Same record shape, so they
+        # mix freely with trailheads below.
+        candidates = candidates + get_surface_change_candidates(coords, VALHALLA_URL)
         if not candidates:
             return None
-        # Closest-to-route first, then cap.
+        # Closest-to-route first, then cap the combined list.
         line = LineString([(lon, lat) for (lat, lon) in coords])
         candidates.sort(key=lambda th: line.distance(Point(th["lon"], th["lat"])))
         candidates = candidates[:HYBRID_MAX_TRAILHEADS]

backend/services/navi_offroute/tests/test_mvum_surface_change.py

@@ -0,0 +1,165 @@
+"""MVUM Layer 3c tests: surface-change transition candidate extraction.
+
+The boundary tests feed synthetic trace_attributes ``edges`` straight into the pure
+``_edges_to_candidates`` (no Valhalla). The integration test stubs both candidate
+sources and self.route on a bare router to confirm surface-change candidates flow
+through _try_hybrid_auto alongside trailheads.
+"""
+import pytest
+
+from services.navi_offroute.mvum_surface_change import (
+    classify_surface, _edges_to_candidates, encode_polyline6,
+    PAVED, UNPAVED, TRACK, TRAIL,
+)
+from services.navi_offroute.router import OffrouteRouter
+
+
+# Coords ~50 m apart along a meridian at lat 44 (0.00045 deg lat ~= 50 m).
+def _line(n):
+    return [(44.0 + i * 0.00045, -114.0) for i in range(n)]
+
+
+def _edge(surface=None, use="road", road_class="unclassified", bi=0, ei=1):
+    return {"surface": surface, "use": use, "road_class": road_class,
+            "begin_shape_index": bi, "end_shape_index": ei}
+
+
+def _run(category_edges):
+    """category_edges: list of (surface, use, road_class) -> one edge per vertex step."""
+    return [_edge(surf, use, rc, bi=i, ei=i + 1)
+            for i, (surf, use, rc) in enumerate(category_edges)]
+
+
+# ── classify_surface ───────────────────────────────────────────────────────
+
+def test_classify_surface():
+    assert classify_surface({"surface": "asphalt", "use": "road"}) == PAVED
+    assert classify_surface({"surface": "paved_smooth", "use": "road"}) == PAVED
+    assert classify_surface({"surface": "gravel", "use": "road"}) == UNPAVED
+    assert classify_surface({"surface": "dirt"}) == UNPAVED          # no use -> unpaved
+    assert classify_surface({"surface": "dirt", "use": "road"}) == TRACK   # dirt road
+    assert classify_surface({"surface": "compacted", "use": "road"}) == TRACK
+    assert classify_surface({"use": "track", "surface": "gravel"}) == TRACK
+    assert classify_surface({"use": "path"}) == TRAIL
+    assert classify_surface({"use": "cycleway"}) == TRAIL
+    # paved alley (service_other) must NOT be a track
+    assert classify_surface({"surface": "paved_smooth", "use": "alley",
+                             "road_class": "service_other"}) == PAVED
+    # unpaved service road IS a track
+    assert classify_surface({"surface": "dirt", "use": "alley",
+                             "road_class": "service_other"}) == TRACK
+    assert classify_surface({"surface": "something_weird"}) is None
+
+
+# ── boundary extraction ──────────────────────────────────────────────────────
+
+def test_extract_boundaries():
+    # 5 paved edges (0..5, ~250 m) then 5 unpaved (5..10, ~250 m). Boundary at vertex 5.
+    edges = _run([("paved_smooth", "road", "tertiary")] * 5
+                 + [("gravel", "road", "unclassified")] * 5)
+    coords = _line(11)
+    out = _edges_to_candidates(edges, coords)
+    assert len(out) == 1
+    c = out[0]
+    assert c["name"] == "Surface change: paved→unpaved"
+    assert c["lat"] == coords[5][0] and c["lon"] == coords[5][1]
+    assert c["road_class"] == "tertiary"   # from the *previous* (paved) edge
+
+
+def test_noise_filter():
+    # paved (0..5, 250 m), 1-step unpaved blip (5..6, ~50 m < 100 m), paved (6..11).
+    edges = _run([("paved_smooth", "road", "tertiary")] * 5
+                 + [("gravel", "road", "unclassified")]
+                 + [("paved_smooth", "road", "tertiary")] * 5)
+    coords = _line(12)
+    out = _edges_to_candidates(edges, coords)
+    assert out == []   # blip collapsed; surrounding paved runs re-merged
+
+
+def test_no_changes_returns_empty():
+    edges = _run([("paved_smooth", "road", "tertiary")] * 8)
+    out = _edges_to_candidates(edges, _line(9))
+    assert out == []
+
+
+def test_encode_polyline6_roundtrips_with_router_decoder():
+    coords = [(43.6135, -116.2024), (43.62, -116.21), (43.6535, -116.2524)]
+    r = object.__new__(OffrouteRouter)
+    decoded = r._decode_polyline(encode_polyline6(coords))   # [lon, lat]
+    back = [(round(c[1], 5), round(c[0], 5)) for c in decoded]
+    assert back == [(round(la, 5), round(lo, 5)) for la, lo in coords]
+
+
+# ── integration with _try_hybrid_auto ───────────────────────────────────────
+
+def _winning_single_mode(distance_km, minutes):
+    return {
+        "status": "ok",
+        "route": {"type": "FeatureCollection", "features": [
+            {"type": "Feature", "properties": {"segment_type": "combined"},
+             "geometry": {"type": "LineString",
+                          "coordinates": [[-114.0, 44.0], [-114.5, 44.0]]}},
+        ]},
+        "summary": {"total_distance_km": distance_km, "total_effort_minutes": minutes,
+                    "network_distance_km": distance_km, "network_duration_minutes": minutes,
+                    "wilderness_distance_km": 0.0, "wilderness_effort_minutes": 0.0,
+                    "scenario": "D"},
+        "selected_mode": "vehicle",
+    }
+
+
+def _ok_leg(distance_km, minutes):
+    return {
+        "status": "ok",
+        "route": {"type": "FeatureCollection", "features": [
+            {"type": "Feature",
+             "properties": {"segment_type": "network", "network_mode": "x"},
+             "geometry": {"type": "LineString",
+                          "coordinates": [[-114.0, 44.0], [-114.1, 44.0]]}},
+        ]},
+        "summary": {"total_distance_km": distance_km, "total_effort_minutes": minutes,
+                    "network_distance_km": distance_km, "network_duration_minutes": minutes,
+                    "wilderness_distance_km": 0.0, "wilderness_effort_minutes": 0.0,
+                    "scenario": "D"},
+    }
+
+
+class _FakeTrailheads:
+    def __init__(self, records):
+        self._records = records
+
+    def query_trailheads_near_line(self, coords, buffer_m=2000):
+        return list(self._records)
+
+
+def test_integration_with_hybrid(monkeypatch):
+    trailhead = {"lat": 44.0, "lon": -114.20, "name": "Iron Creek TH", "road_class": "track"}
+    surface = {"lat": 44.0, "lon": -114.30, "name": "Surface change: paved→track",
+               "road_class": "unclassified"}
+    # Surface-change source returns one candidate; trailheads return one.
+    monkeypatch.setattr(
+        "services.navi_offroute.router.get_surface_change_candidates",
+        lambda coords, url: [surface])
+
+    seen_dests = []
+
+    def fake_route(self, s_lat, s_lon, e_lat, e_lon, mode="foot",
+                   boundary_mode="pragmatic", annotate_mvum=True, **k):
+        seen_dests.append((round(e_lat, 4), round(e_lon, 4)))
+        if mode == "vehicle":
+            return _ok_leg(12.0, 20.0)
+        return _ok_leg(4.0, 30.0)
+    monkeypatch.setattr(OffrouteRouter, "route", fake_route)
+
+    r = object.__new__(OffrouteRouter)
+    r.spatial_index = None
+    r.trailhead_index = _FakeTrailheads([trailhead])
+
+    best = _winning_single_mode(distance_km=20.0, minutes=120.0)
+    out = r._try_hybrid_auto(44.0, -114.0, 44.0, -114.5, "pragmatic",
+                             best, 120.0, frozenset({"vehicle", "4w", "2w", "foot"}))
+    assert out is not None
+    assert out["selected_mode"] == "hybrid"
+    # BOTH candidate types were probed as leg-1 destinations (drive-to-transition).
+    assert (round(trailhead["lat"], 4), round(trailhead["lon"], 4)) in seen_dests
+    assert (round(surface["lat"], 4), round(surface["lon"], 4)) in seen_dests