UI

selfdrive/controls/lib/lane_centering.py

@@ -85,7 +85,8 @@ class LaneCenteringController:
   def _covers(x, distance: float) -> bool:
     return bool(x[0] <= distance <= x[-1])
 
-  def _raw_correction(self, model_v2, v_ego: float, offset: float, e2e_authority: float) -> tuple[bool, float]:
+  @staticmethod
+  def _raw_correction(model_v2, v_ego: float, offset: float, e2e_authority: float) -> tuple[bool, float]:
     try:
       lane_lines = model_v2.laneLines
       probs = np.asarray(model_v2.laneLineProbs, dtype=float)
@@ -105,11 +106,13 @@ class LaneCenteringController:
       right_y = np.asarray(lane_lines[2].y, dtype=float)
       pos_x = np.asarray(model_v2.position.x, dtype=float)
       pos_y = np.asarray(model_v2.position.y, dtype=float)
-      if not (self._valid_path(left_x, left_y) and self._valid_path(right_x, right_y) and self._valid_path(pos_x, pos_y)):
+      if not (LaneCenteringController._valid_path(left_x, left_y) and
+              LaneCenteringController._valid_path(right_x, right_y) and
+              LaneCenteringController._valid_path(pos_x, pos_y)):
         return False, 0.0
 
       lookahead = float(np.clip(v_ego, 8.0, 35.0))
-      if not all(self._covers(x, lookahead) for x in (left_x, right_x, pos_x)):
+      if not all(LaneCenteringController._covers(x, lookahead) for x in (left_x, right_x, pos_x)):
         return False, 0.0
 
       left = float(np.interp(lookahead, left_x, left_y))
@@ -130,7 +133,7 @@ class LaneCenteringController:
 
       try:
         pos_y_std = np.asarray(model_v2.position.yStd, dtype=float)
-        if self._valid_path(pos_x, pos_y_std):
+        if LaneCenteringController._valid_path(pos_x, pos_y_std):
           path_std = float(np.interp(lookahead, pos_x, pos_y_std))
           if 0.0 <= path_std <= _E2E_MAX_PATH_STD:
             break_in = np.clip(
@@ -145,3 +148,38 @@ class LaneCenteringController:
       return True, float(2.0 * error / lookahead ** 2)
     except (AttributeError, IndexError, TypeError, ValueError):
       return False, 0.0
+
+
+def get_raw_lane_centering_correction(model_v2, v_ego: float, offset: float,
+                                      e2e_authority: float) -> tuple[bool, float]:
+  """Return the instantaneous lane-centering correction without controller filtering."""
+  return LaneCenteringController._raw_correction(model_v2, v_ego, offset, e2e_authority)
+
+
+def get_lane_centering_visual_direction(model_v2, v_ego: float, offset: float, e2e_authority: float,
+                                        enabled: bool, lat_active: bool, pause_on_signal: bool = False,
+                                        turn_signal_active: bool = False) -> int:
+  """Return 1 for a right correction, -1 for left, and 0 when no correction is active."""
+  if not enabled or not lat_active or (pause_on_signal and turn_signal_active):
+    return 0
+
+  try:
+    v_ego = float(v_ego)
+    offset = float(offset)
+    e2e_authority = float(e2e_authority)
+    if not np.isfinite([v_ego, offset, e2e_authority]).all() or v_ego < _MIN_V_EGO:
+      return 0
+    if model_v2.meta.laneChangeState != log.LaneChangeState.off:
+      return 0
+  except (AttributeError, TypeError, ValueError):
+    return 0
+
+  valid, correction = get_raw_lane_centering_correction(
+    model_v2,
+    v_ego,
+    float(np.clip(offset, -_MAX_OFFSET, _MAX_OFFSET)),
+    float(np.clip(e2e_authority, 0.0, 1.0)),
+  )
+  if not valid or not np.isfinite(correction) or correction == 0.0:
+    return 0
+  return 1 if correction > 0.0 else -1

selfdrive/controls/tests/test_lane_centering.py

@@ -3,7 +3,7 @@ from types import SimpleNamespace
 import numpy as np
 import pytest
 
-from openpilot.selfdrive.controls.lib.lane_centering import LaneCenteringController
+from openpilot.selfdrive.controls.lib.lane_centering import LaneCenteringController, get_lane_centering_visual_direction
 
 
 _V_EGO = 20.0
@@ -174,3 +174,15 @@ def test_correction_is_smoothed_and_capped():
   _, steady = _converge(model, authority=0.0)
   assert 0.0 < first < steady
   assert np.isclose(steady, 0.004 * 0.30, atol=1e-6)
+
+
+def test_visual_direction_matches_curvature_sign():
+  # Positive curvature is right in this tree's convention.
+  assert get_lane_centering_visual_direction(_model(left=-1.5, right=2.1), _V_EGO, 0.0, 0.0, True, True) == 1
+  assert get_lane_centering_visual_direction(_model(left=-2.1, right=1.5), _V_EGO, 0.0, 0.0, True, True) == -1
+
+
+def test_visual_direction_requires_both_primary_lane_lines():
+  model = _model(left=-1.5, right=2.1)
+  model.laneLineProbs[2] = 0.2
+  assert get_lane_centering_visual_direction(model, _V_EGO, 0.0, 0.0, True, True) == 0

selfdrive/ui/mici/layouts/settings/software.py

@@ -1,6 +1,11 @@
+import json
 import os
 import threading
+import time
+import urllib.error
+import urllib.request
 from collections.abc import Callable
+from dataclasses import dataclass
 from enum import IntEnum
 
 import pyray as rl
@@ -18,6 +23,40 @@ from openpilot.system.ui.widgets.label import UnifiedLabel
 from openpilot.system.ui.widgets.scroller import NavScroller
 
 UPDATER_TIMEOUT = 10.0
+FAST_UPDATE_HOLD_SECONDS = 0.8
+FAST_UPDATE_POLL_SECONDS = 0.5
+FAST_UPDATE_REQUEST_TIMEOUT = 5.0
+
+
+@dataclass
+class FastUpdateDisplayState:
+  stage: str = "idle"
+  message: str = ""
+  error: str = ""
+
+
+def _galaxy_api_url(path: str) -> str:
+  port = os.getenv("SP_GALAXY_PORT", "8082")
+  return f"http://127.0.0.1:{port}/api/update/fast{path}"
+
+
+def _galaxy_fast_update_request(path: str = "", method: str = "GET") -> dict:
+  request = urllib.request.Request(_galaxy_api_url(path), method=method)
+  try:
+    with urllib.request.urlopen(request, timeout=FAST_UPDATE_REQUEST_TIMEOUT) as response:
+      payload = json.loads(response.read().decode("utf-8"))
+  except urllib.error.HTTPError as error:
+    try:
+      payload = json.loads(error.read().decode("utf-8"))
+    except (json.JSONDecodeError, UnicodeDecodeError):
+      payload = {}
+    raise RuntimeError(payload.get("error") or str(error)) from error
+  except (OSError, urllib.error.URLError, json.JSONDecodeError, UnicodeDecodeError) as error:
+    raise RuntimeError(f"Galaxy fast update unavailable: {error}") from error
+
+  if not isinstance(payload, dict):
+    raise RuntimeError("Galaxy returned an invalid fast update response")
+  return payload
 
 
 def _split_description(desc: str) -> tuple[str, str, str, str] | None:
@@ -101,6 +140,10 @@ class CheckUpdateButton(BigButton):
     self._waiting_for_updater_t: float | None = None
     self._hide_value_t: float | None = None
     self._state: UpdaterState = UpdaterState.IDLE
+    self._press_start_t: float | None = None
+    self._press_action = ""
+    self._fast_update_state = FastUpdateDisplayState()
+    self._fast_update_state_lock = threading.Lock()
 
     ui_state.add_offroad_transition_callback(self.offroad_transition)
 
@@ -108,9 +151,24 @@ class CheckUpdateButton(BigButton):
     if ui_state.is_offroad():
       self.set_enabled(True)
 
+  def _handle_mouse_press(self, mouse_pos: MousePos):
+    super()._handle_mouse_press(mouse_pos)
+    self._press_start_t = rl.get_time()
+    self._press_action = self.get_value()
+
   def _handle_mouse_release(self, mouse_pos: MousePos):
+    held_for = 0.0 if self._press_start_t is None else rl.get_time() - self._press_start_t
+    self._press_start_t = None
     super()._handle_mouse_release(mouse_pos)
 
+    if held_for >= FAST_UPDATE_HOLD_SECONDS:
+      self._show_fast_update_confirmation()
+      return
+
+    if self._get_fast_update_state().stage == "error":
+      self._set_fast_update_state(stage="idle")
+      return
+
     if not system_time_valid():
       dlg = BigDialog("", tr("Please connect to Wi-Fi to update."))
       gui_app.push_widget(dlg)
@@ -121,13 +179,75 @@ class CheckUpdateButton(BigButton):
     self.set_icon(self._txt_update_icon)
 
     def run():
-      if self.get_value() == "download update":
+      if self._press_action == "download update":
         _request_update_download()
       else:
         _request_update_check()
 
     threading.Thread(target=run, daemon=True).start()
 
+  def _show_fast_update_confirmation(self):
+    if not system_time_valid():
+      gui_app.push_widget(BigDialog("", tr("Please connect to Wi-Fi to update.")))
+      return
+    if ui_state.started:
+      return
+
+    gui_app.push_widget(BigConfirmationDialog(
+      "slide to\nfast update",
+      self._txt_update_icon,
+      self._start_fast_update,
+      red=True,
+    ))
+
+  def _set_fast_update_state(self, *, stage: str, message: str = "", error: str = ""):
+    with self._fast_update_state_lock:
+      self._fast_update_state = FastUpdateDisplayState(stage, message, error)
+
+  def _get_fast_update_state(self) -> FastUpdateDisplayState:
+    with self._fast_update_state_lock:
+      state = self._fast_update_state
+      return FastUpdateDisplayState(state.stage, state.message, state.error)
+
+  def _start_fast_update(self):
+    if ui_state.started:
+      return
+
+    state = self._get_fast_update_state()
+    if state.stage not in ("idle", "error"):
+      return
+
+    self._set_fast_update_state(stage="starting", message="starting fast update...")
+    threading.Thread(target=self._run_fast_update, daemon=True).start()
+
+  def _run_fast_update(self):
+    try:
+      _galaxy_fast_update_request(method="POST")
+
+      while True:
+        status = _galaxy_fast_update_request("/status")
+        stage = str(status.get("stage") or "updating")
+        error = str(status.get("lastError") or "").strip()
+        message = str(
+          status.get("progressDetail") or
+          status.get("progressLabel") or
+          status.get("message") or
+          "fast update in progress..."
+        ).strip()
+
+        if error or stage == "error":
+          self._set_fast_update_state(stage="error", message="fast update failed", error=error or message)
+          return
+
+        self._set_fast_update_state(stage=stage, message=message)
+        if not bool(status.get("running")):
+          return
+        time.sleep(FAST_UPDATE_POLL_SECONDS)
+    except Exception as error:
+      current = self._get_fast_update_state()
+      if current.stage != "rebooting":
+        self._set_fast_update_state(stage="error", message="fast update failed", error=str(error))
+
   def set_value(self, value: str):
     super().set_value(value)
     self.set_text("" if value else "check for update")
@@ -136,9 +256,25 @@ class CheckUpdateButton(BigButton):
     super()._update_state()
 
     if ui_state.started:
+      self._press_start_t = None
       self.set_enabled(False)
       return
 
+    fast_update_state = self._get_fast_update_state()
+    if fast_update_state.stage != "idle":
+      self.set_rotate_icon(fast_update_state.stage not in ("error", "rebooting"))
+      if fast_update_state.stage == "error":
+        self.set_enabled(True)
+        self.set_value(fast_update_state.error or fast_update_state.message)
+      elif fast_update_state.stage == "rebooting":
+        self.set_enabled(False)
+        self.set_value("update complete\nrebooting...")
+      else:
+        self.set_enabled(False)
+        self.set_value(fast_update_state.message or "fast update in progress...")
+      self.set_text("fast update")
+      return
+
     updater_state = ui_state.params.get("UpdaterState") or ""
 
     if self._state == UpdaterState.WAITING_FOR_UPDATER:

selfdrive/ui/mici/layouts/settings/tests/test_software_fast_update.py

@@ -0,0 +1,98 @@
+from types import SimpleNamespace
+
+from openpilot.selfdrive.ui.mici.layouts.settings import software
+from openpilot.selfdrive.ui.mici.widgets.button import BigButton
+
+
+def _make_button() -> software.CheckUpdateButton:
+  button = object.__new__(software.CheckUpdateButton)
+  button._press_start_t = None
+  button._press_action = ""
+  button._fast_update_state = software.FastUpdateDisplayState()
+  button._fast_update_state_lock = software.threading.Lock()
+  return button
+
+
+def test_fast_update_uses_local_galaxy_api(monkeypatch):
+  monkeypatch.delenv("SP_GALAXY_PORT", raising=False)
+
+  assert software._galaxy_api_url("") == "http://127.0.0.1:8082/api/update/fast"
+  assert software._galaxy_api_url("/status") == "http://127.0.0.1:8082/api/update/fast/status"
+
+
+def test_long_press_opens_fast_update_confirmation(monkeypatch):
+  button = _make_button()
+  button._press_start_t = 1.0
+  confirmation_calls = []
+
+  monkeypatch.setattr(software.rl, "get_time", lambda: 1.0 + software.FAST_UPDATE_HOLD_SECONDS)
+  monkeypatch.setattr(BigButton, "_handle_mouse_release", lambda *_args: None)
+  monkeypatch.setattr(button, "_show_fast_update_confirmation", lambda: confirmation_calls.append(True))
+
+  button._handle_mouse_release(SimpleNamespace())
+
+  assert confirmation_calls == [True]
+
+
+def test_short_press_keeps_normal_updater_path(monkeypatch):
+  button = _make_button()
+  button._press_start_t = 1.0
+  button._press_action = "download update"
+  downloads = []
+  confirmations = []
+
+  monkeypatch.setattr(software.rl, "get_time", lambda: 1.0 + software.FAST_UPDATE_HOLD_SECONDS - 0.1)
+  monkeypatch.setattr(BigButton, "_handle_mouse_release", lambda *_args: None)
+  monkeypatch.setattr(software, "system_time_valid", lambda: True)
+  monkeypatch.setattr(software, "_request_update_download", lambda: downloads.append(True))
+  monkeypatch.setattr(button, "_show_fast_update_confirmation", lambda: confirmations.append(True))
+
+  class ImmediateThread:
+    def __init__(self, target, daemon):
+      self.target = target
+
+    def start(self):
+      self.target()
+
+  monkeypatch.setattr(software.threading, "Thread", ImmediateThread)
+  button.set_enabled = lambda *_args: None
+  button.set_icon = lambda *_args: None
+  button._state = software.UpdaterState.IDLE
+  button._txt_update_icon = None
+
+  button._handle_mouse_release(SimpleNamespace())
+
+  assert downloads == [True]
+  assert confirmations == []
+
+
+def test_fast_update_worker_uses_galaxy_endpoint(monkeypatch):
+  button = _make_button()
+  requests = []
+  responses = [
+    {"message": "started"},
+    {
+      "running": True,
+      "stage": "updating",
+      "progressDetail": "Fetching latest shallow commit...",
+      "lastError": "",
+    },
+    {
+      "running": False,
+      "stage": "rebooting",
+      "progressDetail": "Update complete. Please wait for device to reboot.",
+      "lastError": "",
+    },
+  ]
+
+  def request(path="", method="GET"):
+    requests.append((path, method))
+    return responses.pop(0)
+
+  monkeypatch.setattr(software, "_galaxy_fast_update_request", request)
+  monkeypatch.setattr(software.time, "sleep", lambda *_args: None)
+
+  button._run_fast_update()
+
+  assert requests == [("", "POST"), ("/status", "GET"), ("/status", "GET")]
+  assert button._get_fast_update_state().stage == "rebooting"

selfdrive/ui/mici/onroad/model_renderer.py

@@ -6,6 +6,7 @@ from cereal import messaging, car
 from dataclasses import dataclass, field
 from openpilot.common.constants import CV
 from openpilot.common.filter_simple import FirstOrderFilter
+from openpilot.selfdrive.controls.lib.lane_centering import get_lane_centering_visual_direction
 from openpilot.selfdrive.locationd.calibrationd import HEIGHT_INIT
 from openpilot.selfdrive.ui.lib.starpilot_theme import get_param_color, get_theme_color, get_visual_color, is_stock_color_scheme, with_alpha
 from openpilot.selfdrive.ui.onroad.starpilot.rainbow_path import RainbowPath
@@ -361,7 +362,24 @@ class ModelRenderer(Widget):
 
     return LeadVehicle(glow=glow, chevron=chevron, fill_alpha=int(fill_alpha))
 
-  def _lane_line_palette(self) -> tuple[bool, rl.Color, rl.Color, bool]:
+  def _lane_centering_direction(self) -> int:
+    toggles = ui_state.starpilot_toggles
+    sm = ui_state.sm
+    if not sm.valid.get("modelV2", False) or not sm.valid.get("carState", False):
+      return 0
+
+    car_state = sm["carState"]
+    return get_lane_centering_visual_direction(
+      sm["modelV2"], car_state.vEgo,
+      toggles.get("lane_center_offset", 0.0),
+      toggles.get("lane_centering_e2e_authority", 1.0),
+      bool(toggles.get("lane_centering", False)),
+      ui_state.status == UIStatus.ENGAGED or ui_state.always_on_lateral_active,
+      bool(toggles.get("lane_centering_pause_on_signal", True)),
+      bool(car_state.leftBlinker or car_state.rightBlinker),
+    )
+
+  def _lane_line_palette(self) -> tuple[bool, rl.Color, rl.Color, int]:
     stock_scheme = is_stock_color_scheme(self._params)
     line_status = UIStatus.ENGAGED if ui_state.status == UIStatus.DISENGAGED and ui_state.always_on_lateral_active else ui_state.status
 
@@ -374,15 +392,15 @@ class ModelRenderer(Widget):
     if lane_color is None:
       lane_color = STOCK_LANE_LINES_COLOR if stock_scheme else get_theme_color("LaneLines", STOCK_LANE_LINES_COLOR)
 
-    lane_centering_active = bool(ui_state.starpilot_toggles.get("lane_centering", False)) and (
-      ui_state.status == UIStatus.ENGAGED or ui_state.always_on_lateral_active
-    )
-    return stock_scheme, edge_color, lane_color, lane_centering_active
+    lane_centering_direction = self._lane_centering_direction()
+    return stock_scheme, edge_color, lane_color, lane_centering_direction
 
   def _get_ll_color(self, prob: float, adjacent: bool, left: bool, stock_scheme: bool,
-                    edge_color: rl.Color, lane_color: rl.Color, lane_centering_active: bool = False):
+                    edge_color: rl.Color, lane_color: rl.Color, lane_centering_direction: int = 0):
     alpha = np.clip(prob, 0.0, 0.7)
-    if lane_centering_active:
+    lane_centering_line = adjacent and ((lane_centering_direction > 0 and not left) or
+                                        (lane_centering_direction < 0 and left))
+    if lane_centering_line:
       color = rl.Color(OCEAN_BLUE_LANE_LINES_COLOR.r, OCEAN_BLUE_LANE_LINES_COLOR.g,
                        OCEAN_BLUE_LANE_LINES_COLOR.b, int(alpha * OCEAN_BLUE_LANE_LINES_COLOR.a))
     elif adjacent:
@@ -408,13 +426,13 @@ class ModelRenderer(Widget):
   def _draw_lane_lines(self):
     """Draw lane lines and road edges"""
     """Two closest lines should be green (lane line or road edges)"""
-    stock_scheme, edge_color, lane_color, lane_centering_active = self._lane_line_palette()
+    stock_scheme, edge_color, lane_color, lane_centering_direction = self._lane_line_palette()
     for i, lane_line in enumerate(self._lane_lines):
       if lane_line.projected_points.size == 0:
         continue
 
       color = self._get_ll_color(float(self._lane_line_probs[i]), i in (1, 2), i in (0, 1),
-                                 stock_scheme, edge_color, lane_color, lane_centering_active)
+                                 stock_scheme, edge_color, lane_color, lane_centering_direction)
       draw_polygon(self._rect, lane_line.projected_points, color)
 
     for i, road_edge in enumerate(self._road_edges):

selfdrive/ui/onroad/model_renderer.py

@@ -5,6 +5,7 @@ from cereal import messaging, car
 from dataclasses import dataclass, field
 from openpilot.common.filter_simple import FirstOrderFilter
 from openpilot.common.constants import CV
+from openpilot.selfdrive.controls.lib.lane_centering import get_lane_centering_visual_direction
 from openpilot.selfdrive.locationd.calibrationd import HEIGHT_INIT
 from openpilot.selfdrive.ui.lib.starpilot_theme import get_param_color, get_theme_color, get_visual_color, is_stock_color_scheme, with_alpha
 from openpilot.selfdrive.ui.onroad.radar_tracks import project_radar_points
@@ -369,15 +370,28 @@ class ModelRenderer(Widget):
 
     return LeadVehicle(glow=glow, chevron=chevron, fill_alpha=int(fill_alpha))
 
+  def _lane_centering_direction(self) -> int:
+    toggles = ui_state.starpilot_toggles
+    sm = ui_state.sm
+    if not sm.valid.get("modelV2", False) or not sm.valid.get("carState", False):
+      return 0
+
+    car_state = sm["carState"]
+    return get_lane_centering_visual_direction(
+      sm["modelV2"], car_state.vEgo,
+      toggles.get("lane_center_offset", 0.0),
+      toggles.get("lane_centering_e2e_authority", 1.0),
+      bool(toggles.get("lane_centering", False)),
+      ui_state.status == UIStatus.ENGAGED or ui_state.always_on_lateral_active,
+      bool(toggles.get("lane_centering_pause_on_signal", True)),
+      bool(car_state.leftBlinker or car_state.rightBlinker),
+    )
+
   def _draw_lane_lines(self):
     """Draw lane lines and road edges"""
-    lane_centering_active = bool(ui_state.starpilot_toggles.get("lane_centering", False)) and (
-      ui_state.status == UIStatus.ENGAGED or ui_state.always_on_lateral_active
-    )
+    lane_centering_direction = self._lane_centering_direction()
     lane_lines_override = get_param_color(self._params, "LaneLinesColor", STOCK_LANE_LINES_COLOR.a)
-    if lane_centering_active:
-      lane_lines_color = OCEAN_BLUE_LANE_LINES_COLOR
-    elif lane_lines_override is not None:
+    if lane_lines_override is not None:
       lane_lines_color = lane_lines_override
     elif is_stock_color_scheme(self._params):
       lane_lines_color = STOCK_LANE_LINES_COLOR
@@ -389,7 +403,10 @@ class ModelRenderer(Widget):
         continue
 
       alpha = np.clip(self._lane_line_probs[i], 0.0, 0.7)
-      color = with_alpha(lane_lines_color, int(alpha * lane_lines_color.a))
+      lane_centering_line = (lane_centering_direction > 0 and i == 2) or \
+                            (lane_centering_direction < 0 and i == 1)
+      line_color = OCEAN_BLUE_LANE_LINES_COLOR if lane_centering_line else lane_lines_color
+      color = with_alpha(line_color, int(alpha * line_color.a))
       draw_polygon(self._rect, lane_line.projected_points, color)
 
     for i, road_edge in enumerate(self._road_edges):