diff --git a/scripts/launch_onroad_desktop.sh b/scripts/launch_onroad_desktop.sh index fe1978c03..1c21cbf5e 100755 --- a/scripts/launch_onroad_desktop.sh +++ b/scripts/launch_onroad_desktop.sh @@ -27,13 +27,14 @@ env_var_truthy() { usage() { cat <<'EOF' Usage: - ./onroad [jobs] [--c3 | --c4 | --raybig | --all | --replay-only] [--galaxy] [-nav] [-alert] [--cem] [--prefix name] + ./onroad [jobs] [--c3 | --c4 | --raybig | --all | --replay-only] [--galaxy] [-nav] [--offroad] [-alert] [--cem] [--prefix name] Examples: ./onroad ./onroad --c3 ./onroad --c4 --start 30 ./onroad --c4 -nav + ./onroad --raybig --nav --offroad --demo ./onroad --c4 --cem --demo ./onroad --raybig --cem --demo ./onroad --raybig --cem -alert --demo --no-loop @@ -47,6 +48,7 @@ Notes: - Use multiple UI flags together if you want more than one desktop UI at once. - --galaxy starts a local Galaxy web session with the same preview params and prints the localhost URL. It blocks replay's logged customReserved9 stream so Galaxy can own the live Testing Grounds publisher. - -nav injects a fake navigation demo stream and blocks replay from publishing navInstruction/navRoute. + - --offroad is only valid with --nav; together they preview the offroad Quick Start card and do not start the fake on-road nav publisher. - --cem publishes fake CEM statuses for desktop visual review in the raylib UIs. - --csc publishes a fake starpilotPlan stream that forces the CSC glow to render on desktop UI. - -alert blocks replay from publishing selfdriveState and fires a fake critical full-screen red alert (alertSize=full, alertStatus=critical) 20 seconds after the demo publisher starts (10s for replay route + UI to come up, plus 10s for the user to open Settings). Default alert text mimics a real controlsMismatch event; run tools/replay/fake_alert_demo.py directly to override --text1/--text2/--delay. @@ -66,6 +68,7 @@ UI_SELECTION_EXPLICIT=0 LEGACY_UI_SELECTION="" REPLAY_ONLY=0 NAV_DEMO=0 +OFFROAD_DEMO=0 CEM_DEMO=0 ALERT_DEMO=0 CSC_DEMO=0 @@ -116,6 +119,10 @@ parse_args() { NAV_DEMO=1 shift ;; + --offroad) + OFFROAD_DEMO=1 + shift + ;; --cem|--mici-widget-demo|--widget-demo) CEM_DEMO=1 shift @@ -398,6 +405,7 @@ prepare_env() { export SP_RAYBIG_FAKE_WIFI=0 export SP_ALLOW_DESKTOP_FAKE_WIFI=0 export SP_ONROAD_NAV_DEMO="${NAV_DEMO}" + export SP_ONROAD_OFFROAD_DEMO="${OFFROAD_DEMO}" export SP_CEM_DEMO="${CEM_DEMO}" export SP_ONROAD_ALERT_DEMO="${ALERT_DEMO}" export SP_ONROAD_CSC_DEMO="${CSC_DEMO}" @@ -601,6 +609,9 @@ launch_python_ui() { local big="$1" ( export BIG="${big}" + if [[ "${OFFROAD_DEMO}" == "1" ]]; then + export PRIME_TYPE=0 + fi exec "${ROOT_DIR}/.venv/bin/python3" "${ROOT_DIR}/selfdrive/ui/ui.py" ) & UI_PIDS+=("$!") @@ -631,6 +642,11 @@ if [[ "${NAV_DEMO}" == "1" ]]; then ensure_nav_demo_replay_blocklist fi +if [[ "${OFFROAD_DEMO}" == "1" && "${NAV_DEMO}" != "1" ]]; then + echo "--offroad requires --nav." >&2 + exit 1 +fi + if [[ "${ALERT_DEMO}" == "1" ]]; then ensure_alert_demo_replay_blocklist fi @@ -694,7 +710,7 @@ fi echo "Starting replay: ${REPLAY_ARGS[*]}" launch_replay -if [[ "${NAV_DEMO}" == "1" ]]; then +if [[ "${NAV_DEMO}" == "1" && "${OFFROAD_DEMO}" != "1" ]]; then launch_nav_demo fi diff --git a/selfdrive/ui/layouts/home.py b/selfdrive/ui/layouts/home.py index 2c06190b3..bac83f2b7 100644 --- a/selfdrive/ui/layouts/home.py +++ b/selfdrive/ui/layouts/home.py @@ -6,6 +6,7 @@ from openpilot.selfdrive.ui.ui_state import ui_state from openpilot.selfdrive.ui.widgets.offroad_alerts import UpdateAlert, OffroadAlert from openpilot.selfdrive.ui.widgets.exp_mode_button import ExperimentalModeButton from openpilot.selfdrive.ui.widgets.drive_stats import DriveStatsDashboard +from openpilot.selfdrive.ui.widgets.home_info_card import HomeInfoCard from openpilot.selfdrive.ui.widgets.setup import SetupWidget from openpilot.selfdrive.ui.lib.starpilot_version import starpilot_display_description from openpilot.system.ui.lib.text_measure import measure_text_cached @@ -58,12 +59,14 @@ class HomeLayout(Widget): self._drive_stats = DriveStatsDashboard(self.params) self._setup_widget = SetupWidget() + self._home_info_card = self._child(HomeInfoCard(params=self.params, drive_stats=self._drive_stats)) self._exp_mode_button = ExperimentalModeButton() self._setup_callbacks() def show_event(self): self._exp_mode_button.show_event() + super().show_event() self.last_refresh = time.monotonic() self._refresh() @@ -212,12 +215,13 @@ class HomeLayout(Widget): self.right_column_rect.height - exp_height - SPACING, ) if ui_state.prime_state.is_paired(): - self._drive_stats.render_records(setup_rect) + self._home_info_card.render(setup_rect) else: self._setup_widget.render(setup_rect) def _refresh(self): self._drive_stats.refresh() + self._home_info_card.refresh() self._version_text = self._get_version_text() update_available = self.update_alert.refresh() alert_count = self.offroad_alert.refresh() diff --git a/selfdrive/ui/layouts/settings/starpilot/navigation.py b/selfdrive/ui/layouts/settings/starpilot/navigation.py index d417bd1c2..e74e00f33 100644 --- a/selfdrive/ui/layouts/settings/starpilot/navigation.py +++ b/selfdrive/ui/layouts/settings/starpilot/navigation.py @@ -1,6 +1,5 @@ from __future__ import annotations -import hashlib import json import math import queue @@ -26,11 +25,21 @@ from openpilot.selfdrive.ui.layouts.settings.starpilot.aethergrid import ( with_alpha, ) from openpilot.selfdrive.ui.layouts.settings.starpilot.panel import FrameCachedParams, _SettingsPage +from openpilot.starpilot.navigation import destination_store as _destination_store from openpilot.starpilot.navigation.destination_store import ( - load_recent_destinations, + NavigationDestinationStore, + add_favorite_destination, + favorite_destination_id, + favorite_matches_target, + favorite_payload_for_galaxy, + load_favorite_destinations, normalize_destination_payload, - parse_destination_json, - update_recent_destinations, + normalize_favorite_destination, + ordered_favorite_destinations, + remove_favorite_destination, + routing_configured, + same_destination, + update_favorite_destination, ) from openpilot.system.ui.lib.application import gui_app from openpilot.system.ui.lib.multilang import tr @@ -39,231 +48,21 @@ from openpilot.system.ui.widgets.confirm_dialog import ConfirmDialog from openpilot.system.ui.widgets.keyboard import Keyboard -NAVIGATION_DESTINATION_KEY = "NavDestination" -RECENT_DESTINATIONS_KEY = "ApiCache_NavDestinations" -FAVORITE_DESTINATIONS_KEY = "FavoriteDestinations" -NAV_INSTRUCTION_STATE_KEY = "NavInstructionState" -NAV_INSTRUCTION_COLLAPSED_KEY = "NavInstructionCollapsed" +FAVORITE_DESTINATIONS_KEY = _destination_store.FAVORITE_DESTINATIONS_KEY +NAVIGATION_DESTINATION_KEY = _destination_store.NAVIGATION_DESTINATION_KEY +NAV_INSTRUCTION_COLLAPSED_KEY = _destination_store.NAV_INSTRUCTION_COLLAPSED_KEY +NAV_INSTRUCTION_STATE_KEY = _destination_store.NAV_INSTRUCTION_STATE_KEY +RECENT_DESTINATIONS_KEY = _destination_store.RECENT_DESTINATIONS_KEY - -def _coerce_float(value: Any) -> float | None: - try: - return float(value) - except (TypeError, ValueError): - return None - - -def _json_value(raw_value: Any, default: Any) -> Any: - if isinstance(raw_value, (list, dict)): - return raw_value - if isinstance(raw_value, bytes): - raw_value = raw_value.decode("utf-8", errors="replace") - if not raw_value: - return default - try: - return json.loads(raw_value) - except (TypeError, ValueError, json.JSONDecodeError): - return default - - -def _favorite_destination_id(payload: dict[str, Any]) -> str: - """Keep the exact favorite ID formula used by Galaxy's existing backend.""" - raw = f"{payload.get('longitude')},{payload.get('latitude')}|{payload.get('routeId') or ''}|{payload.get('name') or ''}" - return hashlib.sha1(raw.encode()).hexdigest() - - -def _favorite_payload_for_galaxy(destination: dict[str, Any]) -> dict[str, Any]: - favorite = dict(destination) - favorite.setdefault("routeId", "main") - return favorite - - -def _normalize_favorite_payload(payload: Any) -> dict[str, Any] | None: - if not isinstance(payload, dict): - return None - - name = str(payload.get("name") or payload.get("place_name") or "").strip() - latitude = _coerce_float(payload.get("latitude")) - longitude = _coerce_float(payload.get("longitude")) - if not name or latitude is None or longitude is None: - return None - - normalized = dict(payload) - normalized.update({"name": name, "latitude": latitude, "longitude": longitude}) - normalized["id"] = str(payload.get("id") or _favorite_destination_id(payload)) - return normalized - - -def _load_favorite_destinations(raw_value: Any) -> list[dict[str, Any]]: - payload = _json_value(raw_value, []) - if not isinstance(payload, list): - return [] - return [ - normalized - for entry in payload - if (normalized := _normalize_favorite_payload(entry)) is not None - ] - - -def _favorite_matches_target( - favorite: dict[str, Any], - target: dict[str, Any], - *, - allow_route_id_only: bool = False, -) -> bool: - target_id = target.get("id") - if target_id: - return favorite.get("id") == target_id - if allow_route_id_only and target.get("routeId"): - return favorite.get("routeId") == target.get("routeId") - return ( - favorite.get("routeId") == target.get("routeId") and - favorite.get("latitude") == target.get("latitude") and - favorite.get("longitude") == target.get("longitude") and - favorite.get("name") == target.get("name") - ) - - -def _add_favorite_destination(raw_value: Any, favorite: dict[str, Any]) -> list[dict[str, Any]]: - favorites = _load_favorite_destinations(raw_value) - normalized = _normalize_favorite_payload(favorite) - if normalized is not None and not any(item.get("id") == normalized["id"] for item in favorites): - favorites.append(normalized) - return favorites - - -def _remove_favorite_destination(raw_value: Any, target: dict[str, Any]) -> list[dict[str, Any]]: - favorites = _load_favorite_destinations(raw_value) - return [favorite for favorite in favorites if not _favorite_matches_target(favorite, target)] - - -def _update_favorite_destination( - raw_value: Any, - target: dict[str, Any], - *, - name: str | None = None, - is_home: bool | None = None, - is_work: bool | None = None, -) -> list[dict[str, Any]] | None: - favorites = _load_favorite_destinations(raw_value) - target_index = next( - ( - index for index, favorite in enumerate(favorites) - if _favorite_matches_target(favorite, target, allow_route_id_only=True) - ), - None, - ) - if target_index is None: - return None - - if is_home: - for favorite in favorites: - favorite.pop("is_home", None) - if is_work: - for favorite in favorites: - favorite.pop("is_work", None) - - favorite = favorites[target_index] - if name is not None: - normalized_name = str(name).strip() - if normalized_name: - favorite["name"] = normalized_name - if is_home is not None: - if is_home: - favorite["is_home"] = True - favorite.pop("is_work", None) - else: - favorite.pop("is_home", None) - if is_work is not None: - if is_work: - favorite["is_work"] = True - favorite.pop("is_home", None) - else: - favorite.pop("is_work", None) - return favorites - - -class _NavigationParams: - """Deep Params interface for this panel, matching Galaxy's existing storage.""" - - def __init__(self, params, params_memory=None): - self.params = params - self.params_memory = params_memory or params - - @staticmethod - def _get(params, key: str, default: Any = "") -> Any: - try: - return params.get(key, encoding="utf-8", default=default) - except TypeError: - return params.get(key) - - def active_destination(self) -> dict[str, Any] | None: - return parse_destination_json(self._get(self.params, NAVIGATION_DESTINATION_KEY)) - - def recent_destinations(self) -> list[dict[str, Any]]: - raw_value = self._get(self.params, RECENT_DESTINATIONS_KEY, "[]") - if isinstance(raw_value, (list, dict)): - raw_value = json.dumps(raw_value) - return load_recent_destinations(raw_value) - - def favorite_destinations(self) -> list[dict[str, Any]]: - raw_value = self._get(self.params, FAVORITE_DESTINATIONS_KEY, "[]") - favorites = _load_favorite_destinations(raw_value) - raw_payload = _json_value(raw_value, []) - if isinstance(raw_payload, list) and any( - isinstance(entry, dict) and not entry.get("id") for entry in raw_payload - ): - self.params.put(FAVORITE_DESTINATIONS_KEY, favorites) - return favorites - - def set_destination(self, payload: Any) -> dict[str, Any] | None: - destination = normalize_destination_payload(payload) - if destination is None: - return None - raw_recent_destinations = self._get(self.params, RECENT_DESTINATIONS_KEY, "[]") - if isinstance(raw_recent_destinations, (list, dict)): - raw_recent_destinations = json.dumps(raw_recent_destinations) - recent_destinations = update_recent_destinations( - raw_recent_destinations, - destination, - ) - self.params.put(NAVIGATION_DESTINATION_KEY, json.dumps(destination)) - self.params.put(RECENT_DESTINATIONS_KEY, recent_destinations) - return destination - - def clear_navigation(self) -> bool: - collapsed_supported = True - for params, key in ( - (self.params, NAVIGATION_DESTINATION_KEY), - (self.params_memory, NAV_INSTRUCTION_STATE_KEY), - (self.params_memory, NAV_INSTRUCTION_COLLAPSED_KEY), - ): - try: - params.remove(key) - except Exception: - if key == NAV_INSTRUCTION_COLLAPSED_KEY: - collapsed_supported = False - return collapsed_supported - - def add_favorite(self, favorite: dict[str, Any]) -> list[dict[str, Any]]: - updated = _add_favorite_destination(self._get(self.params, FAVORITE_DESTINATIONS_KEY, "[]"), favorite) - self.params.put(FAVORITE_DESTINATIONS_KEY, updated) - return updated - - def remove_favorite(self, favorite: dict[str, Any]) -> list[dict[str, Any]]: - updated = _remove_favorite_destination(self._get(self.params, FAVORITE_DESTINATIONS_KEY, "[]"), favorite) - self.params.put(FAVORITE_DESTINATIONS_KEY, updated) - return updated - - def update_favorite(self, favorite: dict[str, Any], **changes: Any) -> list[dict[str, Any]] | None: - updated = _update_favorite_destination( - self._get(self.params, FAVORITE_DESTINATIONS_KEY, "[]"), - favorite, - **changes, - ) - if updated is not None: - self.params.put(FAVORITE_DESTINATIONS_KEY, updated) - return updated +_NavigationParams = NavigationDestinationStore +_add_favorite_destination = add_favorite_destination +_favorite_destination_id = favorite_destination_id +_favorite_matches_target = favorite_matches_target +_favorite_payload_for_galaxy = favorite_payload_for_galaxy +_load_favorite_destinations = load_favorite_destinations +_normalize_favorite_payload = normalize_favorite_destination +_remove_favorite_destination = remove_favorite_destination +_update_favorite_destination = update_favorite_destination class MapboxSearchError(RuntimeError): @@ -548,7 +347,7 @@ class StarPilotNavigationLayout(_SettingsPage): return str(self._params.get("MapboxPublicKey", encoding="utf-8") or "").strip() def _routing_available(self) -> bool: - return bool(str(self._params.get("MapboxSecretKey", encoding="utf-8") or "").strip()) + return routing_configured(self._params) def _language_code(self) -> str: language = str(self._params.get("LanguageSetting", encoding="utf-8") or "").strip() @@ -650,15 +449,7 @@ class StarPilotNavigationLayout(_SettingsPage): @staticmethod def _same_destination(left: dict[str, Any] | None, right: dict[str, Any] | None) -> bool: - if not left or not right: - return False - try: - return ( - abs(float(left.get("latitude")) - float(right.get("latitude"))) <= 1e-6 and - abs(float(left.get("longitude")) - float(right.get("longitude"))) <= 1e-6 - ) - except (TypeError, ValueError): - return False + return same_destination(left, right) def _favorite_for_destination(self, destination: dict[str, Any] | None) -> dict[str, Any] | None: return next((favorite for favorite in self._favorites if self._same_destination(favorite, destination)), None) @@ -988,14 +779,7 @@ class StarPilotNavigationLayout(_SettingsPage): style=PANEL_STYLE, ) y += NAV_SECTION_HEIGHT - ordered_favorites = sorted( - self._favorites, - key=lambda item: ( - not bool(item.get("is_home")), - not bool(item.get("is_work")), - str(item.get("name") or "").casefold(), - ), - ) + ordered_favorites = ordered_favorite_destinations(self._favorites) for index, favorite in enumerate(ordered_favorites): row_rect = rl.Rectangle(x, y, width, NAV_ROW_HEIGHT) target_id = f"favorite:{favorite.get('id') or index}" diff --git a/selfdrive/ui/tests/test_home_info_card.py b/selfdrive/ui/tests/test_home_info_card.py new file mode 100644 index 000000000..767a1d12c --- /dev/null +++ b/selfdrive/ui/tests/test_home_info_card.py @@ -0,0 +1,129 @@ +import json + +import pyray as rl + +from openpilot.selfdrive.ui.widgets.home_info_card import HomeInfoCard +from openpilot.starpilot.navigation.destination_store import ( + FAVORITE_DESTINATIONS_KEY, + NAVIGATION_DESTINATION_KEY, + RECENT_DESTINATIONS_KEY, + same_destination, +) + + +class FakeParams: + def __init__(self, values=None): + self.values = dict(values or {}) + self.writes = [] + + def get(self, key, encoding=None, default=None): + value = self.values.get(key, default) + if encoding == "utf-8" and isinstance(value, bytes): + return value.decode("utf-8") + return value + + def put(self, key, value): + self.values[key] = value + self.writes.append((key, value)) + + +class FakeDriveStats: + def __init__(self): + self.records_rendered = 0 + + def render_records(self, _rect): + self.records_rendered += 1 + + +def _favorite(name, latitude, longitude, **flags): + return {"name": name, "latitude": latitude, "longitude": longitude, **flags} + + +def _card(params): + card = HomeInfoCard(params, FakeDriveStats()) + card.set_rect(rl.Rectangle(100, 200, 750, 745)) + card.refresh() + return card + + +def test_quick_start_requires_secret_and_at_least_one_valid_favorite(): + no_secret = _card(FakeParams({ + FAVORITE_DESTINATIONS_KEY: json.dumps([_favorite("Home", 1, 2)]), + })) + no_valid_favorites = _card(FakeParams({ + "MapboxSecretKey": "secret", + FAVORITE_DESTINATIONS_KEY: json.dumps([_favorite("Overflow", 10 ** 1000, 2)]), + })) + one_favorite = _card(FakeParams({ + "MapboxSecretKey": "secret", + FAVORITE_DESTINATIONS_KEY: json.dumps([_favorite("Home", 1, 2)]), + })) + + assert not no_secret.quick_start_available + assert no_secret.show_records + assert not no_valid_favorites.quick_start_available + assert no_valid_favorites.show_records + assert one_favorite.quick_start_available + assert not one_favorite.show_records + + +def test_quick_start_uses_canonical_order_and_three_row_limit(): + params = FakeParams({ + "MapboxSecretKey": "secret", + FAVORITE_DESTINATIONS_KEY: json.dumps([ + _favorite("Zulu", 1, 1), + _favorite("Work", 2, 2, is_work=True), + _favorite("Bravo", 3, 3), + _favorite("Home", 4, 4, is_home=True), + _favorite("Alpha", 5, 5), + ]), + }) + + card = _card(params) + + assert [favorite["name"] for favorite in card.favorites] == ["Home", "Work", "Alpha"] + + +def test_page_switch_is_in_memory_and_destination_rows_select_once(): + params = FakeParams({ + "MapboxSecretKey": "secret", + FAVORITE_DESTINATIONS_KEY: json.dumps([_favorite("Home", 1, 2)]), + RECENT_DESTINATIONS_KEY: [], + }) + card = _card(params) + + flip_center = rl.Vector2(card._flip_rect.x + card._flip_rect.width / 2, card._flip_rect.y + card._flip_rect.height / 2) + card._handle_mouse_release(flip_center) + assert card.show_records + card._handle_mouse_release(flip_center) + assert not card.show_records + + row = card._destination_rects[0] + row_center = rl.Vector2(row.x + row.width / 2, row.y + row.height / 2) + card._handle_mouse_release(row_center) + assert json.loads(params.values[NAVIGATION_DESTINATION_KEY])["name"] == "Home" + assert params.values[RECENT_DESTINATIONS_KEY][0]["place_name"] == "Home" + first_write_count = len(params.writes) + assert card.active_destination["name"] == "Home" + + card._handle_mouse_release(row_center) + assert len(params.writes) == first_write_count + + +def test_refresh_falls_back_without_clearing_an_active_non_favorite_destination(): + active = {"name": "Old destination", "latitude": 9, "longitude": 10} + params = FakeParams({ + "MapboxSecretKey": "secret", + FAVORITE_DESTINATIONS_KEY: json.dumps([_favorite("Home", 1, 2)]), + NAVIGATION_DESTINATION_KEY: json.dumps(active), + }) + card = _card(params) + + assert card.active_destination["name"] == "Old destination" + assert all(not same_destination(card.active_destination, favorite) for favorite in card.favorites) + + params.values["MapboxSecretKey"] = "" + card.refresh() + assert not card.quick_start_available + assert card.show_records + assert card.active_destination["name"] == "Old destination" diff --git a/selfdrive/ui/widgets/home_info_card.py b/selfdrive/ui/widgets/home_info_card.py new file mode 100644 index 000000000..67498fce8 --- /dev/null +++ b/selfdrive/ui/widgets/home_info_card.py @@ -0,0 +1,312 @@ +from __future__ import annotations + +from typing import Any + +import pyray as rl + +from openpilot.selfdrive.ui.widgets.drive_stats import ( + CARD_BORDER, + CARD_COLOR, + MUTED_COLOR, + TEXT_COLOR, + TRACK_COLOR, + TEAL, +) +from openpilot.starpilot.navigation.destination_store import ( + FAVORITE_DESTINATIONS_KEY, + NavigationDestinationStore, + load_favorite_destinations, + ordered_favorite_destinations, + routing_configured, + same_destination, +) +from openpilot.system.ui.lib.application import FontWeight, MousePos, gui_app +from openpilot.system.ui.lib.multilang import tr +from openpilot.system.ui.lib.text_measure import measure_text_cached +from openpilot.system.ui.widgets import Widget + + +HEADER_HEIGHT = 82.0 +ROW_COUNT = 3 +PAGE_SWITCH_SIZE = 72.0 +TEXT_LEFT_INSET = 30.0 +TEXT_RIGHT_INSET = 34.0 + + +class HomeInfoCard(Widget): + """Paired offroad Home card for Quick Start and Personal Records.""" + + def __init__(self, params: Any, drive_stats: Any): + super().__init__() + self._params = params + self._store = NavigationDestinationStore(params) + self._drive_stats = drive_stats + + self._show_records = False + self._quick_start_available = False + self._favorites: list[dict[str, Any]] = [] + self._active_destination: dict[str, Any] | None = None + self._flip_rect = rl.Rectangle(0, 0, 0, 0) + self._destination_rects: list[rl.Rectangle] = [] + + self._font_semi_bold: rl.Font | None = None + self._font_medium: rl.Font | None = None + + @property + def quick_start_available(self) -> bool: + return self._quick_start_available + + @property + def show_records(self) -> bool: + return self._show_records + + @property + def favorites(self) -> list[dict[str, Any]]: + return list(self._favorites) + + @property + def active_destination(self) -> dict[str, Any] | None: + return self._active_destination + + def show_event(self): + self._show_records = False + self.refresh() + super().show_event() + + def refresh(self) -> None: + was_available = self._quick_start_available + self._active_destination = self._store.active_destination() + + raw_favorites = self._params.get(FAVORITE_DESTINATIONS_KEY, encoding="utf-8", default="[]") + favorites = load_favorite_destinations(raw_favorites) + self._favorites = ordered_favorite_destinations(favorites, limit=ROW_COUNT) + self._quick_start_available = routing_configured(self._params) and bool(self._favorites) + + if not self._quick_start_available: + self._show_records = True + elif not was_available: + self._show_records = False + + def _update_layout_rects(self) -> None: + self._flip_rect = rl.Rectangle( + self._rect.x + max(0.0, self._rect.width - PAGE_SWITCH_SIZE - 12.0), + self._rect.y + 5.0, + PAGE_SWITCH_SIZE, + PAGE_SWITCH_SIZE, + ) + + row_height = max(0.0, (self._rect.height - HEADER_HEIGHT - 12.0) / ROW_COUNT) + self._destination_rects = [ + rl.Rectangle( + self._rect.x, + self._rect.y + HEADER_HEIGHT + index * row_height, + self._rect.width, + row_height, + ) + for index in range(ROW_COUNT) + ] + + def _handle_mouse_release(self, mouse_pos: MousePos) -> None: + super()._handle_mouse_release(mouse_pos) + + if not self._quick_start_available: + return + + if rl.check_collision_point_rec(mouse_pos, self._flip_rect): + self._show_records = not self._show_records + return + + if self._show_records: + return + + for index, row_rect in enumerate(self._destination_rects): + if rl.check_collision_point_rec(mouse_pos, row_rect): + self._select_destination(index) + return + + def _select_destination(self, index: int) -> None: + if index < 0 or index >= len(self._favorites): + return + + favorite = self._favorites[index] + destination = self._store.set_destination(favorite, skip_if_same=True) + if destination is not None: + self._active_destination = destination + + def _font(self, weight: FontWeight) -> rl.Font: + if weight == FontWeight.SEMI_BOLD: + if self._font_semi_bold is None: + self._font_semi_bold = gui_app.font(weight) + return self._font_semi_bold + if self._font_medium is None: + self._font_medium = gui_app.font(weight) + return self._font_medium + + @staticmethod + def _fit_text(font: Any, text: str, font_size: int, max_width: float) -> str: + if max_width <= 0: + return "" + if measure_text_cached(font, text, font_size).x <= max_width: + return text + + ellipsis = "…" + fitted = text + while fitted and measure_text_cached(font, f"{fitted}{ellipsis}", font_size).x > max_width: + fitted = fitted[:-1] + return f"{fitted}{ellipsis}" if fitted else ellipsis + + @staticmethod + def _draw_card(rect: rl.Rectangle) -> None: + rl.draw_rectangle_rounded(rect, 0.04, 12, CARD_COLOR) + rl.draw_rectangle_rounded_lines_ex(rect, 0.04, 12, 2, CARD_BORDER) + + @staticmethod + def _draw_checkmark(center_x: float, center_y: float, scale: float = 1.0) -> None: + color = TEAL + rl.draw_line_ex( + rl.Vector2(center_x - 14 * scale, center_y), + rl.Vector2(center_x - 3 * scale, center_y + 11 * scale), + 4 * scale, + color, + ) + rl.draw_line_ex( + rl.Vector2(center_x - 3 * scale, center_y + 11 * scale), + rl.Vector2(center_x + 16 * scale, center_y - 12 * scale), + 4 * scale, + color, + ) + + @staticmethod + def _draw_page_icon(rect: rl.Rectangle) -> None: + center_x = rect.x + rect.width / 2 + center_y = rect.y + rect.height / 2 + color = MUTED_COLOR + thickness = 3.0 + + rl.draw_line_ex( + rl.Vector2(center_x - 16, center_y - 7), + rl.Vector2(center_x + 14, center_y - 7), + thickness, + color, + ) + rl.draw_line_ex( + rl.Vector2(center_x + 14, center_y - 7), + rl.Vector2(center_x + 6, center_y - 14), + thickness, + color, + ) + rl.draw_line_ex( + rl.Vector2(center_x + 14, center_y - 7), + rl.Vector2(center_x + 6, center_y), + thickness, + color, + ) + rl.draw_line_ex( + rl.Vector2(center_x + 16, center_y + 8), + rl.Vector2(center_x - 14, center_y + 8), + thickness, + color, + ) + rl.draw_line_ex( + rl.Vector2(center_x - 14, center_y + 8), + rl.Vector2(center_x - 6, center_y + 1), + thickness, + color, + ) + rl.draw_line_ex( + rl.Vector2(center_x - 14, center_y + 8), + rl.Vector2(center_x - 6, center_y + 15), + thickness, + color, + ) + + def _draw_header(self, rect: rl.Rectangle, title: str) -> None: + rl.draw_text_ex( + self._font(FontWeight.SEMI_BOLD), + title, + rl.Vector2(rect.x + TEXT_LEFT_INSET, rect.y + 26), + 32, + 0, + TEXT_COLOR, + ) + self._draw_page_icon(self._flip_rect) + + def _draw_quick_start(self, rect: rl.Rectangle) -> None: + self._draw_card(rect) + self._draw_header(rect, tr("START NAVIGATION")) + + row_height = (rect.height - HEADER_HEIGHT - 12.0) / ROW_COUNT + title_font = self._font(FontWeight.MEDIUM) + for index in range(ROW_COUNT): + row_y = rect.y + HEADER_HEIGHT + index * row_height + if index > 0: + rl.draw_line( + int(rect.x + 24), + int(row_y), + int(rect.x + rect.width - 24), + int(row_y), + TRACK_COLOR, + ) + + favorite = self._favorites[index] if index < len(self._favorites) else None + if favorite is None: + if index == len(self._favorites): + empty_text = tr("Add favorites in Navigation") + empty_size = measure_text_cached(title_font, empty_text, 28) + rl.draw_text_ex( + title_font, + empty_text, + rl.Vector2(rect.x + (rect.width - empty_size.x) / 2, row_y + (row_height - empty_size.y) / 2), + 28, + 0, + MUTED_COLOR, + ) + continue + + selected = same_destination(self._active_destination, favorite) + name = str(favorite.get("name") or tr("Favorite destination")) + name_width = rect.width - TEXT_LEFT_INSET - TEXT_RIGHT_INSET - (74 if selected else 0) + name = self._fit_text(title_font, name, 38, name_width) + name_size = measure_text_cached(title_font, name, 38) + text_x = rect.x + TEXT_LEFT_INSET + center_y = row_y + row_height / 2 + rl.draw_text_ex( + title_font, + name, + rl.Vector2(text_x, center_y - name_size.y - 5), + 38, + 0, + TEXT_COLOR, + ) + + if selected: + subtitle = tr("Selected for next drive") + elif favorite.get("is_home"): + subtitle = tr("Home") + elif favorite.get("is_work"): + subtitle = tr("Work") + else: + subtitle = tr("Favorite destination") + subtitle = self._fit_text(title_font, subtitle, 25, name_width) + rl.draw_text_ex( + title_font, + subtitle, + rl.Vector2(text_x, center_y + 17), + 25, + 0, + MUTED_COLOR if not selected else TEAL, + ) + + if selected: + self._draw_checkmark(rect.x + rect.width - 60, center_y, 0.9) + + def _render(self, rect: rl.Rectangle): + if not self._quick_start_available: + self._drive_stats.render_records(rect) + return + + if self._show_records: + self._drive_stats.render_records(rect) + self._draw_page_icon(self._flip_rect) + else: + self._draw_quick_start(rect) diff --git a/starpilot/navigation/destination_store.py b/starpilot/navigation/destination_store.py index 8393d28a1..0a3b9dc1b 100644 --- a/starpilot/navigation/destination_store.py +++ b/starpilot/navigation/destination_store.py @@ -1,16 +1,54 @@ from __future__ import annotations import json +import hashlib +import math from typing import Any +NAVIGATION_DESTINATION_KEY = "NavDestination" +RECENT_DESTINATIONS_KEY = "ApiCache_NavDestinations" +FAVORITE_DESTINATIONS_KEY = "FavoriteDestinations" +NAV_INSTRUCTION_STATE_KEY = "NavInstructionState" +NAV_INSTRUCTION_COLLAPSED_KEY = "NavInstructionCollapsed" + RECENT_DESTINATIONS_LIMIT = 10 def _coerce_float(value: Any) -> float | None: try: - return float(value) - except (TypeError, ValueError): + parsed = float(value) + except (OverflowError, TypeError, ValueError): return None + return parsed if math.isfinite(parsed) else None + + +def _json_value(raw_value: Any, default: Any) -> Any: + if isinstance(raw_value, (list, dict)): + return raw_value + if isinstance(raw_value, bytes): + raw_value = raw_value.decode("utf-8", errors="replace") + if not raw_value: + return default + try: + return json.loads(raw_value) + except (TypeError, ValueError): + return default + + +def _param_get(params: Any, key: str, default: Any = "") -> Any: + try: + return params.get(key, encoding="utf-8", default=default) + except TypeError: + try: + return params.get(key, default=default) + except TypeError: + return params.get(key) + + +def _text(value: Any) -> str: + if isinstance(value, bytes): + return value.decode("utf-8", errors="replace") + return str(value or "") def normalize_destination_payload(payload: Any) -> dict[str, Any] | None: @@ -32,18 +70,200 @@ def normalize_destination_payload(payload: Any) -> dict[str, Any] | None: } -def parse_destination_json(raw_value: str | bytes | None) -> dict[str, Any] | None: +def parse_destination_json(raw_value: str | bytes | dict[str, Any] | None) -> dict[str, Any] | None: if not raw_value: return None - try: - payload = json.loads(raw_value) - except (TypeError, ValueError, json.JSONDecodeError): + payload = _json_value(raw_value, None) + if payload is None: return None return normalize_destination_payload(payload) +def _favorite_destination_id(payload: dict[str, Any]) -> str: + """Keep the exact favorite ID formula used by Galaxy's existing backend.""" + raw = f"{payload.get('longitude')},{payload.get('latitude')}|{payload.get('routeId') or ''}|{payload.get('name') or ''}" + return hashlib.sha1(raw.encode()).hexdigest() + + +def favorite_destination_id(payload: dict[str, Any]) -> str: + return _favorite_destination_id(payload) + + +def favorite_payload_for_galaxy(destination: dict[str, Any]) -> dict[str, Any]: + favorite = dict(destination) + favorite.setdefault("routeId", "main") + return favorite + + +def normalize_favorite_destination(payload: Any) -> dict[str, Any] | None: + if not isinstance(payload, dict): + return None + + name = str(payload.get("name") or payload.get("place_name") or "").strip() + latitude = _coerce_float(payload.get("latitude")) + longitude = _coerce_float(payload.get("longitude")) + if not name or latitude is None or longitude is None: + return None + + normalized = dict(payload) + normalized.update({"name": name, "latitude": latitude, "longitude": longitude}) + normalized["id"] = str(payload.get("id") or _favorite_destination_id(payload)) + return normalized + + +def load_favorite_destinations(raw_value: Any) -> list[dict[str, Any]]: + payload = _json_value(raw_value, []) + if not isinstance(payload, list): + return [] + return [ + normalized + for entry in payload + if (normalized := normalize_favorite_destination(entry)) is not None + ] + + +def ordered_favorite_destinations(favorites: list[dict[str, Any]], limit: int | None = None) -> list[dict[str, Any]]: + normalized = [ + favorite + for entry in favorites + if (favorite := normalize_favorite_destination(entry)) is not None + ] + + def order_key(favorite: dict[str, Any]) -> tuple[bool, bool, str]: + return ( + not bool(favorite.get("is_home")), + not bool(favorite.get("is_work")), + str(favorite.get("name") or "").casefold(), + ) + + ordered = sorted(normalized, key=order_key) + return ordered if limit is None else ordered[:max(0, limit)] + + +def same_destination(left: dict[str, Any] | None, right: dict[str, Any] | None) -> bool: + if not left or not right: + return False + try: + left_latitude = _coerce_float(left.get("latitude")) + right_latitude = _coerce_float(right.get("latitude")) + left_longitude = _coerce_float(left.get("longitude")) + right_longitude = _coerce_float(right.get("longitude")) + if ( + left_latitude is None or right_latitude is None or + left_longitude is None or right_longitude is None + ): + return False + return bool( + abs(left_latitude - right_latitude) <= 1e-6 and + abs(left_longitude - right_longitude) <= 1e-6 + ) + except (TypeError, ValueError): + return False + + +def routing_configured(params: Any) -> bool: + return bool(_text(_param_get(params, "MapboxSecretKey", "")).strip()) + + +def set_navigation_destination(params: Any, payload: Any, *, skip_if_same: bool = False) -> dict[str, Any] | None: + destination = normalize_destination_payload(payload) + if destination is None: + return None + + if skip_if_same: + current = parse_destination_json(_param_get(params, NAVIGATION_DESTINATION_KEY, "")) + if same_destination(current, destination): + return destination + + raw_recent_destinations = _param_get(params, RECENT_DESTINATIONS_KEY, "[]") + if isinstance(raw_recent_destinations, (list, dict)): + raw_recent_destinations = json.dumps(raw_recent_destinations) + recent_destinations = update_recent_destinations(raw_recent_destinations, destination) + params.put(NAVIGATION_DESTINATION_KEY, json.dumps(destination)) + params.put(RECENT_DESTINATIONS_KEY, recent_destinations) + return destination + + +def favorite_matches_target( + favorite: dict[str, Any], + target: dict[str, Any], + *, + allow_route_id_only: bool = False, +) -> bool: + target_id = target.get("id") + if target_id: + return bool(favorite.get("id") == target_id) + if allow_route_id_only and target.get("routeId"): + return bool(favorite.get("routeId") == target.get("routeId")) + return bool( + favorite.get("routeId") == target.get("routeId") and + favorite.get("latitude") == target.get("latitude") and + favorite.get("longitude") == target.get("longitude") and + favorite.get("name") == target.get("name") + ) + + +def add_favorite_destination(raw_value: Any, favorite: dict[str, Any]) -> list[dict[str, Any]]: + favorites = load_favorite_destinations(raw_value) + normalized = normalize_favorite_destination(favorite) + if normalized is not None and not any(item.get("id") == normalized["id"] for item in favorites): + favorites.append(normalized) + return favorites + + +def remove_favorite_destination(raw_value: Any, target: dict[str, Any]) -> list[dict[str, Any]]: + favorites = load_favorite_destinations(raw_value) + return [favorite for favorite in favorites if not favorite_matches_target(favorite, target)] + + +def update_favorite_destination( + raw_value: Any, + target: dict[str, Any], + *, + name: str | None = None, + is_home: bool | None = None, + is_work: bool | None = None, +) -> list[dict[str, Any]] | None: + favorites = load_favorite_destinations(raw_value) + target_index = next( + ( + index for index, favorite in enumerate(favorites) + if favorite_matches_target(favorite, target, allow_route_id_only=True) + ), + None, + ) + if target_index is None: + return None + + if is_home: + for favorite in favorites: + favorite.pop("is_home", None) + if is_work: + for favorite in favorites: + favorite.pop("is_work", None) + + favorite = favorites[target_index] + if name is not None: + normalized_name = str(name).strip() + if normalized_name: + favorite["name"] = normalized_name + if is_home is not None: + if is_home: + favorite["is_home"] = True + favorite.pop("is_work", None) + else: + favorite.pop("is_home", None) + if is_work is not None: + if is_work: + favorite["is_work"] = True + favorite.pop("is_home", None) + else: + favorite.pop("is_work", None) + return favorites + + def normalize_recent_destination_entry(entry: Any) -> dict[str, Any] | None: if not isinstance(entry, dict): return None @@ -102,3 +322,70 @@ def update_recent_destinations(raw_value: str | bytes | None, destination: dict[ break return [entry for entry in updated if entry is not None] + + +class NavigationDestinationStore: + """Shared local storage boundary for navigation destinations and favorites.""" + + def __init__(self, params: Any, params_memory: Any | None = None): + self.params = params + self.params_memory = params_memory or params + + def active_destination(self) -> dict[str, Any] | None: + return parse_destination_json(_param_get(self.params, NAVIGATION_DESTINATION_KEY, "")) + + def recent_destinations(self) -> list[dict[str, Any]]: + raw_value = _param_get(self.params, RECENT_DESTINATIONS_KEY, "[]") + if isinstance(raw_value, (list, dict)): + raw_value = json.dumps(raw_value) + return load_recent_destinations(raw_value) + + def favorite_destinations(self) -> list[dict[str, Any]]: + raw_value = _param_get(self.params, FAVORITE_DESTINATIONS_KEY, "[]") + favorites = load_favorite_destinations(raw_value) + raw_payload = _json_value(raw_value, []) + if isinstance(raw_payload, list) and any( + isinstance(entry, dict) and not entry.get("id") for entry in raw_payload + ): + self.params.put(FAVORITE_DESTINATIONS_KEY, favorites) + return favorites + + def set_destination(self, payload: Any, *, skip_if_same: bool = False) -> dict[str, Any] | None: + return set_navigation_destination(self.params, payload, skip_if_same=skip_if_same) + + def clear_navigation(self) -> bool: + collapsed_supported = True + for params, key in ( + (self.params, NAVIGATION_DESTINATION_KEY), + (self.params_memory, NAV_INSTRUCTION_STATE_KEY), + (self.params_memory, NAV_INSTRUCTION_COLLAPSED_KEY), + ): + try: + params.remove(key) + except Exception: + if key == NAV_INSTRUCTION_COLLAPSED_KEY: + collapsed_supported = False + return collapsed_supported + + def add_favorite(self, favorite: dict[str, Any]) -> list[dict[str, Any]]: + updated = add_favorite_destination(_param_get(self.params, FAVORITE_DESTINATIONS_KEY, "[]"), favorite) + self.params.put(FAVORITE_DESTINATIONS_KEY, updated) + return updated + + def remove_favorite(self, favorite: dict[str, Any]) -> list[dict[str, Any]]: + updated = remove_favorite_destination(_param_get(self.params, FAVORITE_DESTINATIONS_KEY, "[]"), favorite) + self.params.put(FAVORITE_DESTINATIONS_KEY, updated) + return updated + + def update_favorite(self, favorite: dict[str, Any], **changes: Any) -> list[dict[str, Any]] | None: + updated = update_favorite_destination( + _param_get(self.params, FAVORITE_DESTINATIONS_KEY, "[]"), + favorite, + **changes, + ) + if updated is not None: + self.params.put(FAVORITE_DESTINATIONS_KEY, updated) + return updated + + def routing_configured(self) -> bool: + return routing_configured(self.params) diff --git a/starpilot/navigation/test_destination_store.py b/starpilot/navigation/test_destination_store.py index 9841aee6e..d358e9f89 100644 --- a/starpilot/navigation/test_destination_store.py +++ b/starpilot/navigation/test_destination_store.py @@ -1,7 +1,18 @@ import json from openpilot.starpilot.navigation.destination_store import ( + FAVORITE_DESTINATIONS_KEY, + NAVIGATION_DESTINATION_KEY, + RECENT_DESTINATIONS_KEY, + NavigationDestinationStore, + favorite_destination_id, + load_favorite_destinations, normalize_destination_payload, + normalize_favorite_destination, + ordered_favorite_destinations, + routing_configured, + same_destination, + set_navigation_destination, update_recent_destinations, ) @@ -61,3 +72,127 @@ def test_recent_destinations_retain_coordinates_for_saved_name(): "latitude": 41.881832, "longitude": -87.623177, } + + +def test_favorite_normalization_rejects_malformed_and_non_finite_values(): + assert normalize_favorite_destination({"name": "Missing coordinates"}) is None + assert normalize_favorite_destination({"name": "Bad", "latitude": "nan", "longitude": 2}) is None + assert normalize_favorite_destination({"name": "Bad", "latitude": 1, "longitude": "inf"}) is None + + normalized = normalize_favorite_destination({"place_name": "Home", "latitude": "1", "longitude": "2"}) + assert normalized is not None + assert normalized["name"] == "Home" + assert normalized["latitude"] == 1.0 + assert normalized["longitude"] == 2.0 + + +def test_load_favorites_is_safe_for_malformed_json_and_filters_invalid_entries(): + raw = json.dumps([ + {"name": "Valid", "latitude": 1, "longitude": 2}, + {"name": "Missing longitude", "latitude": 1}, + {"name": "Overflow", "latitude": 10 ** 1000, "longitude": 2}, + "not a favorite", + ]) + + assert [favorite["name"] for favorite in load_favorite_destinations(raw)] == ["Valid"] + assert load_favorite_destinations("not json") == [] + assert load_favorite_destinations(json.dumps({"name": "not a list"})) == [] + + +def test_ordered_favorites_put_home_then_work_then_remaining_alphabetically(): + favorites = [ + {"name": "zulu", "latitude": 1, "longitude": 1}, + {"name": "Work", "latitude": 2, "longitude": 2, "is_work": True}, + {"name": "bravo", "latitude": 3, "longitude": 3}, + {"name": "Home", "latitude": 4, "longitude": 4, "is_home": True}, + {"name": "alpha", "latitude": 5, "longitude": 5}, + ] + + ordered = ordered_favorite_destinations(favorites) + + assert [favorite["name"] for favorite in ordered] == ["Home", "Work", "alpha", "bravo", "zulu"] + assert [favorite["name"] for favorite in ordered_favorite_destinations(favorites, limit=3)] == ["Home", "Work", "alpha"] + + +def test_destination_equality_uses_coordinate_tolerance(): + left = {"name": "A", "latitude": 1.0, "longitude": 2.0} + almost_same = {"name": "B", "latitude": 1.0000005, "longitude": 1.9999995} + different = {"name": "A", "latitude": 1.01, "longitude": 2.0} + + assert same_destination(left, almost_same) + assert not same_destination(left, different) + assert not same_destination(left, None) + + +class FakeParams: + def __init__(self, values=None): + self.values = dict(values or {}) + self.writes = [] + self.removed = [] + + def get(self, key, encoding=None, default=None): + value = self.values.get(key, default) + if encoding == "utf-8" and isinstance(value, bytes): + return value.decode("utf-8") + return value + + def put(self, key, value): + self.values[key] = value + self.writes.append((key, value)) + + def remove(self, key): + self.values.pop(key, None) + self.removed.append(key) + + +def test_destination_write_updates_active_destination_and_recents(): + params = FakeParams({RECENT_DESTINATIONS_KEY: "[]"}) + + destination = set_navigation_destination(params, {"name": "Home", "latitude": 1, "longitude": 2}) + + assert destination == {"name": "Home", "place_name": "Home", "latitude": 1.0, "longitude": 2.0} + assert json.loads(params.values[NAVIGATION_DESTINATION_KEY]) == destination + assert params.values[RECENT_DESTINATIONS_KEY][0]["place_name"] == "Home" + + +def test_same_destination_write_is_idempotent_and_does_not_touch_recents(): + params = FakeParams({ + NAVIGATION_DESTINATION_KEY: json.dumps({"name": "Home", "latitude": 1, "longitude": 2}), + RECENT_DESTINATIONS_KEY: [{"place_name": "Existing"}], + }) + + result = set_navigation_destination( + params, + {"name": "Home", "latitude": 1.0000005, "longitude": 2}, + skip_if_same=True, + ) + + assert result is not None + assert params.writes == [] + assert params.values[RECENT_DESTINATIONS_KEY] == [{"place_name": "Existing"}] + + settings_params = FakeParams(dict(params.values)) + set_navigation_destination(settings_params, {"name": "Home", "latitude": 1.0, "longitude": 2.0}) + assert [key for key, _value in settings_params.writes] == [NAVIGATION_DESTINATION_KEY, RECENT_DESTINATIONS_KEY] + + +def test_navigation_destination_store_keeps_settings_favorite_migration_and_mutations(): + params = FakeParams({ + FAVORITE_DESTINATIONS_KEY: json.dumps([{"name": "Home", "latitude": 1, "longitude": 2}]), + }) + store = NavigationDestinationStore(params) + + favorites = store.favorite_destinations() + assert favorites[0]["id"] == favorite_destination_id({"name": "Home", "latitude": 1, "longitude": 2}) + assert params.writes[0][0] == FAVORITE_DESTINATIONS_KEY + + added = store.add_favorite({"name": "Work", "latitude": 3, "longitude": 4}) + assert [favorite["name"] for favorite in added] == ["Home", "Work"] + assert store.update_favorite(added[1], is_work=True)[1]["is_work"] is True + assert [favorite["name"] for favorite in store.remove_favorite(added[0])] == ["Work"] + + +def test_routing_configured_only_requires_a_non_empty_secret_key(): + assert not routing_configured(FakeParams()) + assert not routing_configured(FakeParams({"MapboxSecretKey": " "})) + assert routing_configured(FakeParams({"MapboxSecretKey": "secret"})) diff --git a/tools/replay/onroad_config.py b/tools/replay/onroad_config.py index cc250198c..b1599286f 100644 --- a/tools/replay/onroad_config.py +++ b/tools/replay/onroad_config.py @@ -4,17 +4,26 @@ from __future__ import annotations import os import re import sys +from collections.abc import Sequence from dataclasses import dataclass from pathlib import Path -from typing import Any, Sequence +from typing import Any from openpilot.common.params import Params, UnknownKeyName from openpilot.system.version import terms_version, training_version from openpilot.tools.lib.logreader import LogReader, ReadMode, parse_direct, parse_indirect from openpilot.tools.lib.route import SegmentRange +from openpilot.starpilot.navigation.destination_store import FAVORITE_DESTINATIONS_KEY, load_favorite_destinations DEMO_ROUTE = "a2a0ccea32023010|2023-07-27--13-01-19" +NAV_DEMO_MAPBOX_SECRET = "desktop-nav-demo-placeholder" +NAV_DEMO_FAVORITES = [{ + "name": "Demo Home", + "latitude": 0.01, + "longitude": 0.02, + "is_home": True, +}] _VALUE_OPTIONS = { "-a", "--allow", @@ -120,7 +129,7 @@ def first_segment_identifier(route: str) -> str: parsed = parse_indirect(route) direct = parse_direct(parsed) if direct is not None: - return direct + return str(direct) sr = SegmentRange(parsed) selector = sr.selector or "a" @@ -238,6 +247,16 @@ def seed_logged_params(init_data: Any | None, params: Params) -> int: return seeded +def seed_nav_offroad_preview(params: Params) -> None: + secret = params.get("MapboxSecretKey", encoding="utf-8") or "" + if not str(secret).strip(): + params.put("MapboxSecretKey", NAV_DEMO_MAPBOX_SECRET) + + raw_favorites = params.get(FAVORITE_DESTINATIONS_KEY, encoding="utf-8") + if not load_favorite_destinations(raw_favorites): + params.put(FAVORITE_DESTINATIONS_KEY, NAV_DEMO_FAVORITES) + + def seed_desktop_overrides(params: Params) -> None: params.put("HasAcceptedTerms", terms_version) params.put("CompletedTrainingVersion", training_version) @@ -247,6 +266,9 @@ def seed_desktop_overrides(params: Params) -> None: params.put_bool("ForceOffroad", False) if _truthy_env("SP_ONROAD_NAV_DEMO"): params.put_bool("NavigationUI", True) + if _truthy_env("SP_ONROAD_OFFROAD_DEMO"): + params.put_bool("ForceOffroad", True) + seed_nav_offroad_preview(params) def seed_onroad_params(init_data: Any | None, params: Params | None = None) -> int: diff --git a/tools/replay/tests/test_onroad_config.py b/tools/replay/tests/test_onroad_config.py index 60f0b15e7..aa09acddb 100644 --- a/tools/replay/tests/test_onroad_config.py +++ b/tools/replay/tests/test_onroad_config.py @@ -1,5 +1,6 @@ from __future__ import annotations +import json from types import SimpleNamespace from openpilot.tools.replay import onroad_config @@ -9,8 +10,14 @@ TEST_ROUTE = "344c5c15b34f2d8a/2024-01-03--09-37-12" class FakeParams: - def __init__(self): - self.values = {} + def __init__(self, values=None): + self.values = dict(values or {}) + + def get(self, key, encoding=None, default=None): + value = self.values.get(key, default) + if encoding == "utf-8" and isinstance(value, bytes): + return value.decode("utf-8") + return value def cpp2python(self, key, value): if key == "ShowSLCOffset": @@ -91,3 +98,34 @@ def test_seed_onroad_params_uses_logged_disabled_bool_and_desktop_overrides(monk assert "AccessToken" not in params.values assert params.values["OpenpilotEnabledToggle"] is True assert params.values["NavigationUI"] is True + assert params.values["ForceOffroad"] is False + assert "MapboxSecretKey" not in params.values + assert "FavoriteDestinations" not in params.values + + +def test_seed_nav_offroad_preview_provides_quick_start_requirements(monkeypatch): + monkeypatch.setenv("SP_ONROAD_NAV_DEMO", "1") + monkeypatch.setenv("SP_ONROAD_OFFROAD_DEMO", "1") + params = FakeParams() + + onroad_config.seed_onroad_params(None, params) + + assert params.values["ForceOffroad"] is True + assert params.values["ForceOnroad"] is False + assert params.values["MapboxSecretKey"] == onroad_config.NAV_DEMO_MAPBOX_SECRET + assert params.values["FavoriteDestinations"] == onroad_config.NAV_DEMO_FAVORITES + + +def test_seed_nav_offroad_preview_preserves_existing_navigation_data(monkeypatch): + monkeypatch.setenv("SP_ONROAD_NAV_DEMO", "1") + monkeypatch.setenv("SP_ONROAD_OFFROAD_DEMO", "1") + favorites = [{"name": "Existing", "latitude": 1.0, "longitude": 2.0}] + params = FakeParams({ + "MapboxSecretKey": "existing-secret", + "FavoriteDestinations": json.dumps(favorites), + }) + + onroad_config.seed_onroad_params(None, params) + + assert params.values["MapboxSecretKey"] == "existing-secret" + assert json.loads(params.values["FavoriteDestinations"]) == favorites