mawei/StarPilot
1
0
Fork 0
mirror of https://github.com/firestar5683/StarPilot.git synced 2026-08-02 13:29:26 +08:00
Code Issues Packages Projects Releases Wiki Activity
Files
CodenameAdeeb
StarPilot/selfdrive/ui/layouts/settings/starpilot/maps.py
T
firestarsdog 370654637a Big UI : Map Data update
2026-07-22 21:08:38 -04:00

968 lines
36 KiB
Python
Raw Permalink Blame History

from __future__ import annotations
import shutil
import threading
from dataclasses import dataclass, replace
from pathlib import Path
from cereal import log, messaging
import pyray as rl
from openpilot.common.params import Params
from openpilot.selfdrive.ui.ui_state import device, ui_state
from openpilot.system.ui.lib.application import FontWeight, gui_app
from openpilot.system.ui.lib.multilang import tr, tr_noop, trn
from openpilot.system.ui.lib.text_measure import measure_text_cached
from openpilot.system.ui.widgets import DialogResult
from openpilot.system.ui.widgets.confirm_dialog import ConfirmDialog, alert_dialog
from openpilot.system.ui.widgets.label import gui_label, gui_text_box
from openpilot.system.ui.widgets.option_dialog import MultiOptionDialog
from openpilot.selfdrive.ui.layouts.settings.starpilot.aethergrid import (
AETHER_LIST_METRICS,
AetherButton,
AetherListColors,
AetherSegmentedControl,
DEFAULT_PANEL_STYLE,
PanelManagerView,
aether_begin_scissor_mode,
aether_end_scissor_mode,
draw_action_pill,
draw_busy_ring,
draw_empty_state_card,
draw_list_group_shell,
draw_metric_strip,
draw_section_header,
draw_selection_list_row,
with_alpha,
)
from openpilot.selfdrive.ui.layouts.settings.starpilot.panel import _SettingsPage
from openpilot.starpilot.common.maps_catalog import (
MAPS_CATALOG,
MAP_TOKEN_LABELS,
MAP_SCHEDULE_LABELS,
sanitize_selected_locations_csv,
schedule_label,
)
from openpilot.starpilot.common.maps_selection import COUNTRY_PREFIX, STATE_PREFIX
NetworkType = log.DeviceState.NetworkType
OFFLINE_MAPS_PATH = Path("/data/media/0/osm/offline")
CANCEL_REQUEST_TIMEOUT = 3.0
PANEL_STYLE = DEFAULT_PANEL_STYLE
MAPS_METRICS = replace(AETHER_LIST_METRICS, header_height=0)
COUNTRIES_SECTION = next(section for section in MAPS_CATALOG if section["key"] == "countries")
STATES_SECTION = next(section for section in MAPS_CATALOG if section["key"] == "states")
US_COUNTRY_TOKEN = f"{COUNTRY_PREFIX}US"
ALL_US_STATE_TOKENS = frozenset({
region["token"] for group in STATES_SECTION["groups"] for region in group["regions"]
})
REGIONAL_PACKAGES = {
"pkg:midwest": {
"title": tr_noop("U.S. Midwest Region (12 States)"),
"subtitle": tr_noop("IL, IN, IA, KS, MI, MN, MO, NE, ND, OH, SD, WI"),
"tokens": frozenset({r["token"] for g in STATES_SECTION["groups"] if g["key"] == "midwest" for r in g["regions"]}),
},
"pkg:northeast": {
"title": tr_noop("U.S. Northeast Region (9 States)"),
"subtitle": tr_noop("CT, ME, MA, NH, NJ, NY, PA, RI, VT"),
"tokens": frozenset({r["token"] for g in STATES_SECTION["groups"] if g["key"] == "northeast" for r in g["regions"]}),
},
"pkg:south": {
"title": tr_noop("U.S. South Region (17 States)"),
"subtitle": tr_noop("AL, AR, DE, DC, FL, GA, KY, LA, MD, MS, NC, OK, SC, TN, TX, VA, WV"),
"tokens": frozenset({r["token"] for g in STATES_SECTION["groups"] if g["key"] == "south" for r in g["regions"]}),
},
"pkg:west": {
"title": tr_noop("U.S. West Region (13 States)"),
"subtitle": tr_noop("AK, AZ, CA, CO, HI, ID, MT, NV, NM, OR, UT, WA, WY"),
"tokens": frozenset({r["token"] for g in STATES_SECTION["groups"] if g["key"] == "west" for r in g["regions"]}),
},
"pkg:territories": {
"title": tr_noop("U.S. Territories (5 Territories)"),
"subtitle": tr_noop("AS, GU, MP, PR, VI"),
"tokens": frozenset({r["token"] for g in STATES_SECTION["groups"] if g["key"] == "territories" for r in g["regions"]}),
},
}
STATUS_CARD_HEIGHT = 232.0
SEGMENTED_CONTROL_HEIGHT = 68.0
BROWSER_SECTION_HEADER_HEIGHT = 56.0
BROWSER_REGION_ROW_HEIGHT = 104.0
BROWSER_EMPTY_STATE_HEIGHT = 128.0
BROWSER_INSET = 18.0
HEADER_GAP = 16.0
SUBHEADER_GAP = 12.0
LIST_TOP_GAP = 12.0
FIXED_HEADER_HEIGHT = STATUS_CARD_HEIGHT + HEADER_GAP + SEGMENTED_CONTROL_HEIGHT + SUBHEADER_GAP + BROWSER_SECTION_HEADER_HEIGHT + LIST_TOP_GAP
def _format_mb(size_bytes: int) -> str:
mb = size_bytes / (1024 * 1024)
if mb >= 1024:
return f"{mb / 1024:.2f} GB"
return f"{mb:.2f} MB"
def _format_elapsed_ms(elapsed_ms: int) -> str:
if elapsed_ms <= 0:
return tr("Calculating...")
total_seconds = elapsed_ms // 1000
hours = total_seconds // 3600
minutes = (total_seconds % 3600) // 60
seconds = total_seconds % 60
if hours > 0:
return f"{hours:d}:{minutes:02d}:{seconds:02d}"
return f"{minutes:d}:{seconds:02d}"
def _format_eta_ms(elapsed_ms: int, downloaded_files: int, total_files: int) -> str:
if elapsed_ms <= 0 or downloaded_files <= 0 or total_files <= 0 or downloaded_files >= total_files:
return tr("Calculating...")
remaining_files = total_files - downloaded_files
if remaining_files <= 0:
return tr("Almost done")
files_per_ms = downloaded_files / max(elapsed_ms, 1)
if files_per_ms <= 0:
return tr("Calculating...")
remaining_ms = int(remaining_files / files_per_ms)
return _format_elapsed_ms(remaining_ms)
def _localized_schedule_label(value) -> str:
return tr(schedule_label(value))
def _selected_token_set(selected_raw: str | bytes | None) -> set[str]:
normalized = sanitize_selected_locations_csv(selected_raw or "")
return {token for token in normalized.split(",") if token}
@dataclass(slots=True)
class MapsDownloadState:
active: bool = False
cancelled: bool = False
total_files: int = 0
downloaded_files: int = 0
primary_location: str = ""
percent: int = 0
progress_text: str = ""
class MapsManagerView(PanelManagerView):
METRICS = MAPS_METRICS
PANEL_STYLE = PANEL_STYLE
def __init__(self, controller: StarPilotMapsLayout):
super().__init__()
self._controller = controller
self._remove_rect = rl.Rectangle(0, 0, 0, 0)
self._source_segmented_control = self._child(
AetherSegmentedControl(
[tr("U.S. States & Regions"), tr("Other Countries")],
self._controller._get_source_segment_index,
self._controller._on_source_segment_change,
style=PANEL_STYLE,
suppress_background=True,
)
)
def _draw_header(self, rect: rl.Rectangle):
pass
def _measure_content_height(self, content_width: float) -> float:
downloaded, downloadable = self._controller._browse_regions_for_active_view()
total_rows = len(downloaded) + len(downloadable)
if total_rows <= 0:
return BROWSER_EMPTY_STATE_HEIGHT
h = total_rows * BROWSER_REGION_ROW_HEIGHT
if downloaded and downloadable:
h += BROWSER_SECTION_HEADER_HEIGHT + 16.0
return h
def _draw_static_elements(self, scroll_rect: rl.Rectangle, content_width: float):
# 1. Pinned Status Card
status_rect = rl.Rectangle(scroll_rect.x, scroll_rect.y, content_width, STATUS_CARD_HEIGHT)
self._draw_status_card(status_rect)
# 2. Pinned Source Segmented Control ([ U.S. States & Regions | Other Countries ])
seg_y = scroll_rect.y + STATUS_CARD_HEIGHT + HEADER_GAP
seg_rect = rl.Rectangle(scroll_rect.x + BROWSER_INSET, seg_y, content_width - BROWSER_INSET * 2, SEGMENTED_CONTROL_HEIGHT)
self._source_segmented_control.render(seg_rect)
# 3. Pinned Master Section Header
header_y = seg_y + SEGMENTED_CONTROL_HEIGHT + SUBHEADER_GAP
draw_section_header(
rl.Rectangle(scroll_rect.x + BROWSER_INSET, header_y, content_width - BROWSER_INSET * 2, BROWSER_SECTION_HEADER_HEIGHT),
tr("Map Packages & Regions"),
trailing_text=self._controller._active_view_count_text(),
title_size=36,
trailing_size=24,
style=PANEL_STYLE,
)
def _draw_scroll_content(self, scroll_rect: rl.Rectangle, content_width: float):
rows_top_y = scroll_rect.y + FIXED_HEADER_HEIGHT
rows_visible_h = max(100.0, scroll_rect.height - FIXED_HEADER_HEIGHT)
rows_scissor_rect = rl.Rectangle(scroll_rect.x, rows_top_y, content_width, rows_visible_h)
aether_begin_scissor_mode(
int(rows_scissor_rect.x),
int(rows_scissor_rect.y),
int(rows_scissor_rect.width),
int(rows_scissor_rect.height),
)
downloaded, downloadable = self._controller._browse_regions_for_active_view()
if not downloaded and not downloadable:
title, body = self._controller._browse_empty_state()
draw_empty_state_card(
rl.Rectangle(scroll_rect.x + BROWSER_INSET, rows_top_y + self._scroll_offset, content_width - BROWSER_INSET * 2, BROWSER_EMPTY_STATE_HEIGHT),
title,
body,
title_size=32,
body_size=24,
border=with_alpha(PANEL_STYLE.surface_border, 10),
style=PANEL_STYLE,
)
aether_end_scissor_mode()
return
y = rows_top_y + self._scroll_offset
# ── TOP SECTION: Downloaded / Installed Maps ──
if downloaded:
for index, region in enumerate(downloaded):
token = region["token"]
row_rect = rl.Rectangle(scroll_rect.x + BROWSER_INSET, y, content_width - BROWSER_INSET * 2, BROWSER_REGION_ROW_HEIGHT)
target_id = f"region:{token}"
hovered, pressed = self._interactive_state(target_id, row_rect, pad_y=0)
draw_selection_list_row(
row_rect,
title=tr(region["label"]),
subtitle=tr("Downloaded on device"),
action_text=tr("Downloaded"),
current=True,
hovered=hovered,
pressed=pressed,
is_last=index == len(downloaded) - 1 and not downloadable,
action_width=164,
action_pill=True,
action_text_size=26,
action_pill_height=56,
action_pill_width=154,
title_size=32,
subtitle_size=22,
row_separator=PANEL_STYLE.divider_color,
current_bg=PANEL_STYLE.current_fill,
current_border=PANEL_STYLE.current_border,
action_fill=with_alpha(AetherListColors.SUCCESS_SOFT, 36),
action_border=with_alpha(AetherListColors.SUCCESS, 60),
action_text_color=AetherListColors.HEADER,
)
y += BROWSER_REGION_ROW_HEIGHT
# ── THIN DIVIDER LINE + SECTION HEADER: Separating Downloaded from Downloadable ──
if downloadable:
y += 8.0
draw_section_header(
rl.Rectangle(scroll_rect.x + BROWSER_INSET, y, content_width - BROWSER_INSET * 2, BROWSER_SECTION_HEADER_HEIGHT),
tr("Available for Download"),
trailing_text=tr("{} available").format(len(downloadable)),
title_size=30,
trailing_size=22,
style=PANEL_STYLE,
)
y += BROWSER_SECTION_HEADER_HEIGHT + 8.0
# ── BOTTOM SECTION: Available / Downloadable Maps ──
for index, region in enumerate(downloadable):
token = region["token"]
selected = self._controller._get_map_state(token)
row_rect = rl.Rectangle(scroll_rect.x + BROWSER_INSET, y, content_width - BROWSER_INSET * 2, BROWSER_REGION_ROW_HEIGHT)
target_id = f"region:{token}"
hovered, pressed = self._interactive_state(target_id, row_rect, pad_y=0)
action_text = self._controller._region_action_text(token)
draw_selection_list_row(
row_rect,
title=tr(region["label"]),
subtitle=self._controller._region_primary_text(token),
action_text=action_text,
current=selected,
hovered=hovered,
pressed=pressed,
is_last=index == len(downloadable) - 1,
action_width=164,
action_pill=True,
action_text_size=26,
action_pill_height=56,
action_pill_width=154 if selected else 128,
title_size=32,
subtitle_size=22,
row_separator=PANEL_STYLE.divider_color,
current_bg=PANEL_STYLE.current_fill,
current_border=PANEL_STYLE.current_border,
action_fill=with_alpha(AetherListColors.SUCCESS_SOFT, 24) if selected else rl.Color(255, 255, 255, 8),
action_border=with_alpha(AetherListColors.SUCCESS, 48) if selected else rl.Color(255, 255, 255, 20),
action_text_color=AetherListColors.HEADER,
)
y += BROWSER_REGION_ROW_HEIGHT
aether_end_scissor_mode()
if self._content_height > rows_visible_h:
self._scrollbar.render(rows_scissor_rect, self._content_height, self._scroll_offset)
def _draw_status_card(self, rect: rl.Rectangle):
draw_list_group_shell(rect, style=PANEL_STYLE)
inset = 24.0
actions_w = min(400.0, max(320.0, rect.width * 0.36))
actions_x = rect.x + rect.width - inset - actions_w
content_x = rect.x + inset
summary_w = max(240.0, actions_x - 20.0 - content_x)
# Title (32pt Bold)
title_text = self._controller._progress_title()
gui_label(rl.Rectangle(content_x, rect.y + 18, summary_w, 34), title_text, 32, AetherListColors.HEADER, FontWeight.SEMI_BOLD)
# Selection Summary Badge next to title if selected (24pt, height 38)
if self._controller._selected_count() > 0:
chip_text = self._controller._selected_summary_text()
title_measured_w = measure_text_cached(gui_app.font(FontWeight.SEMI_BOLD), title_text, 32, spacing=1).x
chip_width = measure_text_cached(gui_app.font(FontWeight.SEMI_BOLD), chip_text, 22, spacing=1).x + 30.0
chip_x = content_x + title_measured_w + 16.0
if chip_x + chip_width <= content_x + summary_w:
draw_action_pill(
rl.Rectangle(chip_x, rect.y + 16, chip_width, 38),
chip_text,
with_alpha(AetherListColors.SUCCESS_SOFT, 30),
AetherListColors.SUCCESS_SOFT,
AetherListColors.HEADER,
font_size=24,
)
# Subtitle / Body Progress Description (26pt)
gui_text_box(
rl.Rectangle(content_x, rect.y + 58, summary_w, 54),
self._controller._progress_body(),
26,
AetherListColors.SUBTEXT,
font_weight=FontWeight.NORMAL,
line_scale=0.95,
)
# Telemetry Strip (Storage & Last Updated — Label 24pt, Value 32pt, value_top_offset=30 for zero overlap)
metrics = [
(tr("Storage"), self._controller._storage_text),
(tr("Last Updated"), self._controller._last_updated_text()),
]
draw_metric_strip(
rl.Rectangle(content_x, rect.y + 128, summary_w, 72),
metrics,
gap=36.0,
min_col_width=140.0,
label_size=24,
value_size=32,
style=PANEL_STYLE,
label_top_offset=0,
value_top_offset=30,
divider_top_offset=4,
divider_bottom_offset=54,
)
# Action Column
primary_h = 84.0
primary_rect = rl.Rectangle(actions_x, rect.y + 20, actions_w, primary_h)
self._controller._download_button.render(primary_rect)
if self._controller._download_state.active:
center = rl.Vector2(primary_rect.x + primary_rect.width - 28, primary_rect.y + primary_rect.height / 2)
draw_busy_ring(center, rl.get_time() * 160, PANEL_STYLE.accent, inner_radius=10, outer_radius=14, sweep=210, thickness=20)
# Bottom Row Actions: Schedule Button & Remove Maps Pill (64px height)
bottom_y = rect.y + 116.0
bottom_h = 64.0
col_gap = 10.0
sched_w = (actions_w - col_gap) * 0.58
remove_w = (actions_w - col_gap) * 0.42
sched_rect = rl.Rectangle(actions_x, bottom_y, sched_w, bottom_h)
self._controller._schedule_button.render(sched_rect)
self._remove_rect = rl.Rectangle(actions_x + sched_w + col_gap, bottom_y, remove_w, bottom_h)
enabled = self._controller._remove_enabled()
hovered, pressed = self._interactive_state("action:remove_maps", self._remove_rect, pad_y=4)
remove_bg = with_alpha(AetherListColors.DANGER_SOFT, 36 if (pressed or hovered) else (24 if enabled else 12))
remove_border = with_alpha(AetherListColors.DANGER, 90 if (pressed or hovered) else (56 if enabled else 24))
draw_action_pill(
self._remove_rect,
tr("Remove"),
remove_bg,
remove_border,
AetherListColors.HEADER if enabled else AetherListColors.MUTED,
font_size=26,
)
def _activate_target(self, target_id: str | None):
if not target_id:
return
if target_id == "action:remove_maps":
self._controller._on_remove()
return
if target_id.startswith("region:"):
self._controller._toggle_region(target_id.split(":", 1)[1])
class StarPilotMapsLayout(_SettingsPage):
VIEW_STATES = 0
VIEW_COUNTRIES = 1
def __init__(self):
super().__init__()
self._params_memory = Params(memory=True)
self._worker_params = Params()
self._map_sm = messaging.SubMaster(["mapdExtendedOut", "starpilotCarState"])
self._storage_text = "0 MB"
self._has_downloaded_data = False
self._storage_updated_at = 0.0
self._storage_refresh_thread: threading.Thread | None = None
self._storage_refresh_pending = False
self._storage_refresh_generation = 0
self._pending_storage_state: tuple[int, str, bool] | None = None
self._download_started_at: float | None = None
self._cancel_requested_at: float | None = None
self._cancel_visual_until = 0.0
self._download_state = MapsDownloadState()
self._view_index = self.VIEW_STATES
self._cached_selected_tokens: set[str] = set()
self._download_button = self._child(
AetherButton(
self._primary_action_label,
self._on_primary_action,
enabled=self._primary_action_enabled,
emphasized=True,
)
)
self._schedule_button = self._child(
AetherButton(
lambda: tr("Update: {}").format(_localized_schedule_label(self._params.get('PreferredSchedule'))),
self._on_schedule,
emphasized=False,
)
)
self._manager_view = MapsManagerView(self)
self._refresh_storage_cache(force=True)
self._sync_download_state(force=True)
def show_event(self):
super().show_event()
# On panel launch, if no offline map files exist on disk and no download is actively running,
# reset any stale or default MapsSelected parameter so 0 regions show as selected on clean boot!
if not self._has_downloaded_data and not self._download_in_flight():
raw_selected = self._params.get("MapsSelected", encoding="utf-8") or ""
if raw_selected:
self._params.put("MapsSelected", "")
self._cached_selected_tokens = set()
if self._cancel_requested() and self._cancel_requested_at is None:
self._cancel_requested_at = rl.get_time()
if self._cancel_requested() and self._cancel_visual_until <= rl.get_time():
self._cancel_visual_until = rl.get_time() + 2.5
self._refresh_storage_cache(force=True)
self._sync_download_state(force=True)
def hide_event(self):
super().hide_event()
device.set_override_interactive_timeout(None)
def _update_state(self):
super()._update_state()
self._cached_selected_tokens = _selected_token_set(
self._params.get("MapsSelected", encoding="utf-8") or ""
)
self._sync_download_state()
if self._pending_storage_state is not None:
generation, storage_text, has_downloaded_data = self._pending_storage_state
self._pending_storage_state = None
if generation == self._storage_refresh_generation:
self._storage_text = storage_text
self._has_downloaded_data = has_downloaded_data
self._storage_updated_at = rl.get_time()
self._storage_refresh_pending = False
self._refresh_storage_cache()
if self._download_state.active:
device.set_override_interactive_timeout(300)
else:
device.set_override_interactive_timeout(None)
if self._cancel_requested_at is not None and not self._download_state.active:
if (rl.get_time() - self._cancel_requested_at) >= CANCEL_REQUEST_TIMEOUT:
self._params_memory.remove("CancelDownloadMaps")
self._cancel_requested_at = None
if self._cancel_visual_until and rl.get_time() >= self._cancel_visual_until and not self._download_state.active:
self._cancel_visual_until = 0.0
def _sync_download_state(self, force: bool = False):
del force
self._map_sm.update(0)
progress = self._map_sm["mapdExtendedOut"].downloadProgress if self._map_sm.valid.get("mapdExtendedOut", False) else None
active = bool(progress.active) if progress is not None else False
if active and self._download_started_at is None:
self._download_started_at = rl.get_time()
if active:
self._cancel_requested_at = None
self._cancel_visual_until = 0.0
if not self._download_in_flight() and self._download_started_at is not None:
self._download_started_at = None
total_files = int(progress.totalFiles) if progress is not None else 0
downloaded_files = int(progress.downloadedFiles) if progress is not None else 0
percent = int((downloaded_files * 100) / max(total_files, 1)) if total_files > 0 else 0
primary_location = ""
if progress is not None and len(progress.locationDetails) > 0:
primary_location = str(progress.locationDetails[0].location)
elif progress is not None and len(progress.locations) > 0:
primary_location = str(progress.locations[0])
progress_text = ""
if active:
progress_text = tr("{} / {} ({}%)").format(downloaded_files, total_files, percent)
if primary_location:
progress_text = f"{progress_text} {primary_location}"
self._download_state = MapsDownloadState(
active=active,
cancelled=bool(progress.cancelled) if progress is not None else False,
total_files=total_files,
downloaded_files=downloaded_files,
primary_location=primary_location,
percent=percent,
progress_text=progress_text,
)
def _refresh_storage_cache(self, force: bool = False):
now = rl.get_time()
if self._storage_refresh_pending:
return
if not force and (now - self._storage_updated_at) < 4.0:
return
generation = self._storage_refresh_generation + 1
self._storage_refresh_generation = generation
def refresh_worker():
result: tuple[str, bool] | None = None
try:
result = self._calculate_storage_state()
finally:
if result is None:
self._storage_refresh_pending = False
else:
self._pending_storage_state = (generation, result[0], result[1])
self._storage_refresh_pending = True
self._storage_updated_at = now
self._storage_refresh_thread = threading.Thread(target=refresh_worker, daemon=True)
self._storage_refresh_thread.start()
def _calculate_storage_state(self) -> tuple[str, bool]:
if not OFFLINE_MAPS_PATH.exists():
return "0 MB", False
total_size = 0
has_files = False
for path in OFFLINE_MAPS_PATH.rglob("*"):
try:
if not path.is_file():
continue
has_files = True
total_size += path.stat().st_size
except OSError:
continue
return _format_mb(total_size), has_files
def _selected_tokens(self) -> set[str]:
return self._cached_selected_tokens
def _selected_count(self) -> int:
tokens = self._selected_tokens()
if US_COUNTRY_TOKEN in tokens:
return len(ALL_US_STATE_TOKENS)
return len(tokens)
def _selected_summary_text(self) -> str:
count = self._selected_count()
if count == 0:
return tr("No regions selected")
if US_COUNTRY_TOKEN in self._selected_tokens():
return tr("Whole U.S. (56 states)")
return trn("{} region selected", "{} regions selected", count).format(count)
def _selected_primary_token(self) -> str | None:
selected = sorted(self._selected_tokens())
if not selected:
return None
return selected[0]
def _selected_primary_label(self) -> str:
token = self._selected_primary_token()
if not token:
return ""
if token == US_COUNTRY_TOKEN:
return tr("Whole U.S.")
return tr(MAP_TOKEN_LABELS.get(token, token))
def _selection_preview_text(self) -> str:
count = self._selected_count()
if count <= 0:
return tr("No regions selected yet")
if US_COUNTRY_TOKEN in self._selected_tokens():
return tr("Whole U.S. package")
primary_label = self._selected_primary_label()
if count == 1:
return primary_label
return tr("{} + {} more").format(primary_label, count - 1)
def _has_full_us_selected(self) -> bool:
tokens = self._selected_tokens()
return (US_COUNTRY_TOKEN in tokens) or ALL_US_STATE_TOKENS.issubset(tokens)
def _is_states_view(self) -> bool:
return self._view_index == self.VIEW_STATES
def _get_source_segment_index(self) -> int:
return 0 if self._is_states_view() else 1
def _on_source_segment_change(self, idx: int):
self._view_index = self.VIEW_STATES if idx == 0 else self.VIEW_COUNTRIES
def _all_country_regions(self) -> list[dict]:
regions = [region for group in COUNTRIES_SECTION["groups"] for region in group["regions"]]
return sorted(regions, key=lambda r: tr(r["label"]))
def _all_us_states_regions(self) -> list[dict]:
regions = [region for group in STATES_SECTION["groups"] for region in group["regions"]]
return sorted(regions, key=lambda r: tr(r["label"]))
def _bulk_package_regions(self) -> list[dict]:
return [
{"token": US_COUNTRY_TOKEN, "label": tr("Whole U.S. (All 50 States & Territories)")},
*[
{"token": key, "label": tr(spec["title"])}
for key, spec in REGIONAL_PACKAGES.items()
]
]
def _active_view_count_text(self) -> str:
if not self._is_states_view():
total = len(self._all_country_regions())
selected_count = sum(1 for region in self._all_country_regions() if self._get_map_state(region["token"]))
else:
total = len(self._all_us_states_regions())
selected_count = self._selected_count()
if self._has_full_us_selected():
return tr("Whole U.S. Selected")
if selected_count <= 0:
return tr("{} available - Tap to select").format(total)
return tr("{} selected | {} available").format(selected_count, max(0, total - selected_count))
def _toggle_region(self, token: str):
self._set_map_state(token, not self._get_map_state(token))
def _is_region_downloaded(self, token: str) -> bool:
"""Returns True ONLY if map data is actually downloaded on disk AND token is part of installed set."""
if not self._has_downloaded_data:
return False
return self._get_map_state(token)
def _region_primary_text(self, token: str) -> str:
if token == US_COUNTRY_TOKEN:
return tr("Master Package - Full country map data")
if token in REGIONAL_PACKAGES:
return tr(REGIONAL_PACKAGES[token]["subtitle"])
if self._get_map_state(token):
return tr("Selected for download")
return tr("Tap to select state")
def _region_action_text(self, token: str) -> str:
if self._get_map_state(token):
return tr("Selected")
return tr("Add")
def _browse_empty_state(self) -> tuple[str, str]:
return tr("No regions available"), tr("Switch sources to keep browsing maps.")
def _browse_regions_for_active_view(self) -> tuple[list[dict], list[dict]]:
"""Partition active view regions into (downloaded_on_device, available_for_download)."""
if not self._is_states_view():
all_regions = self._all_country_regions()
else:
all_regions = [*self._bulk_package_regions(), *self._all_us_states_regions()]
# If NO map data exists on disk, ZERO regions are downloaded on device!
if not self._has_downloaded_data:
return [], all_regions
downloaded = []
downloadable = []
for region in all_regions:
token = region["token"]
if self._is_region_downloaded(token):
downloaded.append(region)
else:
downloadable.append(region)
return downloaded, downloadable
def _get_map_state(self, token: str) -> bool:
tokens = self._selected_tokens()
if not tokens:
return False
if US_COUNTRY_TOKEN in tokens:
return True
if token == US_COUNTRY_TOKEN:
return ALL_US_STATE_TOKENS.issubset(tokens)
if token in REGIONAL_PACKAGES:
return REGIONAL_PACKAGES[token]["tokens"].issubset(tokens)
return token in tokens
def _set_map_state(self, token: str, state: bool):
selected = set(self._cached_selected_tokens)
# Expand Whole U.S. to individual states first if modifying a subset
if US_COUNTRY_TOKEN in selected:
selected.remove(US_COUNTRY_TOKEN)
selected.update(ALL_US_STATE_TOKENS)
if state:
if token == US_COUNTRY_TOKEN:
selected = {US_COUNTRY_TOKEN}
elif token in REGIONAL_PACKAGES:
selected.update(REGIONAL_PACKAGES[token]["tokens"])
else:
selected.add(token)
else:
if token == US_COUNTRY_TOKEN:
selected.difference_update(ALL_US_STATE_TOKENS)
selected.discard(US_COUNTRY_TOKEN)
elif token in REGIONAL_PACKAGES:
selected.difference_update(REGIONAL_PACKAGES[token]["tokens"])
else:
selected.discard(token)
# Auto-consolidate to c:US if all 56 states are selected
if ALL_US_STATE_TOKENS.issubset(selected):
selected = {US_COUNTRY_TOKEN}
self._params.put("MapsSelected", sanitize_selected_locations_csv(sorted(selected)))
self._cached_selected_tokens = selected
def _is_online(self) -> bool:
network_type = ui_state.sm["deviceState"].networkType if ui_state.sm.valid.get("deviceState", False) else NetworkType.none
return network_type != NetworkType.none
def _is_parked(self) -> bool:
if ui_state.is_offroad():
return True
if self._map_sm.valid.get("starpilotCarState", False):
return bool(self._map_sm["starpilotCarState"].isParked)
return False
def _download_requested(self) -> bool:
return self._params_memory.get_bool("DownloadMaps")
def _cancel_requested(self) -> bool:
return self._params_memory.get_bool("CancelDownloadMaps")
def _is_visually_cancelling(self) -> bool:
return self._cancel_requested() or (self._cancel_visual_until > rl.get_time())
def _download_in_flight(self) -> bool:
return self._download_state.active or self._download_requested()
def _remove_enabled(self) -> bool:
return self._has_downloaded_data and self._is_parked() and not self._download_in_flight() and not self._is_visually_cancelling()
def _download_gate_reason(self) -> str:
reasons = []
if self._download_in_flight():
reasons.append(tr("Download in progress"))
if not self._is_online():
reasons.append(tr("Connect to the internet"))
if not self._is_parked():
reasons.append(tr("Park the vehicle to download"))
return "\n".join(reasons) if reasons else ""
def _primary_action_enabled(self) -> bool:
if self._is_visually_cancelling():
return False
if self._download_in_flight():
return True
return self._download_gate_reason() == ""
def _primary_action_label(self) -> str:
if self._is_visually_cancelling():
return tr("Cancelling...")
if self._download_in_flight():
return tr("Cancel Download")
if self._selected_count() == 0:
return tr("Select Regions Below")
return tr("Download Offline Maps")
def _on_primary_action(self):
if self._download_in_flight():
self._on_cancel()
elif self._selected_count() == 0:
gui_app.push_widget(alert_dialog(tr("Please select 'Whole U.S.' or pick individual states from the list below.")))
else:
self._on_download()
def _on_schedule(self):
localized_options = [(value, tr(label)) for value, label in MAP_SCHEDULE_LABELS.items()]
options = [label for _, label in localized_options]
current = _localized_schedule_label(self._params.get("PreferredSchedule"))
value_by_label = {label: value for value, label in localized_options}
def on_select(res):
if res == DialogResult.CONFIRM and dialog.selection:
selected_value = value_by_label.get(dialog.selection)
if selected_value is not None:
self._params.put_int("PreferredSchedule", selected_value)
dialog = MultiOptionDialog(tr("Auto Update Schedule"), options, current, callback=on_select)
gui_app.push_widget(dialog)
def _on_download(self):
gate_reason = self._download_gate_reason()
if gate_reason:
gui_app.push_widget(alert_dialog(gate_reason))
return
current_selected = self._params.get("MapsSelected", encoding="utf-8") or ""
selected_raw = sanitize_selected_locations_csv(current_selected)
if selected_raw != current_selected:
self._params.put("MapsSelected", selected_raw)
if not selected_raw:
gui_app.push_widget(alert_dialog(tr("Please select 'Whole U.S.' or pick individual states from the list below.")))
return
def on_confirm(res):
if res == DialogResult.CONFIRM:
gate_reason = self._download_gate_reason()
if gate_reason:
gui_app.push_widget(alert_dialog(gate_reason))
return
self._params_memory.put_bool("DownloadMaps", True)
self._params_memory.remove("CancelDownloadMaps")
self._download_started_at = rl.get_time()
gui_app.push_widget(ConfirmDialog(tr("Start downloading offline maps for the selected regions?"), tr("Download"), callback=on_confirm))
def _on_cancel(self):
def on_confirm(res):
if res == DialogResult.CONFIRM:
if not self._download_in_flight():
self._cancel_requested_at = None
self._cancel_visual_until = 0.0
return
self._params_memory.put_bool("CancelDownloadMaps", True)
self._params_memory.remove("DownloadMaps")
self._cancel_requested_at = rl.get_time()
self._cancel_visual_until = rl.get_time() + 2.5
gui_app.push_widget(ConfirmDialog(tr("Cancel the current map download?"), tr("Cancel Download"), callback=on_confirm))
def _on_remove(self):
if not self._remove_enabled():
if not self._is_parked():
gui_app.push_widget(alert_dialog(tr("Park to remove downloaded maps.")))
return
def on_confirm(res):
if res == DialogResult.CONFIRM:
if not self._remove_enabled():
if not self._is_parked():
gui_app.push_widget(alert_dialog(tr("Park to remove downloaded maps.")))
return
def remove_worker():
if OFFLINE_MAPS_PATH.exists():
shutil.rmtree(OFFLINE_MAPS_PATH, ignore_errors=True)
OFFLINE_MAPS_PATH.mkdir(parents=True, exist_ok=True)
self._storage_refresh_generation += 1
self._pending_storage_state = (self._storage_refresh_generation, "0 MB", False)
self._storage_refresh_pending = False
self._storage_updated_at = 0.0
threading.Thread(target=remove_worker, daemon=True).start()
gui_app.push_widget(alert_dialog(tr("Removing offline maps...")))
gui_app.push_widget(ConfirmDialog(tr("Delete all downloaded offline map data?"), tr("Remove Maps"), callback=on_confirm))
def _last_updated_text(self) -> str:
last_update = self._worker_params.get("LastMapsUpdate", encoding="utf-8")
return last_update or tr("Never")
def _progress_title(self) -> str:
if self._is_visually_cancelling():
return tr("Cancelling Download")
if self._download_state.active:
return tr("Downloading Maps")
if self._download_requested():
return tr("Starting Download")
if self._has_downloaded_data:
return tr("Offline Maps")
if self._selected_count() == 0:
return tr("Select Map Data")
return tr("Download Readiness")
def _progress_body(self) -> str:
if self._download_state.active:
elapsed_ms = int((rl.get_time() - (self._download_started_at or rl.get_time())) * 1000)
elapsed_text = _format_elapsed_ms(elapsed_ms)
eta_text = _format_eta_ms(elapsed_ms, self._download_state.downloaded_files, self._download_state.total_files)
if self._download_state.primary_location:
return tr("{}\nElapsed {} | ETA {}").format(self._download_state.progress_text, elapsed_text, eta_text)
return tr("{} / {} ({}%)\nElapsed {} | ETA {}").format(
self._download_state.downloaded_files,
self._download_state.total_files,
self._download_state.percent,
elapsed_text,
eta_text,
)
if self._is_visually_cancelling():
return tr("Stop request sent. The current transfer will wind down safely.")
if self._download_requested():
return tr("Preparing the selected regions for download.")
if self._selected_count() == 0:
return tr("No regions selected yet. Tap 'Whole U.S.' or pick specific states below to get started.")
gate_reason = self._download_gate_reason()
if gate_reason:
return gate_reason
return tr("Ready to download {}.").format(self._selection_preview_text())
Reference in New Issue View Git Blame Copy Permalink