MVUM Layer 0: spatial index foundation

Additive only — no routing logic, no response-format, no Valhalla changes.

- mvum.py: add MVUMSpatialIndex. Loads mvum_roads + mvum_trails from navi.db (read-only),
  decodes the pure-WKB shape blobs with shapely, builds a shapely.strtree.STRtree, and
  keeps a parallel list of full feature records (all columns minus the blob, plus the
  parsed geometry) with a by_id lookup. Exposes query_bbox(min_lat,min_lon,max_lat,max_lon)
  and query_buffered_line(coords, tolerance_m) returning candidate records (coarse bbox +
  buffer; full parallelism filter is a TODO for PR-B). Reports road_count, trail_count,
  bbox, build_time_seconds, memory_estimate_mb (RSS delta).
- app.py: build the index once per process (singleton) at service init; stored on
  app.config[MVUM_SPATIAL_INDEX]. Failure is logged and degrades to None.
- admin.py: GET /api/admin/mvum-spatial/info (Authentik-gated, read-only) returning the
  counts/bbox/build-time/memory stats.
- tests/test_mvum_spatial.py: index loads, query_bbox returns Boise-area features,
  query_buffered_line returns a feature, admin endpoint returns counts.

Diagnostic before coding (read-only): roads_with_shape=150568/null=68,
trails_with_shape=21995/null=6746 (green), shape blobs are pure WKB MultiLineString.

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

backend/services/navi_offroute/admin.py

@@ -133,3 +133,19 @@ def navi_offroute_info():
         },
     )
     return jsonify(info)
+
+
+@bp.route('/api/admin/mvum-spatial/info')
+@require_auth
+def mvum_spatial_info():
+    """Read-only stats for the in-memory MVUM spatial index (Layer 0)."""
+    idx = current_app.config.get('MVUM_SPATIAL_INDEX')
+    if idx is None:
+        return jsonify({'status': 'error', 'message': 'MVUM spatial index not loaded'}), 503
+    return jsonify({
+        'road_count': idx.road_count,
+        'trail_count': idx.trail_count,
+        'bbox': idx.bbox,
+        'build_time_seconds': round(idx.build_time_seconds, 3),
+        'memory_estimate_mb': round(idx.memory_estimate_mb, 1),
+    })

backend/services/navi_offroute/app.py

@@ -10,6 +10,19 @@ from flask import Flask
 from shared.git_sha import git_short_sha
 
 from . import offroute_route, admin
+from .mvum import MVUMSpatialIndex
+
+# Process-wide singleton: build the MVUM spatial index once per process (per gunicorn
+# worker in prod; once across create_app() calls in tests), not once per app instance.
+_MVUM_INDEX = None
+
+
+def _get_mvum_index():
+    global _MVUM_INDEX
+    if _MVUM_INDEX is None:
+        _MVUM_INDEX = MVUMSpatialIndex()
+    return _MVUM_INDEX
+
 
 
 def create_app():
@@ -34,6 +47,13 @@ def create_app():
             )
         return response
 
+    # Load the MVUM spatial index once at service init (logs its own load line).
+    try:
+        app.config['MVUM_SPATIAL_INDEX'] = _get_mvum_index()
+    except Exception as e:
+        app.logger.warning("MVUM spatial index failed to load: %s", e)
+        app.config['MVUM_SPATIAL_INDEX'] = None
+
     app.register_blueprint(offroute_route.bp)
     app.register_blueprint(admin.bp)
     return app

backend/services/navi_offroute/mvum.py

@@ -6,17 +6,21 @@ indicating which roads/trails are open or closed to specific vehicle modes.
 
 MVUM is motor-vehicle specific — foot mode should skip this layer entirely.
 """
+import logging
 import os
 import re
 import sqlite3
+import time as _time
 import warnings
 from datetime import datetime
 from pathlib import Path
 from typing import Dict, List, Optional, Tuple, Literal
 
 import numpy as np
+import psutil
 from shapely import wkb
-from shapely.geometry import Point
+from shapely.geometry import Point, LineString, box
+from shapely.strtree import STRtree
 
 # Path to navi.db (single source of truth); env-overridable.
 DEFAULT_NAVI_DB_PATH = Path("/mnt/nav/navi.db")
@@ -27,6 +31,107 @@ def navi_db_path() -> Path:
     return Path(os.environ.get("NAVI_OFFROUTE_NAVI_DB", str(DEFAULT_NAVI_DB_PATH)))
 
 
+logger = logging.getLogger("navi_offroute.mvum_spatial")
+
+# Rough degrees-per-metre for small buffers (latitude scale; good enough for the
+# coarse candidate filter at Layer 0).
+_DEG_PER_M = 1.0 / 111320.0
+
+
+class MVUMSpatialIndex:
+    """In-memory STRtree over MVUM road + trail geometries from navi.db.
+
+    Layer 0 of the MVUM/Valhalla spatial-join work: pure spatial lookup. It parses
+    each row's WKB ``shape`` blob with shapely and indexes it in an STRtree, keeping a
+    parallel list of full feature records (every column except the raw blob, plus the
+    parsed ``geometry``). No routing logic and no response formats are touched.
+    """
+
+    def __init__(self, db_path=None):
+        t0 = _time.perf_counter()
+        proc = psutil.Process()
+        rss_before = proc.memory_info().rss
+
+        self.db_path = Path(db_path) if db_path else navi_db_path()
+        self._records = []          # aligned with self._geoms
+        self._geoms = []
+        self.by_id = {}             # feature_id -> record (full-row lookup)
+        self._min_lon = self._min_lat = float("inf")
+        self._max_lon = self._max_lat = float("-inf")
+
+        self.road_count = self._load_table("mvum_roads", "road")
+        self.trail_count = self._load_table("mvum_trails", "trail")
+
+        self._tree = STRtree(self._geoms) if self._geoms else None
+        self.build_time_seconds = _time.perf_counter() - t0
+        self.memory_estimate_mb = max(
+            0.0, (proc.memory_info().rss - rss_before) / (1024 * 1024)
+        )
+        logger.info(
+            "MVUM spatial index loaded: %d roads + %d trails in %.2f seconds",
+            self.road_count, self.trail_count, self.build_time_seconds,
+        )
+
+    def _load_table(self, table, kind):
+        count = 0
+        conn = sqlite3.connect(f"file:{self.db_path}?mode=ro", uri=True)
+        conn.row_factory = sqlite3.Row
+        try:
+            cur = conn.execute(f"SELECT * FROM {table} WHERE shape IS NOT NULL")
+            cols = [c[0] for c in cur.description]
+            for row in cur:
+                try:
+                    geom = wkb.loads(bytes(row["shape"]))
+                except Exception:
+                    continue
+                if geom.is_empty:
+                    continue
+                rec = {c: row[c] for c in cols if c != "shape"}
+                rec["kind"] = kind
+                rec["feature_id"] = f"{kind}:{row['ogc_fid']}"
+                rec["geometry"] = geom
+                self._records.append(rec)
+                self._geoms.append(geom)
+                self.by_id[rec["feature_id"]] = rec
+                minx, miny, maxx, maxy = geom.bounds
+                self._min_lon = min(self._min_lon, minx)
+                self._min_lat = min(self._min_lat, miny)
+                self._max_lon = max(self._max_lon, maxx)
+                self._max_lat = max(self._max_lat, maxy)
+                count += 1
+        finally:
+            conn.close()
+        return count
+
+    @property
+    def bbox(self):
+        """Overall extent as [min_lon, min_lat, max_lon, max_lat]."""
+        if not self._records:
+            return [0.0, 0.0, 0.0, 0.0]
+        return [self._min_lon, self._min_lat, self._max_lon, self._max_lat]
+
+    def _query_geom(self, geom):
+        if self._tree is None:
+            return []
+        return [self._records[i] for i in self._tree.query(geom)]
+
+    def query_bbox(self, min_lat, min_lon, max_lat, max_lon):
+        """Feature records whose bounding box intersects the lat/lon box (coarse)."""
+        return self._query_geom(box(min_lon, min_lat, max_lon, max_lat))
+
+    def query_buffered_line(self, coords, tolerance_m):
+        """Feature records near a (lat, lon) polyline, within ~tolerance_m.
+
+        Coarse bbox+buffer candidate filter only.
+        TODO(PR-B): apply the full parallelism filter (heading / overlap) so that
+        MVUM features which merely cross the route are rejected, keeping only those
+        that run alongside it.
+        """
+        pts = [(lon, lat) for (lat, lon) in coords]
+        geom = LineString(pts) if len(pts) >= 2 else Point(pts[0])
+        return self._query_geom(geom.buffer(tolerance_m * _DEG_PER_M))
+
+
 def parse_date_range(date_str: str) -> List[Tuple[int, int, int, int]]:
     """
     Parse MVUM date range strings like "05/01-11/30" or "06/15-10/15,12/01-03/31".

backend/services/navi_offroute/tests/test_mvum_spatial.py

@@ -0,0 +1,49 @@
+"""MVUM spatial index foundation (Layer 0) tests.
+
+Exercises MVUMSpatialIndex against the real navi.db on this host (read-only) and the
+admin-info endpoint. Pure spatial lookup — no routing or response-format coverage.
+"""
+import pytest
+
+from services.navi_offroute.mvum import MVUMSpatialIndex
+
+
+@pytest.fixture(scope="module")
+def index():
+    return MVUMSpatialIndex()
+
+
+def test_index_loads_without_error(index):
+    assert index.road_count > 0
+    assert index.trail_count > 0
+    assert len(index.by_id) == index.road_count + index.trail_count
+    assert len(index.bbox) == 4
+
+
+def test_query_bbox_returns_features(index):
+    # Generous Boise National Forest area bbox.
+    feats = index.query_bbox(43.8, -116.3, 44.4, -115.5)
+    assert len(feats) > 0
+    assert all("geometry" in f and "feature_id" in f for f in feats)
+
+
+def test_query_buffered_line_returns_features(index):
+    # Take a point on an actual indexed feature and query a small buffer around it.
+    sample = next(iter(index.by_id.values()))
+    pt = sample["geometry"].representative_point()
+    feats = index.query_buffered_line([(pt.y, pt.x)], tolerance_m=50)
+    assert len(feats) >= 1
+
+
+def test_admin_info_endpoint_returns_counts(index, monkeypatch):
+    import services.navi_offroute.app as app_mod
+    # Reuse the already-built fixture index instead of rebuilding in create_app.
+    monkeypatch.setattr(app_mod, "MVUMSpatialIndex", lambda *a, **k: index)
+    client = app_mod.create_app().test_client()
+    resp = client.get("/api/admin/mvum-spatial/info",
+                      headers={"X-Authentik-Username": "matt"})
+    assert resp.status_code == 200
+    d = resp.get_json()
+    assert d["road_count"] > 0 and d["trail_count"] > 0
+    assert len(d["bbox"]) == 4
+    assert "build_time_seconds" in d and "memory_estimate_mb" in d