mirror of
https://github.com/zvx-echo6/navi.git
synced 2026-08-26 17:31:37 +00:00
navi-offroute: delete _try_hybrid_auto + dead Auto tests (Phase 5) (#41)
Co-authored-by: mj <mj@k7zvx.com> Co-authored-by: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
GitHub
parent
38c71a2d68
commit
521798a2c8
4 changed files with 9 additions and 789 deletions
|
|
@ -98,28 +98,10 @@ MODE_TO_COSTING = {
|
||||||
"vehicle": "auto",
|
"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
|
# 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.
|
# transitions.py). The cost_mult_stack / per-mode arrays are packed in this order.
|
||||||
MODE_ORDER = ["foot", "2w", "4w", "vehicle"]
|
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.
|
# 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).
|
# Looked up exact first, then "key:*" wildcard (see _eligible_modes_from_category).
|
||||||
_MODES_ALL = frozenset({"vehicle", "4w", "2w", "foot"})
|
_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
|
eligibility only SEEDS the start/goal modes (§6); the optimizer decides where to
|
||||||
switch modes (parking / trailhead / road terminus / surface change) instead of
|
switch modes (parking / trailhead / road terminus / surface change) instead of
|
||||||
committing to a single trip-wide mode. Supersedes the capability pick + single
|
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
|
self.route() + _try_hybrid_auto + foot-fallback flow (deleted in Phase 5).
|
||||||
no longer reached from here; Phase 5 removes them). No auto_fallback_from (§7)."""
|
No auto_fallback_from (§7)."""
|
||||||
# 1. Bootstrap eligibility -> seeds (not a routing decision).
|
# 1. Bootstrap eligibility -> seeds (not a routing decision).
|
||||||
snap_cache: dict = {}
|
snap_cache: dict = {}
|
||||||
start_eligible = self._auto_eligible_modes(start_lat, start_lon, start_category, snap_cache)
|
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):
|
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
|
"""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
|
(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)."""
|
selected_mode_set = sorted distinct modes used; NO auto_fallback_from (spec §7)."""
|
||||||
n = path.shape[0]
|
n = path.shape[0]
|
||||||
coords = []
|
coords = []
|
||||||
|
|
@ -1067,211 +1049,6 @@ class OffrouteRouter:
|
||||||
"scenario": "unified",
|
"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(
|
def _route_D_network_only(
|
||||||
self,
|
self,
|
||||||
start_lat: float, start_lon: float,
|
start_lat: float, start_lon: float,
|
||||||
|
|
|
||||||
|
|
@ -1,9 +1,8 @@
|
||||||
"""MVUM Layer 3c tests: surface-change transition candidate extraction.
|
"""MVUM Layer 3c tests: surface-change transition candidate extraction.
|
||||||
|
|
||||||
The boundary tests feed synthetic trace_attributes ``edges`` straight into the pure
|
The boundary tests feed synthetic trace_attributes ``edges`` straight into the pure
|
||||||
``_edges_to_candidates`` (no Valhalla). The integration test stubs both candidate
|
``_edges_to_candidates`` (no Valhalla). (The _try_hybrid_auto integration test was
|
||||||
sources and self.route on a bare router to confirm surface-change candidates flow
|
removed in unified-graph Phase 5 with the hybrid path.)
|
||||||
through _try_hybrid_auto alongside trailheads.
|
|
||||||
"""
|
"""
|
||||||
import pytest
|
import pytest
|
||||||
|
|
||||||
|
|
@ -88,80 +87,3 @@ def test_encode_polyline6_roundtrips_with_router_decoder():
|
||||||
decoded = r._decode_polyline(encode_polyline6(coords)) # [lon, lat]
|
decoded = r._decode_polyline(encode_polyline6(coords)) # [lon, lat]
|
||||||
back = [(round(c[1], 5), round(c[0], 5)) for c in decoded]
|
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]
|
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
|
|
||||||
|
|
|
||||||
|
|
@ -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 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
|
(The multi-modal Auto hybrid tests were removed in unified-graph Phase 5 with
|
||||||
stubbed self.route, so no Valhalla/DEM dependencies are exercised.
|
OffrouteRouter._try_hybrid_auto.)
|
||||||
"""
|
"""
|
||||||
import sqlite3
|
import sqlite3
|
||||||
|
|
||||||
|
|
@ -11,7 +11,6 @@ import numpy as np
|
||||||
import pytest
|
import pytest
|
||||||
|
|
||||||
from services.navi_offroute.mvum_transitions import TrailheadIndex
|
from services.navi_offroute.mvum_transitions import TrailheadIndex
|
||||||
from services.navi_offroute.router import OffrouteRouter
|
|
||||||
|
|
||||||
|
|
||||||
def _trailhead_db(tmp_path, points):
|
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}
|
names = {r["name"] for r in near}
|
||||||
assert "On Line" in names
|
assert "On Line" in names
|
||||||
assert "Far Away" not 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
|
|
||||||
|
|
|
||||||
|
|
@ -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
|
# monkeypatched per-mode; eligibility comes from category hints or a stubbed spatial
|
||||||
# fallback. Exercises _route_auto directly, not the Flask blueprint.
|
# 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"})
|
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 ─────────────────────────────────────────
|
# ── _eligible_modes_from_category ─────────────────────────────────────────
|
||||||
|
|
||||||
def test_eligible_modes_exact_match():
|
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
|
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 ──
|
# ── _spatial_eligible_modes — Valhalla classification.classification + .use ──
|
||||||
# _locate_on_network is monkeypatched to inject snap fixtures (road_class + 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
|
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) ───────
|
# ── 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.astar import astar_multigoal_multimode as _mm
|
||||||
from services.navi_offroute.cost import MODE_PROFILES as _PROFILES
|
from services.navi_offroute.cost import MODE_PROFILES as _PROFILES
|
||||||
|
|
|
||||||
Loading…
Add table
Add a link
Reference in a new issue