mirror of
https://github.com/firestar5683/StarPilot.git
synced 2026-08-20 07:43:48 +08:00
A Quickie
This commit is contained in:
@@ -7,7 +7,6 @@ from openpilot.starpilot.navigation.destination_store import (
|
||||
FAVORITE_DESTINATIONS_KEY,
|
||||
NAVIGATION_DESTINATION_KEY,
|
||||
RECENT_DESTINATIONS_KEY,
|
||||
START_ON_NEXT_DRIVE_KEY,
|
||||
same_destination,
|
||||
)
|
||||
|
||||
@@ -102,9 +101,7 @@ def test_page_switch_is_in_memory_and_destination_rows_select_once():
|
||||
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)
|
||||
stored_destination = json.loads(params.values[NAVIGATION_DESTINATION_KEY])
|
||||
assert stored_destination["name"] == "Home"
|
||||
assert stored_destination[START_ON_NEXT_DRIVE_KEY] is True
|
||||
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"
|
||||
|
||||
@@ -129,7 +129,7 @@ class HomeInfoCard(Widget):
|
||||
return
|
||||
|
||||
favorite = self._favorites[index]
|
||||
destination = self._store.set_destination(favorite, skip_if_same=True, start_on_next_drive=True)
|
||||
destination = self._store.set_destination(favorite, skip_if_same=True)
|
||||
if destination is not None:
|
||||
self._active_destination = destination
|
||||
|
||||
|
||||
@@ -10,8 +10,6 @@ RECENT_DESTINATIONS_KEY = "ApiCache_NavDestinations"
|
||||
FAVORITE_DESTINATIONS_KEY = "FavoriteDestinations"
|
||||
NAV_INSTRUCTION_STATE_KEY = "NavInstructionState"
|
||||
NAV_INSTRUCTION_COLLAPSED_KEY = "NavInstructionCollapsed"
|
||||
# Internal marker: preserve an offroad quick-start selection until the next onroad edge.
|
||||
START_ON_NEXT_DRIVE_KEY = "start_on_next_drive"
|
||||
|
||||
RECENT_DESTINATIONS_LIMIT = 10
|
||||
|
||||
@@ -83,29 +81,6 @@ def parse_destination_json(raw_value: str | bytes | dict[str, Any] | None) -> di
|
||||
return normalize_destination_payload(payload)
|
||||
|
||||
|
||||
def _parse_next_drive_destination(raw_value: Any) -> dict[str, Any] | None:
|
||||
payload = _json_value(raw_value, None)
|
||||
if not isinstance(payload, dict) or payload.get(START_ON_NEXT_DRIVE_KEY) is not True:
|
||||
return None
|
||||
return normalize_destination_payload(payload)
|
||||
|
||||
|
||||
def is_next_drive_destination(raw_value: Any) -> bool:
|
||||
return _parse_next_drive_destination(raw_value) is not None
|
||||
|
||||
|
||||
def activate_next_drive_destination(params: Any) -> dict[str, Any] | None:
|
||||
"""Promote an offroad quick-start destination to a normal active destination."""
|
||||
raw_value = _param_get(params, NAVIGATION_DESTINATION_KEY, "")
|
||||
destination = _parse_next_drive_destination(raw_value)
|
||||
payload = _json_value(raw_value, None)
|
||||
if destination is not None and isinstance(payload, dict):
|
||||
payload = dict(payload)
|
||||
payload.pop(START_ON_NEXT_DRIVE_KEY, None)
|
||||
params.put(NAVIGATION_DESTINATION_KEY, json.dumps(payload))
|
||||
return destination
|
||||
|
||||
|
||||
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 ''}"
|
||||
@@ -197,26 +172,21 @@ def set_navigation_destination(
|
||||
payload: Any,
|
||||
*,
|
||||
skip_if_same: bool = False,
|
||||
start_on_next_drive: bool = False,
|
||||
) -> dict[str, Any] | None:
|
||||
destination = normalize_destination_payload(payload)
|
||||
if destination is None:
|
||||
return None
|
||||
|
||||
current_raw = _param_get(params, NAVIGATION_DESTINATION_KEY, "")
|
||||
if skip_if_same:
|
||||
current = parse_destination_json(current_raw)
|
||||
if same_destination(current, destination) and is_next_drive_destination(current_raw) == start_on_next_drive:
|
||||
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)
|
||||
stored_destination = dict(destination)
|
||||
if start_on_next_drive:
|
||||
stored_destination[START_ON_NEXT_DRIVE_KEY] = True
|
||||
params.put(NAVIGATION_DESTINATION_KEY, json.dumps(stored_destination))
|
||||
params.put(NAVIGATION_DESTINATION_KEY, json.dumps(destination))
|
||||
params.put(RECENT_DESTINATIONS_KEY, recent_destinations)
|
||||
return destination
|
||||
|
||||
@@ -390,14 +360,8 @@ class NavigationDestinationStore:
|
||||
payload: Any,
|
||||
*,
|
||||
skip_if_same: bool = False,
|
||||
start_on_next_drive: bool = False,
|
||||
) -> dict[str, Any] | None:
|
||||
return set_navigation_destination(
|
||||
self.params,
|
||||
payload,
|
||||
skip_if_same=skip_if_same,
|
||||
start_on_next_drive=start_on_next_drive,
|
||||
)
|
||||
return set_navigation_destination(self.params, payload, skip_if_same=skip_if_same)
|
||||
|
||||
def clear_navigation(self) -> bool:
|
||||
collapsed_supported = True
|
||||
|
||||
@@ -4,11 +4,8 @@ from openpilot.starpilot.navigation.destination_store import (
|
||||
FAVORITE_DESTINATIONS_KEY,
|
||||
NAVIGATION_DESTINATION_KEY,
|
||||
RECENT_DESTINATIONS_KEY,
|
||||
START_ON_NEXT_DRIVE_KEY,
|
||||
NavigationDestinationStore,
|
||||
activate_next_drive_destination,
|
||||
favorite_destination_id,
|
||||
is_next_drive_destination,
|
||||
load_favorite_destinations,
|
||||
normalize_destination_payload,
|
||||
normalize_favorite_destination,
|
||||
@@ -158,27 +155,6 @@ def test_destination_write_updates_active_destination_and_recents():
|
||||
assert params.values[RECENT_DESTINATIONS_KEY][0]["place_name"] == "Home"
|
||||
|
||||
|
||||
def test_next_drive_destination_is_marked_then_canonicalized_on_activation():
|
||||
params = FakeParams({RECENT_DESTINATIONS_KEY: "[]"})
|
||||
|
||||
destination = set_navigation_destination(
|
||||
params,
|
||||
{"name": "Home", "latitude": 1, "longitude": 2},
|
||||
start_on_next_drive=True,
|
||||
)
|
||||
|
||||
stored = json.loads(params.values[NAVIGATION_DESTINATION_KEY])
|
||||
assert stored[START_ON_NEXT_DRIVE_KEY] is True
|
||||
assert is_next_drive_destination(stored)
|
||||
|
||||
stored["routeId"] = "main"
|
||||
params.values[NAVIGATION_DESTINATION_KEY] = json.dumps(stored)
|
||||
assert activate_next_drive_destination(params) == destination
|
||||
promoted = json.loads(params.values[NAVIGATION_DESTINATION_KEY])
|
||||
assert promoted == {**destination, "routeId": "main"}
|
||||
assert not is_next_drive_destination(params.values[NAVIGATION_DESTINATION_KEY])
|
||||
|
||||
|
||||
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}),
|
||||
@@ -200,31 +176,6 @@ def test_same_destination_write_is_idempotent_and_does_not_touch_recents():
|
||||
assert [key for key, _value in settings_params.writes] == [NAVIGATION_DESTINATION_KEY, RECENT_DESTINATIONS_KEY]
|
||||
|
||||
|
||||
def test_same_destination_is_rewritten_when_next_drive_intent_changes():
|
||||
params = FakeParams({
|
||||
NAVIGATION_DESTINATION_KEY: json.dumps({"name": "Home", "latitude": 1, "longitude": 2}),
|
||||
RECENT_DESTINATIONS_KEY: [],
|
||||
})
|
||||
|
||||
set_navigation_destination(
|
||||
params,
|
||||
{"name": "Home", "latitude": 1, "longitude": 2},
|
||||
skip_if_same=True,
|
||||
start_on_next_drive=True,
|
||||
)
|
||||
|
||||
assert is_next_drive_destination(params.values[NAVIGATION_DESTINATION_KEY])
|
||||
first_write_count = len(params.writes)
|
||||
|
||||
set_navigation_destination(
|
||||
params,
|
||||
{"name": "Home", "latitude": 1, "longitude": 2},
|
||||
skip_if_same=True,
|
||||
start_on_next_drive=True,
|
||||
)
|
||||
assert len(params.writes) == first_write_count
|
||||
|
||||
|
||||
def test_navigation_destination_store_keeps_settings_favorite_migration_and_mutations():
|
||||
params = FakeParams({
|
||||
FAVORITE_DESTINATIONS_KEY: json.dumps([{"name": "Home", "latitude": 1, "longitude": 2}]),
|
||||
|
||||
+26
-10
@@ -45,7 +45,6 @@ from openpilot.starpilot.common.starpilot_variables import (
|
||||
LEGACY_STARPILOT_STATS_KEY_RENAMES,
|
||||
get_starpilot_toggles,
|
||||
)
|
||||
from openpilot.starpilot.navigation.destination_store import activate_next_drive_destination, is_next_drive_destination
|
||||
|
||||
_MANAGER_IMPORT_DONE = time.monotonic()
|
||||
_manager_import_timing_line = (
|
||||
@@ -134,18 +133,33 @@ def get_nav_offroad_clear_timeout_seconds(params) -> int:
|
||||
return max(_get_int_param_value(params, "ClearNavOnOffroadTimeoutMinutes", 0), 0) * 60
|
||||
|
||||
|
||||
def update_nav_offroad_clear_state(params, started: bool, tracked_destination, tracked_started_at, now: float):
|
||||
nav_destination = params.get("NavDestination")
|
||||
if started or not params.get_bool("ClearNavOnOffroad") or not nav_destination or is_next_drive_destination(nav_destination):
|
||||
def update_nav_offroad_clear_state(
|
||||
params,
|
||||
started: bool,
|
||||
tracked_destination,
|
||||
tracked_started_at,
|
||||
now: float,
|
||||
*,
|
||||
offroad_transition: bool,
|
||||
):
|
||||
if started or not params.get_bool("ClearNavOnOffroad"):
|
||||
return None, None
|
||||
|
||||
if nav_destination != tracked_destination or tracked_started_at is None:
|
||||
nav_destination = params.get("NavDestination")
|
||||
if offroad_transition:
|
||||
if not nav_destination:
|
||||
return None, None
|
||||
tracked_destination = nav_destination
|
||||
tracked_started_at = now
|
||||
elif tracked_destination is None or tracked_started_at is None:
|
||||
return None, None
|
||||
|
||||
if nav_destination != tracked_destination:
|
||||
return None, None
|
||||
|
||||
if now - tracked_started_at >= get_nav_offroad_clear_timeout_seconds(params):
|
||||
current_destination = params.get("NavDestination")
|
||||
if current_destination != nav_destination or is_next_drive_destination(current_destination):
|
||||
if current_destination != tracked_destination:
|
||||
return None, None
|
||||
params.remove("NavDestination")
|
||||
return None, None
|
||||
@@ -1156,11 +1170,13 @@ def manager_thread() -> None:
|
||||
# StarPilot variables
|
||||
params_memory.clear_all(ParamKeyFlag.CLEAR_ON_OFFROAD_TRANSITION)
|
||||
|
||||
if started and not started_prev:
|
||||
activate_next_drive_destination(params)
|
||||
|
||||
offroad_nav_destination, offroad_nav_started_at = update_nav_offroad_clear_state(
|
||||
params, started, offroad_nav_destination, offroad_nav_started_at, time.monotonic()
|
||||
params,
|
||||
started,
|
||||
offroad_nav_destination,
|
||||
offroad_nav_started_at,
|
||||
time.monotonic(),
|
||||
offroad_transition=not started and started_prev,
|
||||
)
|
||||
|
||||
ignition = any(ps.ignitionLine or ps.ignitionCan for ps in sm['pandaStates'] if ps.pandaType != log.PandaState.PandaType.unknown)
|
||||
|
||||
@@ -12,7 +12,6 @@ import openpilot.system.manager.manager as manager
|
||||
from openpilot.system.manager.process import ensure_running
|
||||
from openpilot.system.manager.process_config import BigDeviceUIProcess, managed_processes, procs
|
||||
from openpilot.system.hardware import HARDWARE
|
||||
from openpilot.starpilot.navigation.destination_store import START_ON_NEXT_DRIVE_KEY
|
||||
|
||||
os.environ['FAKEUPLOAD'] = "1"
|
||||
|
||||
@@ -76,57 +75,93 @@ class FileBackedFakeParams:
|
||||
Path(self.get_param_path(key)).unlink(missing_ok=True)
|
||||
|
||||
|
||||
def test_offroad_navigation_cleanup_preserves_next_drive_destination(tmp_path):
|
||||
destination = {
|
||||
"name": "Home",
|
||||
"place_name": "Home",
|
||||
"latitude": 1.0,
|
||||
"longitude": 2.0,
|
||||
START_ON_NEXT_DRIVE_KEY: True,
|
||||
}
|
||||
def test_navigation_selected_while_already_offroad_is_not_tracked_for_cleanup(tmp_path):
|
||||
params = FileBackedFakeParams(tmp_path / "params", {
|
||||
"ClearNavOnOffroad": True,
|
||||
"ClearNavOnOffroadTimeoutMinutes": 0,
|
||||
"NavDestination": destination,
|
||||
})
|
||||
|
||||
state = manager.update_nav_offroad_clear_state(params, False, None, None, 10.0)
|
||||
state = manager.update_nav_offroad_clear_state(
|
||||
params, False, None, None, 10.0, offroad_transition=False
|
||||
)
|
||||
assert state == (None, None)
|
||||
|
||||
destination = {"name": "Home", "latitude": 1.0, "longitude": 2.0}
|
||||
params.put("NavDestination", destination)
|
||||
state = manager.update_nav_offroad_clear_state(
|
||||
params, False, *state, 20.0, offroad_transition=False
|
||||
)
|
||||
|
||||
assert state == (None, None)
|
||||
assert json.loads(params.get("NavDestination")) == destination
|
||||
|
||||
state = manager.update_nav_offroad_clear_state(
|
||||
params, True, *state, 30.0, offroad_transition=False
|
||||
)
|
||||
assert state == (None, None)
|
||||
assert json.loads(params.get("NavDestination")) == destination
|
||||
|
||||
def test_next_drive_selection_disarms_delayed_cleanup_for_same_destination(tmp_path):
|
||||
active_destination = {
|
||||
"name": "Home",
|
||||
"place_name": "Home",
|
||||
"latitude": 1.0,
|
||||
"longitude": 2.0,
|
||||
}
|
||||
state = manager.update_nav_offroad_clear_state(
|
||||
params, False, *state, 40.0, offroad_transition=True
|
||||
)
|
||||
assert state == (None, None)
|
||||
assert params.get("NavDestination") is None
|
||||
|
||||
|
||||
def test_replacement_destination_disarms_delayed_cleanup(tmp_path):
|
||||
params = FileBackedFakeParams(tmp_path / "params", {
|
||||
"ClearNavOnOffroad": True,
|
||||
"ClearNavOnOffroadTimeoutMinutes": 15,
|
||||
"NavDestination": active_destination,
|
||||
"NavDestination": {"name": "Old", "latitude": 1.0, "longitude": 2.0},
|
||||
})
|
||||
|
||||
tracked = manager.update_nav_offroad_clear_state(params, False, None, None, 10.0)
|
||||
assert tracked == (params.get("NavDestination"), 10.0)
|
||||
tracked = manager.update_nav_offroad_clear_state(
|
||||
params, False, None, None, 10.0, offroad_transition=True
|
||||
)
|
||||
replacement = {"name": "New", "latitude": 3.0, "longitude": 4.0}
|
||||
params.put("NavDestination", replacement)
|
||||
|
||||
params.put("NavDestination", {**active_destination, START_ON_NEXT_DRIVE_KEY: True})
|
||||
state = manager.update_nav_offroad_clear_state(params, False, *tracked, 20.0)
|
||||
state = manager.update_nav_offroad_clear_state(
|
||||
params, False, *tracked, 20.0, offroad_transition=False
|
||||
)
|
||||
|
||||
assert state == (None, None)
|
||||
assert json.loads(params.get("NavDestination"))[START_ON_NEXT_DRIVE_KEY] is True
|
||||
assert json.loads(params.get("NavDestination")) == replacement
|
||||
|
||||
|
||||
def test_unmarked_navigation_destination_still_clears_immediately_offroad(tmp_path):
|
||||
def test_active_navigation_clears_on_offroad_transition(tmp_path):
|
||||
params = FileBackedFakeParams(tmp_path / "params", {
|
||||
"ClearNavOnOffroad": True,
|
||||
"ClearNavOnOffroadTimeoutMinutes": 0,
|
||||
"NavDestination": {"name": "Home", "latitude": 1.0, "longitude": 2.0},
|
||||
})
|
||||
|
||||
state = manager.update_nav_offroad_clear_state(params, False, None, None, 10.0)
|
||||
state = manager.update_nav_offroad_clear_state(
|
||||
params, False, None, None, 10.0, offroad_transition=True
|
||||
)
|
||||
|
||||
assert state == (None, None)
|
||||
assert params.get("NavDestination") is None
|
||||
|
||||
|
||||
def test_active_navigation_clears_after_offroad_timeout(tmp_path):
|
||||
params = FileBackedFakeParams(tmp_path / "params", {
|
||||
"ClearNavOnOffroad": True,
|
||||
"ClearNavOnOffroadTimeoutMinutes": 15,
|
||||
"NavDestination": {"name": "Home", "latitude": 1.0, "longitude": 2.0},
|
||||
})
|
||||
|
||||
tracked = manager.update_nav_offroad_clear_state(
|
||||
params, False, None, None, 10.0, offroad_transition=True
|
||||
)
|
||||
tracked = manager.update_nav_offroad_clear_state(
|
||||
params, False, *tracked, 909.0, offroad_transition=False
|
||||
)
|
||||
assert params.get("NavDestination") is not None
|
||||
|
||||
state = manager.update_nav_offroad_clear_state(
|
||||
params, False, *tracked, 910.0, offroad_transition=False
|
||||
)
|
||||
|
||||
assert state == (None, None)
|
||||
assert params.get("NavDestination") is None
|
||||
@@ -134,12 +169,11 @@ def test_unmarked_navigation_destination_still_clears_immediately_offroad(tmp_pa
|
||||
|
||||
def test_offroad_cleanup_does_not_remove_destination_replaced_after_snapshot(tmp_path):
|
||||
old_destination = json.dumps({"name": "Old", "latitude": 1.0, "longitude": 2.0})
|
||||
next_drive_destination = {
|
||||
replacement_destination = {
|
||||
"name": "Home",
|
||||
"place_name": "Home",
|
||||
"latitude": 3.0,
|
||||
"longitude": 4.0,
|
||||
START_ON_NEXT_DRIVE_KEY: True,
|
||||
}
|
||||
|
||||
class SnapshotRaceParams(FileBackedFakeParams):
|
||||
@@ -161,10 +195,12 @@ def test_offroad_cleanup_does_not_remove_destination_replaced_after_snapshot(tmp
|
||||
params = SnapshotRaceParams(tmp_path / "params", {
|
||||
"ClearNavOnOffroad": True,
|
||||
"ClearNavOnOffroadTimeoutMinutes": 0,
|
||||
"NavDestination": next_drive_destination,
|
||||
"NavDestination": replacement_destination,
|
||||
})
|
||||
|
||||
state = manager.update_nav_offroad_clear_state(params, False, None, None, 10.0)
|
||||
state = manager.update_nav_offroad_clear_state(
|
||||
params, False, None, None, 10.0, offroad_transition=True
|
||||
)
|
||||
|
||||
assert state == (None, None)
|
||||
assert "NavDestination" not in params.removed
|
||||
|
||||
Reference in New Issue
Block a user