import os import time import datetime from openpilot.common.time_helpers import system_time_valid from openpilot.selfdrive.ui.ui_state import ui_state from openpilot.system.ui.lib.application import gui_app from openpilot.system.ui.lib.multilang import tr, trn from openpilot.system.ui.widgets import Widget, DialogResult from openpilot.system.ui.widgets.confirm_dialog import ConfirmDialog from openpilot.system.ui.widgets.list_view import button_item, text_item, ListItem from openpilot.system.ui.widgets.option_dialog import MultiOptionDialog from openpilot.system.ui.widgets.scroller_tici import Scroller if gui_app.iqpilot_ui(): from openpilot.system.ui.iqpilot.widgets.list_view import button_item # TODO: remove this. updater fails to respond on startup if time is not correct UPDATED_TIMEOUT = 10 # seconds to wait for updated to respond BRAND_NAME = "IQ.Pilot" # Mapping updater internal states to translated display strings STATE_TO_DISPLAY_TEXT = { "checking...": tr("checking..."), "downloading...": tr("downloading..."), "finalizing update...": tr("finalizing update..."), } def format_updater_description(description: str | None) -> str: if not description: return BRAND_NAME cleaned = description.strip() lower = cleaned.lower() if lower.startswith("iqpilot"): cleaned = cleaned[len("iqpilot"):].lstrip(" -:/") if cleaned.lower().startswith(BRAND_NAME.lower()): return cleaned return f"{BRAND_NAME} {cleaned}" if cleaned else BRAND_NAME def time_ago(date: datetime.datetime | None) -> str: if not date: return tr("never") if not system_time_valid(): return date.strftime("%a %b %d %Y") now = datetime.datetime.now(datetime.UTC) if date.tzinfo is None: date = date.replace(tzinfo=datetime.UTC) diff_seconds = int((now - date).total_seconds()) if diff_seconds < 60: return tr("now") if diff_seconds < 3600: m = diff_seconds // 60 return trn("{} minute ago", "{} minutes ago", m).format(m) if diff_seconds < 86400: h = diff_seconds // 3600 return trn("{} hour ago", "{} hours ago", h).format(h) if diff_seconds < 604800: d = diff_seconds // 86400 return trn("{} day ago", "{} days ago", d).format(d) return date.strftime("%a %b %d %Y") class SoftwareLayout(Widget): def __init__(self): super().__init__() self._onroad_label = ListItem(lambda: tr("Updates are only downloaded while the car is off.")) self._version_item = text_item(lambda: tr("Current Version"), format_updater_description(ui_state.params.get("UpdaterCurrentDescription"))) self._download_btn = button_item(lambda: tr("Download"), lambda: tr("CHECK"), callback=self._on_download_update) # Install button is initially hidden self._install_btn = button_item(lambda: tr("Install Update"), lambda: tr("INSTALL"), callback=self._on_install_update) self._install_btn.set_visible(False) # Track waiting-for-updater transition to avoid brief re-enable while still idle self._waiting_for_updater = False self._waiting_start_ts: float = 0.0 # Branch switcher self._branch_btn = button_item(lambda: tr("Target Branch"), lambda: tr("SELECT"), callback=self._on_select_branch) self._branch_btn.set_visible(not ui_state.params.get_bool("IsTestedBranch")) self._branch_btn.action_item.set_value(ui_state.params.get("UpdaterTargetBranch") or "") self._branch_dialog: MultiOptionDialog | None = None # Git auth for private-branch updates (sits with the Target Branch section) self._auth_btn = button_item(lambda: tr("Git Auth"), lambda: tr("AUTH"), callback=self._on_auth_branch) self._auth_btn.set_visible(not ui_state.params.get_bool("IsTestedBranch")) self._scroller = Scroller([ self._onroad_label, self._version_item, self._download_btn, self._install_btn, self._branch_btn, self._auth_btn, button_item(lambda: tr("Uninstall"), lambda: tr("UNINSTALL"), callback=self._on_uninstall), ], line_separator=True, spacing=0) def show_event(self): self._refresh_auth_value() self._scroller.show_event() def _refresh_auth_value(self): try: from openpilot.common.git_creds import has_credentials configured = has_credentials() except Exception: configured = False self._auth_btn.action_item.set_value(tr("configured") if configured else "") def _render(self, rect): self._scroller.render(rect) def _update_state(self): # Show/hide onroad warning self._onroad_label.set_visible(ui_state.is_onroad()) # Update current version and release notes current_desc = format_updater_description(ui_state.params.get("UpdaterCurrentDescription")) current_release_notes = (ui_state.params.get("UpdaterCurrentReleaseNotes") or b"").decode("utf-8", "replace") self._version_item.action_item.set_text(current_desc) self._version_item.set_description(current_release_notes) # Update download button visibility and state self._download_btn.set_visible(ui_state.is_offroad()) updater_state = ui_state.params.get("UpdaterState") or "idle" failed_count = ui_state.params.get("UpdateFailedCount") or 0 fetch_available = ui_state.params.get_bool("UpdaterFetchAvailable") update_available = ui_state.params.get_bool("UpdateAvailable") if updater_state != "idle": # Updater responded self._waiting_for_updater = False self._download_btn.action_item.set_enabled(False) # Use the mapping, with a fallback to the original state string display_text = STATE_TO_DISPLAY_TEXT.get(updater_state, updater_state) self._download_btn.action_item.set_value(display_text) else: if failed_count > 0: self._download_btn.action_item.set_value(tr("failed to check for update")) self._download_btn.action_item.set_text(tr("CHECK")) elif fetch_available: self._download_btn.action_item.set_value(tr("update available")) self._download_btn.action_item.set_text(tr("DOWNLOAD")) else: last_update = ui_state.params.get("LastUpdateTime") if last_update: formatted = time_ago(last_update) self._download_btn.action_item.set_value(tr("up to date, last checked {}").format(formatted)) else: self._download_btn.action_item.set_value(tr("up to date, last checked never")) self._download_btn.action_item.set_text(tr("CHECK")) # If we've been waiting too long without a state change, reset state if self._waiting_for_updater and (time.monotonic() - self._waiting_start_ts > UPDATED_TIMEOUT): self._waiting_for_updater = False # Only enable if we're not waiting for updater to flip out of idle self._download_btn.action_item.set_enabled(not self._waiting_for_updater) # Update target branch button value current_branch = ui_state.params.get("UpdaterTargetBranch") or "" self._branch_btn.action_item.set_value(current_branch) # Update install button self._install_btn.set_visible(ui_state.is_offroad() and update_available) if update_available: new_desc = format_updater_description(ui_state.params.get("UpdaterNewDescription")) new_release_notes = (ui_state.params.get("UpdaterNewReleaseNotes") or b"").decode("utf-8", "replace") self._install_btn.action_item.set_text(tr("INSTALL")) self._install_btn.action_item.set_value(new_desc) self._install_btn.set_description(new_release_notes) # Enable install button for testing (like Qt showEvent) self._install_btn.action_item.set_enabled(True) else: self._install_btn.set_visible(False) def _on_download_update(self): # Check if we should start checking or start downloading self._download_btn.action_item.set_enabled(False) if self._download_btn.action_item.text == tr("CHECK"): # Start checking for updates self._waiting_for_updater = True self._waiting_start_ts = time.monotonic() os.system("pkill -SIGUSR1 -f system.updated.updated") else: # Start downloading self._waiting_for_updater = True self._waiting_start_ts = time.monotonic() os.system("pkill -SIGHUP -f system.updated.updated") def _on_uninstall(self): def handle_uninstall_confirmation(result): if result == DialogResult.CONFIRM: ui_state.params.put_bool("DoUninstall", True) dialog = ConfirmDialog(tr("Are you sure you want to uninstall?"), tr("Uninstall")) gui_app.set_modal_overlay(dialog, callback=handle_uninstall_confirmation) def _on_install_update(self): # Trigger reboot to install update self._install_btn.action_item.set_enabled(False) ui_state.params.put_bool("DoReboot", True) def _on_select_branch(self): # Get available branches and order current_git_branch = ui_state.params.get("GitBranch") or "" branches_str = ui_state.params.get("UpdaterAvailableBranches") or "" branches = [b for b in branches_str.split(",") if b] for b in [current_git_branch, "devel-staging", "devel", "nightly", "nightly-dev", "master"]: if b in branches: branches.remove(b) branches.insert(0, b) current_target = ui_state.params.get("UpdaterTargetBranch") or "" self._branch_dialog = MultiOptionDialog(tr("Select a branch"), branches, current_target) def handle_selection(result): # Confirmed selection if result == DialogResult.CONFIRM and self._branch_dialog is not None and self._branch_dialog.selection: selection = self._branch_dialog.selection ui_state.params.put("UpdaterTargetBranch", selection) self._branch_btn.action_item.set_value(selection) os.system("pkill -SIGUSR1 -f system.updated.updated") self._branch_dialog = None gui_app.set_modal_overlay(self._branch_dialog, callback=handle_selection) def _on_auth_branch(self): # Collect username then token; store encrypted and signal the updater to # re-check (refreshing the available-branch list for private repos). from openpilot.system.ui.iqpilot.widgets.list_view import open_text_prompt from openpilot.common import git_creds creds = git_creds.get_credentials() current_user = creds[0] if creds else "" def on_username(result, username): if result != DialogResult.CONFIRM: return def on_token(token_result, token): if token_result != DialogResult.CONFIRM: return git_creds.set_credentials(username, token) self._refresh_auth_value() os.system("pkill -SIGUSR1 -f system.updated.updated") open_text_prompt(tr("Git token / password"), tr("leave username and token blank to clear"), password=True, on_done=on_token) open_text_prompt(tr("Git username"), tr("for private branch updates"), initial=current_user, on_done=on_username)