mirror of
https://github.com/firestar5683/StarPilot.git
synced 2026-08-20 07:43:48 +08:00
OnStar Navigation : Always There. Always Ready.
This commit is contained in:
@@ -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] <route-or-replay-args...>
|
||||
./onroad [jobs] [--c3 | --c4 | --raybig | --all | --replay-only] [--galaxy] [-nav] [--offroad] [-alert] [--cem] [--prefix name] <route-or-replay-args...>
|
||||
|
||||
Examples:
|
||||
./onroad <route>
|
||||
./onroad --c3 <route>
|
||||
./onroad --c4 <route> --start 30
|
||||
./onroad --c4 -nav <route>
|
||||
./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
|
||||
|
||||
|
||||
@@ -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()
|
||||
|
||||
@@ -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}"
|
||||
|
||||
@@ -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"
|
||||
@@ -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)
|
||||
@@ -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)
|
||||
|
||||
@@ -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"}))
|
||||
|
||||
@@ -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:
|
||||
|
||||
@@ -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
|
||||
|
||||
Reference in New Issue
Block a user