From f20b25647339b4adb9895de23c47dc56383fa389 Mon Sep 17 00:00:00 2001 From: firestar5683 <168790843+firestar5683@users.noreply.github.com> Date: Wed, 26 Aug 2026 08:28:28 -0500 Subject: [PATCH] UI --- selfdrive/controls/lib/lane_centering.py | 46 +++++- .../controls/tests/test_lane_centering.py | 14 +- .../ui/mici/layouts/settings/software.py | 138 +++++++++++++++++- .../tests/test_software_fast_update.py | 98 +++++++++++++ selfdrive/ui/mici/onroad/model_renderer.py | 36 +++-- selfdrive/ui/onroad/model_renderer.py | 31 +++- 6 files changed, 341 insertions(+), 22 deletions(-) create mode 100644 selfdrive/ui/mici/layouts/settings/tests/test_software_fast_update.py diff --git a/selfdrive/controls/lib/lane_centering.py b/selfdrive/controls/lib/lane_centering.py index d2d17654d..10efbdfdc 100644 --- a/selfdrive/controls/lib/lane_centering.py +++ b/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 diff --git a/selfdrive/controls/tests/test_lane_centering.py b/selfdrive/controls/tests/test_lane_centering.py index a5c6551e1..6a8e88867 100644 --- a/selfdrive/controls/tests/test_lane_centering.py +++ b/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 diff --git a/selfdrive/ui/mici/layouts/settings/software.py b/selfdrive/ui/mici/layouts/settings/software.py index a329fb6d7..25034532b 100644 --- a/selfdrive/ui/mici/layouts/settings/software.py +++ b/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: diff --git a/selfdrive/ui/mici/layouts/settings/tests/test_software_fast_update.py b/selfdrive/ui/mici/layouts/settings/tests/test_software_fast_update.py new file mode 100644 index 000000000..8f06d5630 --- /dev/null +++ b/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" diff --git a/selfdrive/ui/mici/onroad/model_renderer.py b/selfdrive/ui/mici/onroad/model_renderer.py index 17d9b6300..3ad5d1aee 100644 --- a/selfdrive/ui/mici/onroad/model_renderer.py +++ b/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): diff --git a/selfdrive/ui/onroad/model_renderer.py b/selfdrive/ui/onroad/model_renderer.py index ca263b27a..3a49f40d1 100644 --- a/selfdrive/ui/onroad/model_renderer.py +++ b/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):