diff --git a/backend/services/navi_offroute/mvum_annotate.py b/backend/services/navi_offroute/mvum_annotate.py new file mode 100644 index 0000000..ad1e601 --- /dev/null +++ b/backend/services/navi_offroute/mvum_annotate.py @@ -0,0 +1,130 @@ +"""MVUM Layer 1 — per-edge access annotation. + +Walks a network leg's coordinate pairs, finds *parallel* MVUM features via the Layer-0 +spatial index, and tags each edge with its access status for the travel mode. Pure +annotation — no routing-decision changes and no Valhalla changes (that is Layer 2c). +""" +import math +from dataclasses import dataclass +from typing import List, Optional + +from shapely.geometry import Point + +from .mvum import check_access, get_mode_field, symbol_to_access + +PARALLEL_TOLERANCE_M = 10.0 # query buffer around each edge +PARALLEL_MAX_ANGLE_DEG = 45.0 # accept features whose acute angle to the edge <= this + +# Route travel mode -> MVUM access-class key understood by get_mode_field / symbol_to_access. +# foot is handled separately (MVUM is motor-vehicle specific). "auto" never reaches here: +# _route_auto resolves it to a concrete candidate before the scenario handler runs. +_ROUTE_MODE_TO_MVUM = { + "2w": "mtb", # -> e_bike_class1 + "4w": "atv", # -> atv + "vehicle": "vehicle", # -> highclearancevehicle +} + +# Restrictiveness ranking for picking the WORST status across matched features. +_RANK = {"open": 0, "unknown": 1, "closed": 2} + + +@dataclass +class EdgeAnnotation: + coord_pair_index: int + matched_features: List[str] + mvum_status: str # "open" | "closed" | "unknown" + + def to_dict(self): + return { + "coord_pair_index": self.coord_pair_index, + "matched_features": self.matched_features, + "mvum_status": self.mvum_status, + } + + +def _bearing(lat1, lon1, lat2, lon2): + """Initial bearing in degrees [0,360) from point 1 to point 2.""" + p1, p2 = math.radians(lat1), math.radians(lat2) + dl = math.radians(lon2 - lon1) + y = math.sin(dl) * math.cos(p2) + x = math.cos(p1) * math.sin(p2) - math.sin(p1) * math.cos(p2) * math.cos(dl) + return (math.degrees(math.atan2(y, x)) + 360.0) % 360.0 + + +def _acute_angle(b1, b2): + """Acute angle between two bearings, treating opposite directions as parallel (0..90).""" + d = abs(b1 - b2) % 360.0 + if d > 180.0: + d = 360.0 - d + return min(d, 180.0 - d) + + +def _feature_bearing_near(geom, lon, lat): + """Local bearing (deg) of a (Multi)LineString near point (lon,lat); None if not derivable.""" + try: + length = geom.length + if length <= 0: + return None + t = geom.project(Point(lon, lat)) + step = min(max(length * 0.05, 1e-9), length) + a = geom.interpolate(max(0.0, t - step)) + b = geom.interpolate(min(length, t + step)) + if a.equals(b): + return None + return _bearing(a.y, a.x, b.y, b.x) + except Exception: + return None + + +def _status_for_feature(record, status_field, dates_field, mvum_mode, check_date): + acc = check_access(record.get(status_field), record.get(dates_field), + record.get("seasonal"), check_date) + if acc is None: # per-class field null -> SYMBOL fallback + acc = symbol_to_access(record.get("symbol"), mvum_mode, + record.get("operationalmaintlevel")) + if acc is True: + return "open" + if acc is False: + return "closed" + return "unknown" + + +def annotate_network_edges(coords, mode, spatial_index, on_date=None): + """Annotate each consecutive coordinate pair of a network leg with MVUM access. + + coords: list[(lat, lon)]. Returns list[EdgeAnnotation], one per pair (len(coords)-1). + foot -> always "open" (MVUM is motor-vehicle specific). spatial_index None or an + unmappable mode -> "unknown" for every edge (no crash).""" + if not coords or len(coords) < 2: + return [] + n = len(coords) - 1 + + if mode == "foot": + return [EdgeAnnotation(i, [], "open") for i in range(n)] + if spatial_index is None: + return [EdgeAnnotation(i, [], "unknown") for i in range(n)] + mvum_mode = _ROUTE_MODE_TO_MVUM.get(mode) + if mvum_mode is None: + return [EdgeAnnotation(i, [], "unknown") for i in range(n)] + + status_field, dates_field = get_mode_field(mvum_mode) + check_date = (on_date.month, on_date.day) if on_date else None + + out = [] + for i in range(n): + (lat1, lon1), (lat2, lon2) = coords[i], coords[i + 1] + edge_brg = _bearing(lat1, lon1, lat2, lon2) + midlon, midlat = (lon1 + lon2) / 2.0, (lat1 + lat2) / 2.0 + matched = [] + worst = None + for rec in spatial_index.query_buffered_line( + [(lat1, lon1), (lat2, lon2)], PARALLEL_TOLERANCE_M): + fb = _feature_bearing_near(rec.get("geometry"), midlon, midlat) + if fb is None or _acute_angle(edge_brg, fb) > PARALLEL_MAX_ANGLE_DEG: + continue # crossing-but-not-parallel -> reject + matched.append(rec.get("feature_id")) + st = _status_for_feature(rec, status_field, dates_field, mvum_mode, check_date) + if worst is None or _RANK[st] > _RANK[worst]: + worst = st + out.append(EdgeAnnotation(i, matched, worst if worst is not None else "unknown")) + return out diff --git a/backend/services/navi_offroute/offroute_route.py b/backend/services/navi_offroute/offroute_route.py index 0a27c59..3a26252 100644 --- a/backend/services/navi_offroute/offroute_route.py +++ b/backend/services/navi_offroute/offroute_route.py @@ -9,7 +9,7 @@ codes. Ported from recon's lib/api.py:api_offroute / api_mvum. import logging import re -from flask import Blueprint, request, jsonify +from flask import Blueprint, request, jsonify, current_app from .router import OffrouteRouter from .mvum import MVUMReader @@ -71,6 +71,8 @@ def api_offroute(): end_category = data.get("end_category") router = OffrouteRouter() + # Inject the Layer-0 MVUM spatial index singleton for Layer-1 annotation. + router.spatial_index = current_app.config.get('MVUM_SPATIAL_INDEX') try: result = router.route( start_lat=start_lat, start_lon=start_lon, diff --git a/backend/services/navi_offroute/router.py b/backend/services/navi_offroute/router.py index 0afeca1..f754e5f 100755 --- a/backend/services/navi_offroute/router.py +++ b/backend/services/navi_offroute/router.py @@ -16,6 +16,7 @@ The user's selected mode affects: """ import gc import json +import logging import math import os import subprocess @@ -39,6 +40,9 @@ from .friction import FrictionReader, friction_to_multiplier from .barriers import BarrierReader, WildernessReader, wilderness_tif_path from .trails import TrailReader from .mvum import get_mvum_access_grid +from .mvum_annotate import annotate_network_edges + +logger = logging.getLogger("navi_offroute.router") # Configuration via env vars (extraction #8: was profile.offroute.* in recon; # promoted to dedicated env vars here — no deployment_config machinery). Read at @@ -527,6 +531,8 @@ class OffrouteRouter: self.wilderness_reader = None self.trail_reader = None self.entry_index = EntryPointIndex() + self.spatial_index = None # MVUMSpatialIndex (Layer 0), injected by the handler + self.mvum_on_date = None # optional datetime for seasonal MVUM checks def _init_readers(self): """Lazy init readers.""" @@ -880,6 +886,16 @@ class OffrouteRouter: duration_min = summary.get("time", 0) / 60 # Build response in same format as wilderness routes + # MVUM Layer 1: per-edge access annotation for the network leg. + net_edges = [] + if getattr(self, "spatial_index", None) is not None: + net_edges = annotate_network_edges( + [(c[1], c[0]) for c in network_coords], mode, + getattr(self, "spatial_index", None), getattr(self, "mvum_on_date", None)) + else: + logger.debug("MVUM spatial index unavailable; skipping per-edge annotation") + mvum_closed = sum(1 for e in net_edges if e.mvum_status == "closed") + network_feature = { "type": "Feature", "properties": { @@ -888,6 +904,7 @@ class OffrouteRouter: "duration_minutes": duration_min, "maneuvers": maneuvers, "network_mode": mode, + "edge_mvum": [e.to_dict() for e in net_edges], }, "geometry": {"type": "LineString", "coordinates": network_coords} } @@ -917,6 +934,8 @@ class OffrouteRouter: "network_minutes": float(duration_min), "on_trail_pct": 100.0, "barrier_crossings": 0, + "mvum_closed_crossings": mvum_closed, + "mvum_segments_annotated": len(net_edges), "network_mode": mode, "scenario": "D", "computation_time_s": time.time() - t0, @@ -1710,8 +1729,15 @@ class OffrouteRouter: "geometry": {"type": "LineString", "coordinates": wilderness_start} }) - # Network segment + # Network segment (MVUM Layer 1: per-edge access annotation) + net_edges = [] if network_segment: + if getattr(self, "spatial_index", None) is not None: + net_edges = annotate_network_edges( + [(c[1], c[0]) for c in network_segment["coordinates"]], mode, + getattr(self, "spatial_index", None), getattr(self, "mvum_on_date", None)) + else: + logger.debug("MVUM spatial index unavailable; skipping per-edge annotation") features.append({ "type": "Feature", "properties": { @@ -1720,6 +1746,7 @@ class OffrouteRouter: "duration_minutes": network_segment["duration_minutes"], "maneuvers": network_segment["maneuvers"], "network_mode": mode, + "edge_mvum": [e.to_dict() for e in net_edges], }, "geometry": {"type": "LineString", "coordinates": network_segment["coordinates"]} }) @@ -1821,6 +1848,8 @@ class OffrouteRouter: "network_minutes": float(network_duration_minutes), "on_trail_pct": float(on_trail_pct), "barrier_crossings": barrier_crossings, + "mvum_closed_crossings": sum(1 for e in net_edges if e.mvum_status == "closed"), + "mvum_segments_annotated": len(net_edges), "boundary_mode": boundary_mode, "wilderness_mode": "foot", "network_mode": mode, diff --git a/backend/services/navi_offroute/tests/test_mvum_annotate.py b/backend/services/navi_offroute/tests/test_mvum_annotate.py new file mode 100644 index 0000000..ceb40b2 --- /dev/null +++ b/backend/services/navi_offroute/tests/test_mvum_annotate.py @@ -0,0 +1,75 @@ +"""MVUM Layer 1 per-edge annotation tests (isolated; fake spatial index, no DB).""" +from datetime import datetime + +from shapely.geometry import LineString + +from services.navi_offroute.mvum_annotate import annotate_network_edges + + +def _feat(geom, fid="trail:1", **cols): + rec = {"geometry": geom, "feature_id": fid} + rec.update(cols) + return rec + + +class _FakeIndex: + """Returns a preset list of candidate records per query_buffered_line call (in order).""" + def __init__(self, per_call): + self.per_call = per_call + self.i = 0 + + def query_buffered_line(self, coords, tolerance_m): + r = self.per_call[self.i] if self.i < len(self.per_call) else [] + self.i += 1 + return r + + +# A NE-running edge and a parallel / perpendicular NFS-like feature near it. +EDGE = [(44.000, -114.000), (44.010, -113.990)] # (lat, lon), heading NE +PARALLEL = LineString([(-114.002, 43.998), (-113.988, 44.012)]) # NE (lon, lat) +PERPENDICULAR = LineString([(-113.988, 43.998), (-114.002, 44.012)]) # NW + + +def test_parallel_match(): + idx = _FakeIndex([[_feat(PARALLEL, atv="open", atv_datesopen=None, + seasonal="yearlong", symbol="1")]]) + out = annotate_network_edges(EDGE, "4w", idx) + assert len(out) == 1 + assert out[0].mvum_status == "open" + assert out[0].matched_features == ["trail:1"] + + +def test_perpendicular_rejected(): + idx = _FakeIndex([[_feat(PERPENDICULAR, atv="open", seasonal="yearlong", symbol="1")]]) + out = annotate_network_edges(EDGE, "4w", idx) + assert len(out) == 1 + assert out[0].mvum_status == "unknown" # parallelism filter rejected it + assert out[0].matched_features == [] + + +def test_seasonal_closure(): + # ATV open only May-Sep; querying mid-January -> closed. + idx = _FakeIndex([[_feat(PARALLEL, atv="open", atv_datesopen="05/01-09/30", + seasonal="seasonal", symbol="4")]]) + out = annotate_network_edges(EDGE, "4w", idx, on_date=datetime(2026, 1, 15)) + assert out[0].mvum_status == "closed" + + +def test_symbol_fallback(): + # Per-class field NULL, SYMBOL=3 (closed to motorized) -> closed. + idx = _FakeIndex([[_feat(PARALLEL, atv=None, atv_datesopen=None, + seasonal=None, symbol="3", operationalmaintlevel=None)]]) + out = annotate_network_edges(EDGE, "4w", idx) + assert out[0].mvum_status == "closed" + + +def test_summary_count(): + # 4 coords -> 3 edges; middle edge closed, others open -> 1 closed. + coords = [(44.000, -114.000), (44.010, -113.990), + (44.020, -113.980), (44.030, -113.970)] + open_f = _feat(PARALLEL, atv="open", seasonal="yearlong", symbol="1") + closed_f = _feat(PARALLEL, atv=None, symbol="3") + idx = _FakeIndex([[open_f], [closed_f], [open_f]]) + out = annotate_network_edges(coords, "4w", idx) + assert len(out) == 3 + assert sum(1 for e in out if e.mvum_status == "closed") == 1 diff --git a/frontend/src/components/DirectionsPanel.jsx b/frontend/src/components/DirectionsPanel.jsx index beb590c..f6bb70b 100644 --- a/frontend/src/components/DirectionsPanel.jsx +++ b/frontend/src/components/DirectionsPanel.jsx @@ -331,6 +331,17 @@ export default function DirectionsPanel({ onClose }) { )} + {/* MVUM Layer 1: network-leg closures crossed for the selected mode */} + {routeResult?.summary?.mvum_closed_crossings > 0 && ( +
+ + {`⚠ ${routeResult.summary.mvum_closed_crossings} segment(s) cross an MVUM closure for ${SELECTED_MODE_LABEL[routeResult.selected_mode || routeMode] || routeResult.selected_mode || routeMode}`} +
+ )} + {/* Boundary mode selector — hidden only for Drive (vehicle), which is pure Valhalla road routing; Auto/Foot/MTB/ATV may traverse wilderness. */} {routeMode !== "vehicle" && (