diff --git a/selfdrive/ui/layouts/settings/firehose.py b/selfdrive/ui/layouts/settings/firehose.py index 74a8f317d5..b3db1fa5f0 100644 --- a/selfdrive/ui/layouts/settings/firehose.py +++ b/selfdrive/ui/layouts/settings/firehose.py @@ -2,7 +2,7 @@ import pyray as rl import time import threading -from openpilot.common.api import Api, api_get +from openpilot.common.api import api_get from openpilot.common.params import Params from openpilot.common.swaglog import cloudlog from openpilot.selfdrive.ui.ui_state import ui_state @@ -11,7 +11,7 @@ from openpilot.system.ui.lib.application import gui_app, FontWeight from openpilot.system.ui.lib.scroll_panel import GuiScrollPanel from openpilot.system.ui.lib.wrap_text import wrap_text from openpilot.system.ui.widgets import Widget - +from openpilot.selfdrive.ui.lib.api_helpers import get_token TITLE = "Firehose Mode" DESCRIPTION = ( @@ -163,7 +163,7 @@ class FirehoseLayout(Widget): dongle_id = self.params.get("DongleId") if not dongle_id or dongle_id == UNREGISTERED_DONGLE_ID: return - identity_token = Api(dongle_id).get_token() + identity_token = get_token(dongle_id) response = api_get(f"v1/devices/{dongle_id}/firehose_stats", access_token=identity_token) if response.status_code == 200: data = response.json() diff --git a/selfdrive/ui/lib/api_helpers.py b/selfdrive/ui/lib/api_helpers.py new file mode 100644 index 0000000000..b83efedb60 --- /dev/null +++ b/selfdrive/ui/lib/api_helpers.py @@ -0,0 +1,14 @@ +import time +from functools import lru_cache +from openpilot.common.api import Api + +TOKEN_EXPIRY_HOURS = 2 + + +@lru_cache(maxsize=1) +def _get_token(dongle_id: str, t: int): + return Api(dongle_id).get_token(expiry_hours=TOKEN_EXPIRY_HOURS) + + +def get_token(dongle_id: str): + return _get_token(dongle_id, int(time.monotonic() / (TOKEN_EXPIRY_HOURS / 2 * 60 * 60))) diff --git a/selfdrive/ui/lib/prime_state.py b/selfdrive/ui/lib/prime_state.py index a1b2472f30..be2132c1b7 100644 --- a/selfdrive/ui/lib/prime_state.py +++ b/selfdrive/ui/lib/prime_state.py @@ -2,14 +2,12 @@ from enum import IntEnum import os import threading import time -from functools import lru_cache -from openpilot.common.api import Api, api_get +from openpilot.common.api import api_get from openpilot.common.params import Params from openpilot.common.swaglog import cloudlog from openpilot.system.athena.registration import UNREGISTERED_DONGLE_ID - -TOKEN_EXPIRY_HOURS = 2 +from openpilot.selfdrive.ui.lib.api_helpers import get_token class PrimeType(IntEnum): @@ -23,11 +21,6 @@ class PrimeType(IntEnum): PURPLE = 5, -@lru_cache(maxsize=1) -def get_token(dongle_id: str, t: int): - return Api(dongle_id).get_token(expiry_hours=TOKEN_EXPIRY_HOURS) - - class PrimeState: FETCH_INTERVAL = 5.0 # seconds between API calls API_TIMEOUT = 10.0 # seconds for API requests @@ -57,15 +50,13 @@ class PrimeState: return try: - identity_token = get_token(dongle_id, int(time.monotonic() / (TOKEN_EXPIRY_HOURS / 2 * 60 * 60))) + identity_token = get_token(dongle_id) response = api_get(f"v1.1/devices/{dongle_id}", timeout=self.API_TIMEOUT, access_token=identity_token) if response.status_code == 200: data = response.json() is_paired = data.get("is_paired", False) prime_type = data.get("prime_type", 0) self.set_type(PrimeType(prime_type) if is_paired else PrimeType.UNPAIRED) - elif response.status_code == 401: - get_token.cache_clear() except Exception as e: cloudlog.error(f"Failed to fetch prime status: {e}")