From 521798a2c8a76c28379aef4d5860cdda16f863d9 Mon Sep 17 00:00:00 2001 From: malice Date: Wed, 27 May 2026 12:28:42 -0600 Subject: [PATCH] navi-offroute: delete _try_hybrid_auto + dead Auto tests (Phase 5) (#41) Co-authored-by: mj Co-authored-by: Claude Opus 4.7 (1M context) --- backend/services/navi_offroute/router.py | 229 +------------ .../tests/test_mvum_surface_change.py | 82 +---- .../tests/test_mvum_transitions.py | 180 +--------- .../navi_offroute/tests/test_offroute.py | 307 +----------------- 4 files changed, 9 insertions(+), 789 deletions(-) diff --git a/backend/services/navi_offroute/router.py b/backend/services/navi_offroute/router.py index e7fea0f..31f8fab 100755 --- a/backend/services/navi_offroute/router.py +++ b/backend/services/navi_offroute/router.py @@ -98,28 +98,10 @@ MODE_TO_COSTING = { "vehicle": "auto", } -# Auto mode probes these concrete modes in capability order (most -> least -# demanding terrain) and uses the first that yields a usable route. -AUTO_MODE_PRIORITY = ["vehicle", "4w", "2w", "foot"] - # Fixed mode index ordering for the unified-graph kernel (spec §2.1; == MODE_INDEX in # transitions.py). The cost_mult_stack / per-mode arrays are packed in this order. MODE_ORDER = ["foot", "2w", "4w", "vehicle"] -# MVUM Layer 3a: implicit multi-modal Auto. On long trips, "drive in to a trailhead, -# switch vehicles, continue offroad" can beat the single-mode winner; Auto picks that -# hybrid plan when it does. Leg times are summed with NO transition penalty, and the -# minimums below keep the suggestions sensible (no short trips / trivial detours). -MIN_HYBRID_DISTANCE_KM = 24.0 # ~15 mi: in-town trips never enter hybrid eval -HYBRID_MIN_TIME_SAVINGS_MIN = 15.0 # a hybrid must beat the winner by at least this -HYBRID_MIN_OFFROAD_KM = 0.8 # ~0.5 mi: reject trivial offroad detours -HYBRID_MAX_TRAILHEADS = 8 # cap candidates (closest to the route first) -HYBRID_OVERALL_TIMEOUT_S = 6.0 # bail hybrid eval past this, keep single-mode/best-so-far -HYBRID_EARLY_ABORT_MIN = 30.0 # a candidate beating the winner by this much ends probing -HYBRID_TRAILHEAD_BUFFER_M = 2000 # candidate trailheads within 2 km of the route -# (drive_mode, offroad_mode) transition pairs, tried at each candidate trailhead. -HYBRID_PAIRS = [("vehicle", "4w"), ("vehicle", "2w"), ("vehicle", "foot"), ("4w", "foot")] - # Per-endpoint travel-mode eligibility from an OSM-style "key:value" category hint. # Looked up exact first, then "key:*" wildcard (see _eligible_modes_from_category). _MODES_ALL = frozenset({"vehicle", "4w", "2w", "foot"}) @@ -833,8 +815,8 @@ class OffrouteRouter: eligibility only SEEDS the start/goal modes (§6); the optimizer decides where to switch modes (parking / trailhead / road terminus / surface change) instead of committing to a single trip-wide mode. Supersedes the capability pick + single - self.route() + _try_hybrid_auto + foot-fallback flow (those stay in place but are - no longer reached from here; Phase 5 removes them). No auto_fallback_from (§7).""" + self.route() + _try_hybrid_auto + foot-fallback flow (deleted in Phase 5). + No auto_fallback_from (§7).""" # 1. Bootstrap eligibility -> seeds (not a routing decision). snap_cache: dict = {} start_eligible = self._auto_eligible_modes(start_lat, start_lon, start_category, snap_cache) @@ -1008,7 +990,7 @@ class OffrouteRouter: def _render_unified_path(self, path, total_cost, meta, boundary_mode): """Render the (N,3) (row, col, mode) unified path into the GeoJSON response shape (per-mode LineString segments + transition Point markers at mode-change cells + a - combined full-path line), matching _build_response / _build_hybrid_response. + combined full-path line), matching _build_response. selected_mode_set = sorted distinct modes used; NO auto_fallback_from (spec §7).""" n = path.shape[0] coords = [] @@ -1067,211 +1049,6 @@ class OffrouteRouter: "scenario": "unified", } - def _route_coords_latlon(self, result): - """Flatten a route response's polyline to [(lat, lon), ...]. Prefers the - single "combined" full-path feature; otherwise concatenates LineStrings.""" - feats = (result.get("route") or {}).get("features", []) - for f in feats: - if (f.get("properties") or {}).get("segment_type") == "combined": - cs = (f.get("geometry") or {}).get("coordinates") or [] - return [(c[1], c[0]) for c in cs] - out = [] - for f in feats: - if (f.get("geometry") or {}).get("type") != "LineString": - continue - out.extend((c[1], c[0]) for c in (f["geometry"].get("coordinates") or [])) - return out - - def _try_hybrid_auto(self, start_lat, start_lon, end_lat, end_lon, - boundary_mode, best_result, best_minutes, intersection): - """MVUM Layer 3a: consider drive->trailhead->offroad hybrid plans. - - Returns a combined "multi" response if some trailhead transition beats the - single-mode winner by HYBRID_MIN_TIME_SAVINGS_MIN, else None (caller keeps - the single-mode winner). Leg times are summed with no transition penalty. - """ - best_summary = best_result.get("summary") or {} - if best_summary.get("total_distance_km", 0.0) < MIN_HYBRID_DISTANCE_KM: - return None - - coords = self._route_coords_latlon(best_result) - if len(coords) < 2: - return None - - _hybrid_t0 = time.perf_counter() - _gather_t0 = _hybrid_t0 - # Gather transition candidates from every available source; each yields the - # same {lat, lon, name, road_class, ...} record shape, so they mix freely and - # share the closest-first sort + cap below. - candidates = [] - # Layer 3a: MVUM/USFS trailheads near the winning polyline. - th_idx = getattr(self, "trailhead_index", None) - if th_idx is not None: - candidates += th_idx.query_trailheads_near_line( - coords, buffer_m=HYBRID_TRAILHEAD_BUFFER_M) - # Layer 3c: surface-category boundaries along the polyline (e.g. pavement -> dirt). - candidates += get_surface_change_candidates(coords, VALHALLA_URL) - # Layer 3b: OSM parking -- covers BLM/state/private land + urban areas where - # MVUM trailheads don't exist. - pk_idx = getattr(self, "parking_index", None) - if pk_idx is not None: - candidates += pk_idx.query_parking_near_line( - coords, buffer_m=HYBRID_TRAILHEAD_BUFFER_M) - if not candidates: - return None - # 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] - logger.info("hybrid candidate gathering: %d candidates in %.2fs", - len(candidates), time.perf_counter() - _gather_t0) - - # Per trailhead, leg1 depends only on drive_mode and leg2 only on offroad_mode, - # so route each distinct mode once and recombine across pairs. - drive_modes = sorted({d for d, _ in HYBRID_PAIRS}) - offroad_modes = sorted({o for _, o in HYBRID_PAIRS}) - - threshold = best_minutes - HYBRID_MIN_TIME_SAVINGS_MIN - early_abort_at = best_minutes - HYBRID_EARLY_ABORT_MIN - winner = None - winner_minutes = None - _probe_t0 = time.perf_counter() - tested = 0 - for th in candidates: - if time.perf_counter() - _hybrid_t0 > HYBRID_OVERALL_TIMEOUT_S: - logger.warning( - "hybrid eval exceeded %.1fs after %d/%d candidates; using %s", - HYBRID_OVERALL_TIMEOUT_S, tested, len(candidates), - "best hybrid so far" if winner is not None else "single-mode winner") - break - tested += 1 - leg1_by_mode = {} - for dm in drive_modes: - r = self.route(start_lat, start_lon, th["lat"], th["lon"], - mode=dm, boundary_mode=boundary_mode, annotate_mvum=False) - if r.get("status") == "ok": - leg1_by_mode[dm] = r - leg2_by_mode = {} - for om in offroad_modes: - r = self.route(th["lat"], th["lon"], end_lat, end_lon, - mode=om, boundary_mode=boundary_mode, annotate_mvum=False) - if r.get("status") != "ok": - continue - if (r.get("summary") or {}).get("total_distance_km", 0.0) < HYBRID_MIN_OFFROAD_KM: - continue # no trivial offroad detours - leg2_by_mode[om] = r - - for dm, om in HYBRID_PAIRS: - leg1 = leg1_by_mode.get(dm) - leg2 = leg2_by_mode.get(om) - if leg1 is None or leg2 is None: - continue - total = ((leg1.get("summary") or {}).get("total_effort_minutes", float("inf")) - + (leg2.get("summary") or {}).get("total_effort_minutes", float("inf"))) - if total < threshold and (winner_minutes is None or total < winner_minutes): - winner_minutes = total - winner = (leg1, leg2, dm, om, th) - # Early abort: a candidate that beats the single-mode winner by a wide - # margin is good enough -- stop probing the rest and ship it. - if winner_minutes is not None and winner_minutes <= early_abort_at: - logger.info("hybrid early-abort: candidate beats single-mode by >=%.0f min " - "after %d candidates", HYBRID_EARLY_ABORT_MIN, tested) - break - logger.info("hybrid probing took %.2fs across %d tested candidates", - time.perf_counter() - _probe_t0, tested) - - if winner is None: - return None - leg1, leg2, drive_mode, offroad_mode, th = winner - return self._build_hybrid_response(leg1, leg2, drive_mode, offroad_mode, th) - - def _build_hybrid_response(self, leg1, leg2, drive_mode, offroad_mode, trailhead): - """Combine two route legs into one "multi" scenario response with a transition - marker at the trailhead. Each leg is annotated separately (probing ran with - annotate_mvum=False); summary fields are summed across legs.""" - self._annotate_network_segments(leg1, drive_mode) - self._annotate_network_segments(leg2, offroad_mode) - - def leg_features(leg, leg_no, mode): - out = [] - for f in (leg.get("route") or {}).get("features", []): - props = dict(f.get("properties") or {}) - if props.get("segment_type") == "combined": - continue # drop per-leg full-path lines; we keep network/wilderness - # network_mode drives the map's per-mode polyline color; wilderness=foot - if "network_mode" not in props: - props["network_mode"] = ( - "foot" if props.get("segment_type") == "wilderness" else mode) - props["leg"] = leg_no - out.append({"type": "Feature", "properties": props, - "geometry": f.get("geometry")}) - return out - - features = leg_features(leg1, 1, drive_mode) - features.append({ - "type": "Feature", - "properties": { - "segment_type": "transition", - "kind": "transition", - "lat": trailhead["lat"], - "lon": trailhead["lon"], - "name": trailhead.get("name", ""), - "from_mode": drive_mode, - "to_mode": offroad_mode, - }, - "geometry": {"type": "Point", - "coordinates": [trailhead["lon"], trailhead["lat"]]}, - }) - features.extend(leg_features(leg2, 2, offroad_mode)) - - s1 = leg1.get("summary") or {} - s2 = leg2.get("summary") or {} - - def leg_summary(s, mode): - return { - "mode": mode, - "distance_km": float(s.get("total_distance_km", 0.0)), - "minutes": float(s.get("total_effort_minutes", 0.0)), - "segments_summary": { - "scenario": s.get("scenario"), - "network_km": float(s.get("network_distance_km", 0.0)), - "wilderness_km": float(s.get("wilderness_distance_km", 0.0)), - }, - } - - total_distance = (float(s1.get("total_distance_km", 0.0)) - + float(s2.get("total_distance_km", 0.0))) - total_minutes = (float(s1.get("total_effort_minutes", 0.0)) - + float(s2.get("total_effort_minutes", 0.0))) - summary = { - "total_distance_km": total_distance, - "total_effort_minutes": total_minutes, - "wilderness_minutes": (float(s1.get("wilderness_effort_minutes", 0.0)) - + float(s2.get("wilderness_effort_minutes", 0.0))), - "network_minutes": (float(s1.get("network_duration_minutes", 0.0)) - + float(s2.get("network_duration_minutes", 0.0))), - "mvum_closed_crossings": (int(s1.get("mvum_closed_crossings", 0) or 0) - + int(s2.get("mvum_closed_crossings", 0) or 0)), - "mvum_segments_annotated": (int(s1.get("mvum_segments_annotated", 0) or 0) - + int(s2.get("mvum_segments_annotated", 0) or 0)), - "scenario": "multi", - "network_mode": offroad_mode, - "wilderness_mode": "foot", - "legs": [leg_summary(s1, drive_mode), leg_summary(s2, offroad_mode)], - "transition": { - "lat": trailhead["lat"], "lon": trailhead["lon"], - "name": trailhead.get("name", ""), - "from_mode": drive_mode, "to_mode": offroad_mode, - }, - } - return { - "status": "ok", - "route": {"type": "FeatureCollection", "features": features}, - "summary": summary, - "selected_mode": "hybrid", - "scenario": "multi", - } - def _route_D_network_only( self, start_lat: float, start_lon: float, diff --git a/backend/services/navi_offroute/tests/test_mvum_surface_change.py b/backend/services/navi_offroute/tests/test_mvum_surface_change.py index 7bbf7ba..a5fc63b 100644 --- a/backend/services/navi_offroute/tests/test_mvum_surface_change.py +++ b/backend/services/navi_offroute/tests/test_mvum_surface_change.py @@ -1,9 +1,8 @@ """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. +``_edges_to_candidates`` (no Valhalla). (The _try_hybrid_auto integration test was +removed in unified-graph Phase 5 with the hybrid path.) """ import pytest @@ -88,80 +87,3 @@ def test_encode_polyline6_roundtrips_with_router_decoder(): 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]) - - # 70-min single-mode vs ~50-min hybrid = ~20 min savings: qualifies (>15) but is - # below the early-abort margin (30), so BOTH candidate sources are probed. - best = _winning_single_mode(distance_km=30.0, minutes=70.0) - out = r._try_hybrid_auto(44.0, -114.0, 44.0, -114.5, "pragmatic", - best, 70.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 diff --git a/backend/services/navi_offroute/tests/test_mvum_transitions.py b/backend/services/navi_offroute/tests/test_mvum_transitions.py index b93d007..01c6fdf 100644 --- a/backend/services/navi_offroute/tests/test_mvum_transitions.py +++ b/backend/services/navi_offroute/tests/test_mvum_transitions.py @@ -1,8 +1,8 @@ -"""MVUM Layer 3a tests: trailhead transition index + multi-modal Auto hybrids. +"""MVUM Layer 3a tests: trailhead transition index. The index tests build a TrailheadIndex from a synthetic trail_entry_points table. -The hybrid tests drive OffrouteRouter._try_hybrid_auto on a bare instance with a -stubbed self.route, so no Valhalla/DEM dependencies are exercised. +(The multi-modal Auto hybrid tests were removed in unified-graph Phase 5 with +OffrouteRouter._try_hybrid_auto.) """ import sqlite3 @@ -11,7 +11,6 @@ import numpy as np import pytest from services.navi_offroute.mvum_transitions import TrailheadIndex -from services.navi_offroute.router import OffrouteRouter def _trailhead_db(tmp_path, points): @@ -72,176 +71,3 @@ def test_query_trailheads_near_line_returns_close_only(tmp_path): names = {r["name"] for r in near} assert "On Line" in names assert "Far Away" not in names - - -# ── hybrid selection (stubbed self.route) ────────────────────────────────── - -def _ok_leg(distance_km, minutes, scenario="D"): - 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.1]]}}, - ]}, - "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": scenario, - }, - } - - -class _FakeTrailheads: - def __init__(self, records): - self._records = records - - def query_trailheads_near_line(self, coords, buffer_m=2000): - return list(self._records) - - -def _bare_router(trailheads=None): - r = object.__new__(OffrouteRouter) - r.spatial_index = None - r.trailhead_index = trailheads - return r - - -def _winning_single_mode(distance_km, minutes): - """A single-mode best_result with a combined polyline of the given distance.""" - res = _ok_leg(distance_km, minutes) - res["route"]["features"].append({ - "type": "Feature", - "properties": {"segment_type": "combined"}, - "geometry": {"type": "LineString", - "coordinates": [[-114.0, 44.0], [-114.5, 44.0]]}, - }) - res["selected_mode"] = "vehicle" - return res - - -def test_hybrid_meets_mitigations(monkeypatch): - # Short trip (< MIN_HYBRID_DISTANCE_KM) -> never goes hybrid. - th = _FakeTrailheads([{"lat": 44.0, "lon": -114.25, "name": "TH", "road_class": "track"}]) - r = _bare_router(th) - monkeypatch.setattr(OffrouteRouter, "route", - lambda self, *a, **k: _ok_leg(1.0, 5.0)) - best = _winning_single_mode(distance_km=5.0, minutes=60.0) # 5 km < 8 km - out = r._try_hybrid_auto(44.0, -114.0, 44.0, -114.5, "pragmatic", - best, 60.0, frozenset({"vehicle", "4w", "2w", "foot"})) - assert out is None - - -def test_hybrid_wins_with_big_savings(monkeypatch): - th = _FakeTrailheads([{"lat": 44.0, "lon": -114.25, "name": "Sawtooth TH", - "road_class": "track"}]) - r = _bare_router(th) - - # Drive legs are fast; offroad legs are short-but-meaningful and fast. Any leg - # combo sums to ~50 min vs the 120 min single-mode winner -> saves > 15 min. - def fake_route(self, s_lat, s_lon, e_lat, e_lon, mode="foot", - boundary_mode="pragmatic", annotate_mvum=True, **k): - if mode == "vehicle": - return _ok_leg(12.0, 20.0) - return _ok_leg(4.0, 30.0) # 4w/2w/foot offroad legs (>= 0.8 km) - monkeypatch.setattr(OffrouteRouter, "route", fake_route) - - best = _winning_single_mode(distance_km=30.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" - assert out["summary"]["scenario"] == "multi" - assert len(out["summary"]["legs"]) == 2 - assert out["summary"]["total_effort_minutes"] == pytest.approx(50.0) - # one transition marker present in the combined feature collection - kinds = [f["properties"].get("kind") for f in out["route"]["features"]] - assert kinds.count("transition") == 1 - trans = next(f for f in out["route"]["features"] - if f["properties"].get("kind") == "transition") - assert trans["properties"]["name"] == "Sawtooth TH" - assert trans["geometry"]["type"] == "Point" - - -def test_hybrid_skips_trivial_offroad_detour(monkeypatch): - th = _FakeTrailheads([{"lat": 44.0, "lon": -114.25, "name": "TH", "road_class": "track"}]) - r = _bare_router(th) - - # Offroad legs are below HYBRID_MIN_OFFROAD_KM (0.8 km) -> rejected, so even - # though the time math would otherwise win, no hybrid is produced. - def fake_route(self, s_lat, s_lon, e_lat, e_lon, mode="foot", - boundary_mode="pragmatic", annotate_mvum=True, **k): - if mode == "vehicle": - return _ok_leg(12.0, 20.0) - return _ok_leg(0.3, 5.0) # < 0.8 km offroad - monkeypatch.setattr(OffrouteRouter, "route", fake_route) - - best = _winning_single_mode(distance_km=30.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 None - - -def test_no_trailheads_falls_back_to_single_mode(monkeypatch): - r = _bare_router(_FakeTrailheads([])) # no candidates near the line - monkeypatch.setattr(OffrouteRouter, "route", - lambda self, *a, **k: _ok_leg(5.0, 10.0)) - best = _winning_single_mode(distance_km=30.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 None - - -def test_hybrid_not_taken_when_savings_below_threshold(monkeypatch): - # Hybrid total (40 min) is faster than the winner (50 min) but only by 10 min - # (< HYBRID_MIN_TIME_SAVINGS_MIN = 15) -> single-mode winner is kept. - th = _FakeTrailheads([{"lat": 44.0, "lon": -114.25, "name": "TH", "road_class": "track"}]) - r = _bare_router(th) - - def fake_route(self, s_lat, s_lon, e_lat, e_lon, mode="foot", - boundary_mode="pragmatic", annotate_mvum=True, **k): - if mode == "vehicle": - return _ok_leg(12.0, 20.0) - return _ok_leg(4.0, 20.0) - monkeypatch.setattr(OffrouteRouter, "route", fake_route) - - best = _winning_single_mode(distance_km=30.0, minutes=50.0) - out = r._try_hybrid_auto(44.0, -114.0, 44.0, -114.5, "pragmatic", - best, 50.0, frozenset({"vehicle", "4w", "2w", "foot"})) - assert out is None - - -def test_hybrid_early_abort_stops_probing(monkeypatch): - # Three trailheads, each yielding a hybrid that beats the single-mode winner by - # ~70 min (>= HYBRID_EARLY_ABORT_MIN). The first qualifying candidate must end - # probing, so not all three are routed as leg-1 destinations. - ths = [ - {"lat": 44.0, "lon": -114.20, "name": "TH1", "road_class": "track"}, - {"lat": 44.0, "lon": -114.22, "name": "TH2", "road_class": "track"}, - {"lat": 44.0, "lon": -114.24, "name": "TH3", "road_class": "track"}, - ] - r = _bare_router(_FakeTrailheads(ths)) - seen = [] - - def fake_route(self, s_lat, s_lon, e_lat, e_lon, mode="foot", - boundary_mode="pragmatic", annotate_mvum=True, **k): - seen.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) - - best = _winning_single_mode(distance_km=30.0, minutes=120.0) # hybrids save ~70 min - 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 and out["selected_mode"] == "hybrid" - # early-abort: probing stopped before all three trailheads were evaluated - probed_ths = {(d for d in seen)} # noqa: F841 (readability) - th_dests = {(round(t["lat"], 4), round(t["lon"], 4)) for t in ths} - seen_th_dests = th_dests & set(seen) - assert len(seen_th_dests) < 3 # did not probe every candidate diff --git a/backend/services/navi_offroute/tests/test_offroute.py b/backend/services/navi_offroute/tests/test_offroute.py index 0bced44..2e9cbfc 100644 --- a/backend/services/navi_offroute/tests/test_offroute.py +++ b/backend/services/navi_offroute/tests/test_offroute.py @@ -263,22 +263,11 @@ def test_admin_info_no_secrets_and_probes(client, monkeypatch): # monkeypatched per-mode; eligibility comes from category hints or a stubbed spatial # fallback. Exercises _route_auto directly, not the Flask blueprint. -from services.navi_offroute.router import OffrouteRouter, AUTO_MODE_PRIORITY +from services.navi_offroute.router import OffrouteRouter ALL_MODES = frozenset({"vehicle", "4w", "2w", "foot"}) -def _stub_route(per_mode, calls): - def stub(self, start_lat, start_lon, end_lat, end_lon, mode="foot", boundary_mode="pragmatic", **kwargs): - calls.append(mode) - return dict(per_mode[mode]) - return stub - - -def _all_ok(): - return {m: {"status": "ok"} for m in AUTO_MODE_PRIORITY} - - # ── _eligible_modes_from_category ───────────────────────────────────────── def test_eligible_modes_exact_match(): @@ -305,165 +294,6 @@ def test_eligible_modes_none_or_unknown(): assert r._eligible_modes_from_category("bogus:thing") is None -# ── probe-iteration logic (both endpoints typed -> no spatial calls) ────── - -def _typed_all(monkeypatch): - monkeypatch.setattr(OffrouteRouter, "_eligible_modes_from_category", - lambda self, cat: ALL_MODES) - - -@pytest.mark.skip(reason="superseded by unified-graph Phase 4") -def test_route_auto_picks_capability_mode(monkeypatch): - # Classify-once: typed road endpoints -> intersection = all modes -> the first - # AUTO_MODE_PRIORITY mode (vehicle) is picked and routed ONCE (no 4-mode contest). - _typed_all(monkeypatch) - calls = [] - monkeypatch.setattr(OffrouteRouter, "route", _stub_route(_all_ok(), calls)) - r = object.__new__(OffrouteRouter) - out = r._route_auto(42.0, -114.0, 42.5, -114.5, "pragmatic") - assert out["status"] == "ok" - assert out["selected_mode"] == "vehicle" - assert out["selected_mode_set"] == sorted(ALL_MODES) - assert calls == ["vehicle"] # ONE route call, not four - - -@pytest.mark.skip(reason="superseded by unified-graph Phase 4") -def test_route_auto_falls_back_to_foot_on_error(monkeypatch): - # Foot-as-last-resort: the capability-picked mode (vehicle) fails, so Auto retries - # foot ONCE and ships it, tagging auto_fallback_from for the UI. - _typed_all(monkeypatch) - calls = [] - per_mode = {m: {"status": "error", "message": f"{m} failed"} for m in AUTO_MODE_PRIORITY} - per_mode["foot"] = {"status": "ok"} - monkeypatch.setattr(OffrouteRouter, "route", _stub_route(per_mode, calls)) - r = object.__new__(OffrouteRouter) - out = r._route_auto(42.0, -114.0, 42.5, -114.5, "pragmatic") - assert out["status"] == "ok" - assert out["selected_mode"] == "foot" - assert out["auto_fallback_from"] == "vehicle" - assert out["selected_mode_set"] == sorted(ALL_MODES) # original eligibility, not foot-only - assert calls == ["vehicle", "foot"] - - -@pytest.mark.skip(reason="superseded by unified-graph Phase 4") -def test_route_auto_returns_error_when_picked_and_foot_both_fail(monkeypatch): - # Picked mode AND the foot fallback both fail -> original error surfaces, exactly - # two attempts (picked, then foot). - _typed_all(monkeypatch) - calls = [] - per_mode = {m: {"status": "error", "message": f"{m} failed"} for m in AUTO_MODE_PRIORITY} - monkeypatch.setattr(OffrouteRouter, "route", _stub_route(per_mode, calls)) - r = object.__new__(OffrouteRouter) - out = r._route_auto(42.0, -114.0, 42.5, -114.5, "pragmatic") - assert out["status"] == "error" - assert "selected_mode" not in out - assert out["selected_mode_set"] == sorted(ALL_MODES) - assert calls == ["vehicle", "foot"] - - -@pytest.mark.skip(reason="superseded by unified-graph Phase 4") -def test_route_auto_no_fallback_when_picked_is_foot(monkeypatch): - # When the picked mode is already foot (foot-only intersection), there is no second - # attempt -- foot cannot fall back to itself. - calls = [] - per_mode = {"foot": {"status": "error", "message": "foot failed"}} - monkeypatch.setattr(OffrouteRouter, "route", _stub_route(per_mode, calls)) - r = object.__new__(OffrouteRouter) - out = r._route_auto(42.0, -114.0, 42.5, -114.5, "pragmatic", - start_category="building:house", end_category="natural:peak") # -> {foot} - assert out["status"] == "error" - assert out["selected_mode_set"] == ["foot"] - assert calls == ["foot"] # no fallback attempt - - -@pytest.mark.skip(reason="superseded by unified-graph Phase 4") -def test_route_auto_tagged_road_to_road_no_spatial_probe(monkeypatch): - # Tagged road endpoints -> pure category classification, the spatial probe must - # NOT fire, and exactly one route call (mode=vehicle) is made. - calls = [] - spatial_calls = [] - monkeypatch.setattr(OffrouteRouter, "route", _stub_route(_all_ok(), calls)) - monkeypatch.setattr(OffrouteRouter, "_spatial_eligible_modes", - lambda self, lat, lon, sc: spatial_calls.append((lat, lon)) or ALL_MODES) - r = object.__new__(OffrouteRouter) - out = r._route_auto(42.0, -114.0, 42.5, -114.5, "pragmatic", - start_category="building:house", end_category="highway:residential") - assert out["selected_mode"] == "vehicle" - assert calls == ["vehicle"] - assert spatial_calls == [] # no spatial probe for tagged endpoints - - -@pytest.mark.skip(reason="superseded by unified-graph Phase 4") -def test_route_auto_untagged_spatial_called_once_per_endpoint(monkeypatch): - # One untagged endpoint -> _spatial_eligible_modes fires exactly once (for that - # endpoint only); the tagged endpoint stays a dict lookup. - calls = [] - spatial_calls = [] - monkeypatch.setattr(OffrouteRouter, "route", _stub_route(_all_ok(), calls)) - monkeypatch.setattr(OffrouteRouter, "_spatial_eligible_modes", - lambda self, lat, lon, sc: spatial_calls.append((lat, lon)) or ALL_MODES) - r = object.__new__(OffrouteRouter) - out = r._route_auto(42.0, -114.0, 42.5, -114.5, "pragmatic", - start_category="building:house", end_category=None) # end untagged - assert spatial_calls == [(42.5, -114.5)] # exactly once, for the untagged end - assert calls == ["vehicle"] - - -# ── _route_auto with category type hints (real _eligible_modes_from_category) ── - -@pytest.mark.skip(reason="superseded by unified-graph Phase 4") -def test_route_auto_address_to_address_picks_vehicle(monkeypatch): - calls = [] - monkeypatch.setattr(OffrouteRouter, "route", _stub_route(_all_ok(), calls)) - r = object.__new__(OffrouteRouter) - out = r._route_auto(42.0, -114.0, 42.5, -114.5, "pragmatic", - start_category="building:house", end_category="highway:residential") - assert out["selected_mode"] == "vehicle" - assert calls == ["vehicle"] - - -@pytest.mark.skip(reason="superseded by unified-graph Phase 4") -def test_route_auto_address_to_trailhead_picks_atv(monkeypatch): - calls = [] - monkeypatch.setattr(OffrouteRouter, "route", _stub_route(_all_ok(), calls)) - r = object.__new__(OffrouteRouter) - out = r._route_auto(42.0, -114.0, 42.5, -114.5, "pragmatic", - start_category="building:house", end_category="highway:trailhead") - assert out["selected_mode"] == "4w" - assert calls == ["4w"] # capability pick, single call - assert out["selected_mode_set"] == sorted({"4w", "2w", "foot"}) - - -@pytest.mark.skip(reason="superseded by unified-graph Phase 4") -def test_route_auto_address_to_peak_picks_foot(monkeypatch): - calls = [] - monkeypatch.setattr(OffrouteRouter, "route", _stub_route(_all_ok(), calls)) - r = object.__new__(OffrouteRouter) - out = r._route_auto(42.0, -114.0, 42.5, -114.5, "pragmatic", - start_category="building:house", end_category="natural:peak") - assert out["selected_mode"] == "foot" - assert calls == ["foot"] - assert out["selected_mode_set"] == ["foot"] - - -@pytest.mark.skip(reason="superseded by unified-graph Phase 4") -def test_route_auto_both_unknown_uses_spatial_fallback(monkeypatch): - calls = [] - spatial_calls = [] - monkeypatch.setattr(OffrouteRouter, "route", _stub_route(_all_ok(), calls)) - - def fake_spatial(self, lat, lon, snap_cache): - spatial_calls.append((lat, lon)) - return ALL_MODES - - monkeypatch.setattr(OffrouteRouter, "_spatial_eligible_modes", fake_spatial) - r = object.__new__(OffrouteRouter) - out = r._route_auto(42.0, -114.0, 42.5, -114.5, "pragmatic") # no categories - assert len(spatial_calls) == 2 # both endpoints resolved spatially - assert out["selected_mode"] == "vehicle" - assert calls == ["vehicle"] # one route after classification - - # ── _spatial_eligible_modes — Valhalla classification.classification + .use ── # _locate_on_network is monkeypatched to inject snap fixtures (road_class + use). @@ -869,141 +699,6 @@ def test_pathfind_wilderness_bbox_pad_is_1_5km(monkeypatch): assert abs((bounds["east"] - (-115.0)) - 0.015) < 1e-9 -# ── Auto classify-once priority pick (replaces the old min-time contest) ── - -@pytest.mark.skip(reason="superseded by unified-graph Phase 4") -def test_route_auto_picks_priority_not_min_time(monkeypatch): - # Classify-once picks the first AUTO_MODE_PRIORITY mode in the intersection - # (vehicle) and routes ONCE -- it no longer probes all modes to find a faster one. - # (The old min-time contest, where a slower-priority mode could win, is gone -- a - # documented PR1 trade-off in auto-rewrite-plan.md.) - _typed_all(monkeypatch) - calls = [] - per_mode = { - "vehicle": {"status": "ok", "summary": {"total_effort_minutes": 200.0}}, - "4w": {"status": "ok", "summary": {"total_effort_minutes": 50.0}}, - "2w": {"status": "ok", "summary": {"total_effort_minutes": 120.0}}, - "foot": {"status": "ok", "summary": {"total_effort_minutes": 800.0}}, - } - monkeypatch.setattr(OffrouteRouter, "route", _stub_route(per_mode, calls)) - r = object.__new__(OffrouteRouter) - out = r._route_auto(42.0, -114.0, 42.5, -114.5, "pragmatic") - assert out["status"] == "ok" - assert out["selected_mode"] == "vehicle" # first priority, picked by capability - assert calls == ["vehicle"] # routed once, no contest - - -@pytest.mark.skip(reason="superseded by unified-graph Phase 4") -def test_route_auto_per_leg_breakdown(): - # Scenario A (_build_response): foot wilderness leg + network leg -> both > 0. - r = object.__new__(OffrouteRouter) - ws = [[-116.20, 43.60], [-116.21, 43.61]] - ws_stats = {"effort_minutes": 12.0, "distance_km": 1.0, "elevation_gain_m": 10.0, - "elevation_loss_m": 5.0, "on_trail_pct": 50.0, "barrier_crossings": 0} - net = {"distance_km": 60.0, "duration_minutes": 45.0, "maneuvers": [], - "coordinates": [[-116.21, 43.61], [-116.30, 43.70]]} - out = r._build_response(ws, ws_stats, None, net, None, None, None, - "vehicle", "pragmatic", None, None, "A", 0.0, None) - assert out["status"] == "ok" - summ = out["summary"] - assert summ["wilderness_minutes"] > 0 - assert summ["network_minutes"] > 0 - # approx adds up to total - assert abs((summ["wilderness_minutes"] + summ["network_minutes"]) - summ["total_effort_minutes"]) < 1e-6 - - -@pytest.mark.skip(reason="superseded by unified-graph Phase 4") -def test_route_auto_annotates_picked_mode_once(monkeypatch): - # Classify-once routes the picked mode with annotate_mvum=False, then - # _annotate_network_segments runs exactly once, on that picked mode (vehicle). - _typed_all(monkeypatch) - calls = [] - monkeypatch.setattr(OffrouteRouter, "route", _stub_route(_all_ok(), calls)) - annotated = [] - monkeypatch.setattr(OffrouteRouter, "_annotate_network_segments", - lambda self, result, mode: annotated.append(mode)) - r = object.__new__(OffrouteRouter) - out = r._route_auto(42.0, -114.0, 42.5, -114.5, "pragmatic") - assert out["selected_mode"] == "vehicle" - assert calls == ["vehicle"] - assert annotated == ["vehicle"] # annotated once, on the picked mode - - -# ── Layer 3b: parking as a hybrid transition candidate source ── - -def _hybrid_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"}, - } - - -def _hybrid_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, - "scenario": "D"}, - "selected_mode": "vehicle", - } - - -class _FakeParking: - def __init__(self, records): - self._records = records - - def query_parking_near_line(self, coords, buffer_m=2000): - return list(self._records) - - -def test_hybrid_consumes_parking_candidates(monkeypatch): - # Only the parking index supplies candidates (no trailhead index, no surface - # changes); the parking lot must be probed as a leg-1 destination and win. - parking = {"lat": 44.0, "lon": -114.25, "name": "BLM Trailhead Lot", - "road_class": "parking", "parking_type": "surface", "access": None} - monkeypatch.setattr("services.navi_offroute.router.get_surface_change_candidates", - lambda coords, url: []) - - 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 _hybrid_ok_leg(12.0, 20.0) - return _hybrid_ok_leg(4.0, 30.0) - monkeypatch.setattr(OffrouteRouter, "route", fake_route) - - r = object.__new__(OffrouteRouter) - r.spatial_index = None - r.trailhead_index = None # no trailheads -> parking must still be gathered - r.parking_index = _FakeParking([parking]) - - best = _hybrid_winning_single_mode(distance_km=30.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" - # the parking lot was probed as a drive-to (leg-1) destination - assert (round(parking["lat"], 4), round(parking["lon"], 4)) in seen_dests - trans = next(f for f in out["route"]["features"] - if f["properties"].get("kind") == "transition") - assert trans["properties"]["name"] == "BLM Trailhead Lot" - - # ── Multi-mode A* kernel (unified-graph Phase 2; spec §2.3 / §10 / §11) ─────── from services.navi_offroute.astar import astar_multigoal_multimode as _mm from services.navi_offroute.cost import MODE_PROFILES as _PROFILES