MVUM Layer 1: per-edge annotation + accurate closure counts (#23)

* MVUM Layer 1: per-edge annotation + accurate closure counts

Annotate each network-leg segment with its MVUM access status for the selected mode,
using the Layer-0 MVUMSpatialIndex. No routing-decision or Valhalla changes.

- mvum_annotate.py (new): annotate_network_edges(coords,(lat,lon)), mode, spatial_index,
  on_date) -> [EdgeAnnotation{coord_pair_index, matched_features, mvum_status}]. Walks
  consecutive pairs, queries the index (10m buffer), applies a parallelism filter (acute
  angle to the edge <=45deg, both directions), resolves per-mode access via the existing
  check_access/get_mode_field/symbol_to_access (worst/most-restrictive across matches).
  Mode map: foot->open (skip), 2w->e_bike_class1, 4w->atv, vehicle->highclearancevehicle;
  auto is already resolved to a concrete candidate upstream.
- router.py: _route_D_network_only and _build_response annotate the network leg using the
  injected self.spatial_index (getattr-guarded; skipped + debug-logged if None), attach
  edge_mvum to the network feature, and add summary.mvum_closed_crossings +
  summary.mvum_segments_annotated.
- offroute_route.py: inject app.config[MVUM_SPATIAL_INDEX] onto the router per request.
- DirectionsPanel.jsx: warning row when mvum_closed_crossings>0.
- tests/test_mvum_annotate.py: parallel match, perpendicular reject, seasonal closure,
  symbol fallback, summary count.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

* Fix seasonal date default + hoist annotation, single annotate for Auto

- Default mvum_on_date to datetime.now() at annotation time so seasonal MVUM
  openings/closings actually fire in production (was effectively None -> no seasonal).
- Hoist per-edge annotation out of _route_D_network_only and _build_response into a new
  central OffrouteRouter._annotate_network_segments(result, mode), invoked once at the end
  of route() (annotate_mvum=True). _route_auto probes with annotate_mvum=False and
  annotates only the winning candidate -> Auto runs annotation once instead of up to 4x.
  Removed the inline annotation/edge_mvum/summary blocks from both scenario handlers.

Note: the central pass filters network features on properties.segment_type == "network"
(the actual tag) rather than the spec-suggested "kind", which is not a field here.

1 new test: test_route_auto_annotates_only_winner.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

---------

Co-authored-by: Matt <mj@k7zvx.com>
Co-authored-by: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

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

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,

backend/services/navi_offroute/router.py

@@ -16,6 +16,8 @@ The user's selected mode affects:
 """
 import gc
 import json
+import logging
+from datetime import datetime
 import math
 import os
 import subprocess
@@ -39,6 +41,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 +532,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."""
@@ -600,7 +607,8 @@ class OffrouteRouter:
         mode: Literal["auto", "foot", "2w", "4w", "vehicle"] = "foot",
         boundary_mode: Literal["strict", "pragmatic", "emergency"] = "pragmatic",
         start_category: Optional[str] = None,
-        end_category: Optional[str] = None
+        end_category: Optional[str] = None,
+        annotate_mvum: bool = True,
     ) -> Dict:
         """
         Route between two points, handling all four scenarios.
@@ -635,38 +643,72 @@ class OffrouteRouter:
         # users can intentionally pin backcountry points. Auto inherits this via
         # its recursive self.route(..., mode="vehicle", ...) probe.
         if mode == "vehicle":
-            return self._route_D_network_only(
+            result = self._route_D_network_only(
                 start_lat, start_lon, end_lat, end_lon, mode
             )
-
-        # Detect network status for both endpoints
-        start_status = self._locate_on_network(start_lat, start_lon, mode)
-        end_status = self._locate_on_network(end_lat, end_lon, mode)
-
-        start_off_network = not start_status["on_network"]
-        end_off_network = not end_status["on_network"]
-
-        # Dispatch to appropriate handler
-        if not start_off_network and not end_off_network:
-            # Scenario D: on-network → on-network (pure Valhalla)
-            return self._route_D_network_only(
-                start_lat, start_lon, end_lat, end_lon, mode
-            )
-        elif not start_off_network and end_off_network:
-            # Scenario C: on-network → off-network
-            return self._route_C_network_to_wilderness(
-                start_lat, start_lon, end_lat, end_lon, mode, boundary_mode
-            )
-        elif start_off_network and not end_off_network:
-            # Scenario A: off-network → on-network
-            return self._route_A_wilderness_to_network(
-                start_lat, start_lon, end_lat, end_lon, mode, boundary_mode
-            )
         else:
-            # Scenario B: off-network → off-network
-            return self._route_B_wilderness_both(
-                start_lat, start_lon, end_lat, end_lon, mode, boundary_mode
-            )
+            # Detect network status for both endpoints
+            start_status = self._locate_on_network(start_lat, start_lon, mode)
+            end_status = self._locate_on_network(end_lat, end_lon, mode)
+
+            start_off_network = not start_status["on_network"]
+            end_off_network = not end_status["on_network"]
+
+            # Dispatch to appropriate handler
+            if not start_off_network and not end_off_network:
+                # Scenario D: on-network → on-network (pure Valhalla)
+                result = self._route_D_network_only(
+                    start_lat, start_lon, end_lat, end_lon, mode
+                )
+            elif not start_off_network and end_off_network:
+                # Scenario C: on-network → off-network
+                result = self._route_C_network_to_wilderness(
+                    start_lat, start_lon, end_lat, end_lon, mode, boundary_mode
+                )
+            elif start_off_network and not end_off_network:
+                # Scenario A: off-network → on-network
+                result = self._route_A_wilderness_to_network(
+                    start_lat, start_lon, end_lat, end_lon, mode, boundary_mode
+                )
+            else:
+                # Scenario B: off-network → off-network
+                result = self._route_B_wilderness_both(
+                    start_lat, start_lon, end_lat, end_lon, mode, boundary_mode
+                )
+
+        # MVUM Layer 1: annotate the network leg in one central pass. Auto annotates only
+        # its winning candidate (see _route_auto), so probing does not re-annotate.
+        if annotate_mvum and isinstance(result, dict) and result.get("status") == "ok":
+            self._annotate_network_segments(result, mode)
+        return result
+
+    def _annotate_network_segments(self, result, mode):
+        """Mutate result in place: attach edge_mvum to each network feature and write
+        mvum_closed_crossings + mvum_segments_annotated into the summary."""
+        if not isinstance(result, dict) or result.get("status") != "ok":
+            return
+        if getattr(self, "spatial_index", None) is None:
+            return
+        on_date = getattr(self, "mvum_on_date", None) or datetime.now()
+        features = (result.get("route") or {}).get("features", [])
+        total_closed = 0
+        total_annotated = 0
+        for feat in features:
+            props = feat.get("properties") or {}
+            # network features are tagged segment_type == "network"
+            if props.get("segment_type") != "network":
+                continue
+            coords = (feat.get("geometry") or {}).get("coordinates") or []
+            if len(coords) < 2:
+                continue
+            edges = annotate_network_edges(
+                [(c[1], c[0]) for c in coords], mode, self.spatial_index, on_date)
+            props["edge_mvum"] = [e.to_dict() for e in edges]
+            total_closed += sum(1 for e in edges if e.mvum_status == "closed")
+            total_annotated += len(edges)
+        summary = result.setdefault("summary", {})
+        summary["mvum_closed_crossings"] = total_closed
+        summary["mvum_segments_annotated"] = total_annotated
 
     def _eligible_modes_from_category(self, category: Optional[str]):
         """Eligible travel modes for an OSM "key:value" category hint, or None if the
@@ -799,7 +841,7 @@ class OffrouteRouter:
         for candidate in priority:
             result = self.route(
                 start_lat, start_lon, end_lat, end_lon,
-                mode=candidate, boundary_mode=boundary_mode
+                mode=candidate, boundary_mode=boundary_mode, annotate_mvum=False
             )
             if result.get("status") == "ok":
                 minutes = (result.get("summary") or {}).get(
@@ -813,6 +855,7 @@ class OffrouteRouter:
 
         if best_result is not None:
             best_result["selected_mode_set"] = mode_set
+            self._annotate_network_segments(best_result, best_result["selected_mode"])
             return best_result
 
         if last_error is not None:

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

backend/services/navi_offroute/tests/test_offroute.py

@@ -269,7 +269,7 @@ 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"):
+    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
@@ -863,3 +863,24 @@ def test_route_auto_per_leg_breakdown():
     assert summ["network_minutes"] > 0
     # approx adds up to total
     assert abs((summ["wilderness_minutes"] + summ["network_minutes"]) - summ["total_effort_minutes"]) < 1e-6
+
+
+def test_route_auto_annotates_only_winner(monkeypatch):
+    # 4 candidates probed (annotate_mvum=False each); _annotate_network_segments must be
+    # called exactly once, on the min-time winner.
+    _typed_all(monkeypatch)
+    calls = []
+    per_mode = {
+        "vehicle": {"status": "ok", "summary": {"total_effort_minutes": 100.0}},
+        "4w": {"status": "ok", "summary": {"total_effort_minutes": 40.0}},   # fastest
+        "2w": {"status": "ok", "summary": {"total_effort_minutes": 80.0}},
+        "foot": {"status": "ok", "summary": {"total_effort_minutes": 500.0}},
+    }
+    monkeypatch.setattr(OffrouteRouter, "route", _stub_route(per_mode, 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"] == "4w"
+    assert annotated == ["4w"]   # annotated once, on the winner only

frontend/src/components/DirectionsPanel.jsx

@@ -331,6 +331,17 @@ export default function DirectionsPanel({ onClose }) {
         </div>
       )}
 
+      {/* MVUM Layer 1: network-leg closures crossed for the selected mode */}
+      {routeResult?.summary?.mvum_closed_crossings > 0 && (
+        <div
+          className="flex items-center justify-center gap-1 py-1.5 text-xs rounded-lg"
+          style={{ background: "var(--accent-muted)", color: "var(--accent)" }}
+        >
+          <AlertTriangle size={14} />
+          <span>{`⚠ ${routeResult.summary.mvum_closed_crossings} segment(s) cross an MVUM closure for ${SELECTED_MODE_LABEL[routeResult.selected_mode || routeMode] || routeResult.selected_mode || routeMode}`}</span>
+        </div>
+      )}
+
       {/* Boundary mode selector — hidden only for Drive (vehicle), which is pure
           Valhalla road routing; Auto/Foot/MTB/ATV may traverse wilderness. */}
       {routeMode !== "vehicle" && (