diff --git a/backend/services/navi_offroute/mvum_surface_change.py b/backend/services/navi_offroute/mvum_surface_change.py new file mode 100644 index 0000000..df0b01f --- /dev/null +++ b/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) diff --git a/backend/services/navi_offroute/router.py b/backend/services/navi_offroute/router.py index 79d7a9b..4307873 100755 --- a/backend/services/navi_offroute/router.py +++ b/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] diff --git a/backend/services/navi_offroute/tests/test_mvum_surface_change.py b/backend/services/navi_offroute/tests/test_mvum_surface_change.py new file mode 100644 index 0000000..f4de184 --- /dev/null +++ b/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