navigation: allow LAN API host override

openpilot/selfdrive/navigation/auth.py

@@ -16,6 +16,12 @@ def _as_text(value) -> str:
   return str(value or "").strip()
 
 
+def nav_api_host(params: Params | None = None) -> str:
+  params = params if params is not None else Params()
+  override = _as_text(params.get("NavApiHost"))
+  return (override or NAV_API_HOST).rstrip("/")
+
+
 def nav_auth(params: Params | None = None) -> tuple[str, dict[str, str]]:
   params = params if params is not None else Params()
   device_id = _as_text(NAV_DEVICE_ID or params.get("DongleId"))

openpilot/selfdrive/navigation/navd.py

@@ -9,7 +9,7 @@ import openpilot.cereal.messaging as messaging
 from openpilot.common.params import Params
 from openpilot.common.realtime import Ratekeeper
 from openpilot.common.swaglog import cloudlog
-from openpilot.selfdrive.navigation.auth import NAV_API_HOST, nav_auth
+from openpilot.selfdrive.navigation.auth import nav_api_host, nav_auth
 from openpilot.selfdrive.navigation.helpers import latest_gps_source
 from openpilot.selfdrive.navigation.route import along_path, local_meters
 
@@ -52,7 +52,7 @@ class RouteEngine:
     self.reroute_counter = 0
 
     self.mapbox_token = os.environ.get("MAPBOX_TOKEN")
-    self.mapbox_host = "https://api.mapbox.com" if self.mapbox_token is not None else NAV_API_HOST + '/v1/nav'
+    self.mapbox_host = "https://api.mapbox.com" if self.mapbox_token is not None else nav_api_host(self.params) + '/v1/nav'
 
   def update(self):
     self.sm.update(0)

openpilot/selfdrive/navigation/tests/test_navigation.py

@@ -103,6 +103,14 @@ def route_engine(testcase, sm, pm, values):
 
 
 class TestNavAuth(unittest.TestCase):
+  def test_api_host_default_and_override(self):
+    params = MagicMock()
+    params.get.return_value = None
+    with patch.object(nav_auth_module, 'NAV_API_HOST', 'https://nav.example/'):
+      self.assertEqual(nav_auth_module.nav_api_host(params), 'https://nav.example')
+    params.get.return_value = b'http://192.168.3.203:5051/'
+    self.assertEqual(nav_auth_module.nav_api_host(params), 'http://192.168.3.203:5051')
+
   def test_reads_device_token(self):
     with tempfile.TemporaryDirectory() as directory:
       token_path = Path(directory) / 'nav_token'
@@ -187,7 +195,7 @@ class TileServerCase(unittest.TestCase):
       ('nav_auth', self.nav_auth),
       ('Params', MagicMock()),
       ('PC', True),
-      ('NAV_API_HOST', f'http://127.0.0.1:{self.server.server_port}'),
+      ('nav_api_host', lambda _params: f'http://127.0.0.1:{self.server.server_port}'),
     ]
     for name, value in patch_values:
       self.enterContext(patch(f'openpilot.selfdrive.navigation.tiles.{name}', value))

openpilot/selfdrive/navigation/tiles.py

@@ -9,7 +9,7 @@ import requests
 from openpilot.common.hardware import PC
 from openpilot.common.hardware.hw import Paths
 from openpilot.common.params import Params
-from openpilot.selfdrive.navigation.auth import NAV_API_HOST, nav_auth
+from openpilot.selfdrive.navigation.auth import nav_api_host, nav_auth
 from openpilot.selfdrive.navigation.pmtiles import HEADER_LENGTH
 
 CACHE_LIMIT = 64 * 1024 * 1024
@@ -61,7 +61,7 @@ class RemoteTiles:
       if time.time() >= self.expires - 60:  # noqa: TID251 - SAS expiry is a Unix timestamp
         device_id, auth_headers = nav_auth(self.params)
         device_path = urllib.parse.quote(device_id, safe='')
-        response = self._request_with_retries(f'{NAV_API_HOST}/v1/{device_path}/nav/tiles', auth_headers)
+        response = self._request_with_retries(f'{nav_api_host(self.params)}/v1/{device_path}/nav/tiles', auth_headers)
         response.raise_for_status()
         metadata = response.json()
         self.url, self.expires = metadata['url'], metadata['expires']