diff --git a/backend/services/navi_offroute/router.py b/backend/services/navi_offroute/router.py index f10fea7..ba89b15 100755 --- a/backend/services/navi_offroute/router.py +++ b/backend/services/navi_offroute/router.py @@ -838,6 +838,30 @@ class OffrouteRouter: end_eligible = self._auto_eligible_modes(end_lat, end_lon, end_category, snap_cache) seed_set = sorted(start_eligible | end_eligible) + # === Valhalla bypass: pure road↔road skips the raster pipeline entirely === + # When BOTH endpoints are tagged with a vehicle-eligible category (only _MODES_ALL + # contains "vehicle"), the trip is on-road end-to-end -> hand it straight to the same + # inline-Valhalla path explicit vehicle requests use, saving the ~7.5s raster build + + # unified A*. Untagged endpoints (category -> None) and any non-paved category fall + # through to the unified flow. Boundary-mode MVUM exclusions are not applied here (urban + # roads aren't MVUM-gated); this matches a pragmatic in-town vehicle route. + bp_start = self._eligible_modes_from_category(start_category) + bp_end = self._eligible_modes_from_category(end_category) + if (bp_start is not None and bp_end is not None + and "vehicle" in bp_start and "vehicle" in bp_end): + _bt0 = time.perf_counter() + bypass = self._route_D_network_only(start_lat, start_lon, end_lat, end_lon, "vehicle") + if bypass.get("status") == "ok": + bypass["selected_mode"] = "vehicle" + bypass["selected_mode_set"] = ["vehicle"] + bypass.setdefault("summary", {})["auto_bypass"] = True + logger.info("auto bypass: vehicle road↔road via Valhalla in %.3fs", + time.perf_counter() - _bt0) + return bypass + logger.warning("auto bypass attempted but Valhalla returned %s; falling through to unified", + bypass.get("status")) + # === end bypass === + # 2-3. bbox covering both endpoints + the shared rasters (one fetch, all modes). try: (elevation, friction_mult, friction_raw, trails, barriers, diff --git a/backend/services/navi_offroute/tests/test_offroute.py b/backend/services/navi_offroute/tests/test_offroute.py index feb3bb9..0939f4b 100644 --- a/backend/services/navi_offroute/tests/test_offroute.py +++ b/backend/services/navi_offroute/tests/test_offroute.py @@ -1292,3 +1292,83 @@ def test_route_auto_perf_under_5s(monkeypatch): assert out["status"] == "ok", out assert out["selected_mode_set"] == ["foot"] assert elapsed <= 5.0, f"_route_auto on ~50 km grid took {elapsed:.2f}s > 5.0s" + + +# ── Auto Valhalla bypass: pure road↔road skips the raster pipeline ──────────── +import logging as _bp_logging + + +class _BypassOKResp: + status_code = 200 + def json(self): + return {"trip": {"legs": [{"shape": "", "maneuvers": []}], + "summary": {"length": 1.2, "time": 90}}} + + +class _BypassErrResp: + status_code = 500 + text = "valhalla boom" + + +def _bp_fetch_should_not_run(*a, **k): + raise AssertionError("_fetch_auto_rasters must NOT run when the bypass fires") + + +def _bp_fetch_sentinel(*a, **k): + raise RuntimeError("FETCH_REACHED") # proves we fell through to the unified flow + + +def test_route_auto_road_road_uses_valhalla_bypass(monkeypatch): + r = object.__new__(OffrouteRouter) + monkeypatch.setattr(OffrouteRouter, "_fetch_auto_rasters", _bp_fetch_should_not_run) + calls = [] + real_D = OffrouteRouter._route_D_network_only + monkeypatch.setattr(OffrouteRouter, "_route_D_network_only", + lambda self, *a, **k: (calls.append(a) or real_D(self, *a, **k))) + monkeypatch.setattr(_p4router.requests, "post", lambda *a, **k: _BypassOKResp()) + + out = r._route_auto(42.5558, -114.4701, 42.5644, -114.4631, "pragmatic", + start_category="highway:residential", end_category="highway:residential") + assert out["status"] == "ok" + assert out["selected_mode"] == "vehicle" + assert out["selected_mode_set"] == ["vehicle"] + assert out["summary"]["auto_bypass"] is True + assert out["summary"]["scenario"] == "D" + assert len(calls) == 1 # _route_D_network_only called exactly once + + +def test_route_auto_road_offroad_skips_bypass(monkeypatch): + r = object.__new__(OffrouteRouter) + monkeypatch.setattr(OffrouteRouter, "_route_D_network_only", + lambda *a, **k: (_ for _ in ()).throw(AssertionError("D must not run"))) + monkeypatch.setattr(OffrouteRouter, "_fetch_auto_rasters", _bp_fetch_sentinel) + # start = paved road (vehicle eligible), end = footway (foot-only, no vehicle) -> no bypass. + out = r._route_auto(42.5558, -114.4701, 42.5878, -114.5550, "pragmatic", + start_category="highway:residential", end_category="highway:footway") + assert out["status"] == "error" + assert "Failed to load terrain" in out["message"] # reached _fetch_auto_rasters (unified flow) + + +def test_route_auto_untagged_skips_bypass(monkeypatch): + r = object.__new__(OffrouteRouter) + monkeypatch.setattr(OffrouteRouter, "_spatial_eligible_modes", + lambda self, lat, lon, cache: frozenset({"foot"})) + monkeypatch.setattr(OffrouteRouter, "_route_D_network_only", + lambda *a, **k: (_ for _ in ()).throw(AssertionError("D must not run"))) + monkeypatch.setattr(OffrouteRouter, "_fetch_auto_rasters", _bp_fetch_sentinel) + out = r._route_auto(42.5558, -114.4701, 42.5644, -114.4631, "pragmatic") # no categories + assert out["status"] == "error" + assert "Failed to load terrain" in out["message"] # bypass skipped, unified flow reached + + +def test_route_auto_bypass_falls_through_on_valhalla_error(monkeypatch, caplog): + r = object.__new__(OffrouteRouter) + monkeypatch.setattr(_p4router.requests, "post", lambda *a, **k: _BypassErrResp()) + monkeypatch.setattr(OffrouteRouter, "_fetch_auto_rasters", _bp_fetch_sentinel) + with caplog.at_level(_bp_logging.WARNING, logger="navi_offroute.router"): + out = r._route_auto(42.5558, -114.4701, 42.5644, -114.4631, "pragmatic", + start_category="highway:residential", end_category="highway:residential") + assert "auto bypass attempted but Valhalla returned" in caplog.text + assert out["status"] == "error" + assert "Failed to load terrain" in out["message"] # fell through to the unified flow + assert "auto_bypass" not in (out.get("summary") or {})