From dc654b439a946972b2f476465407fb2e743f576a Mon Sep 17 00:00:00 2001 From: Maxime Desroches Date: Mon, 1 Dec 2025 20:48:04 -0800 Subject: [PATCH] Revert "Fix raylib ui spamming API calls (#36700)" (#36744) This reverts commit 26261387f8b4188951d107567a067d22cf87e5e3. --- common/params_keys.h | 2 +- common/tests/test_params.py | 4 +- selfdrive/ui/lib/api_helpers.py | 93 +------------------ selfdrive/ui/lib/prime_state.py | 75 +++++++++++---- .../ui/mici/layouts/settings/firehose.py | 65 +++++++++---- 5 files changed, 108 insertions(+), 131 deletions(-) diff --git a/common/params_keys.h b/common/params_keys.h index c496ad285..d6104e749 100644 --- a/common/params_keys.h +++ b/common/params_keys.h @@ -10,7 +10,7 @@ inline static std::unordered_map keys = { {"AdbEnabled", {PERSISTENT, BOOL}}, {"AlwaysOnDM", {PERSISTENT, BOOL}}, {"ApiCache_Device", {PERSISTENT, STRING}}, - {"ApiCache_FirehoseStats", {PERSISTENT, STRING}}, + {"ApiCache_FirehoseStats", {PERSISTENT, JSON}}, {"AssistNowToken", {PERSISTENT, STRING}}, {"AthenadPid", {PERSISTENT, INT}}, {"AthenadUploadQueue", {PERSISTENT, JSON}}, diff --git a/common/tests/test_params.py b/common/tests/test_params.py index e0f213e02..592bf2c4b 100644 --- a/common/tests/test_params.py +++ b/common/tests/test_params.py @@ -123,8 +123,8 @@ class TestParams: def test_params_get_type(self): # json - self.params.put("LiveParameters", {"a": 0}) - assert self.params.get("LiveParameters") == {"a": 0} + self.params.put("ApiCache_FirehoseStats", {"a": 0}) + assert self.params.get("ApiCache_FirehoseStats") == {"a": 0} # int self.params.put("BootCount", 1441) diff --git a/selfdrive/ui/lib/api_helpers.py b/selfdrive/ui/lib/api_helpers.py index 31e844dd0..8ed1c22a6 100644 --- a/selfdrive/ui/lib/api_helpers.py +++ b/selfdrive/ui/lib/api_helpers.py @@ -1,13 +1,7 @@ import time -import threading -from collections.abc import Callable from functools import lru_cache - -from openpilot.common.api import Api, api_get -from openpilot.common.params import Params -from openpilot.common.swaglog import cloudlog +from openpilot.common.api import Api from openpilot.common.time_helpers import system_time_valid -from openpilot.system.athena.registration import UNREGISTERED_DONGLE_ID TOKEN_EXPIRY_HOURS = 2 @@ -22,88 +16,3 @@ def _get_token(dongle_id: str, t: int): def get_token(dongle_id: str): return _get_token(dongle_id, int(time.monotonic() / (TOKEN_EXPIRY_HOURS / 2 * 60 * 60))) - - -class RequestRepeater: - API_TIMEOUT = 10.0 # seconds for API requests - SLEEP_INTERVAL = 0.5 # seconds to sleep between checks in the worker thread - - def __init__(self, dongle_id: str, request_route: str, period: int, cache_key: str | None = None): - self._dongle_id = dongle_id - self._request_route = request_route - self._period = period # seconds - self._cache_key = cache_key - - self._request_done_callbacks: list[Callable[[str, bool], None]] = [] - self._prev_response_text = None - self._running = False - self._thread = None - self._params = Params() - - if self._cache_key is not None: - # Cache successful responses to params - def cache_response(response: str, success: bool): - if success and response != self._prev_response_text: - self._params.put(self._cache_key, response) - self._prev_response_text = response - - self.add_request_done_callback(cache_response) - - def add_request_done_callback(self, callback: Callable[[str, bool], None]): - self._request_done_callbacks.append(callback) - - def _do_callbacks(self, response_text: str, success: bool): - for callback in self._request_done_callbacks: - try: - callback(response_text, success) - except Exception as e: - cloudlog.error(f"RequestRepeater callback error: {e}") - - def load_cache(self): - # call callbacks with cached response - if self._cache_key is not None: - self._prev_response_text = self._params.get(self._cache_key) - if self._prev_response_text: - self._do_callbacks(self._prev_response_text, True) - - def start(self): - if self._thread and self._thread.is_alive(): - return - self._running = True - self._thread = threading.Thread(target=self._worker_thread, daemon=True) - self._thread.start() - - def stop(self): - self._running = False - if self._thread and self._thread.is_alive(): - self._thread.join(timeout=1.0) - - def _worker_thread(self): - # Avoid circular imports - from openpilot.selfdrive.ui.ui_state import ui_state, device - - while self._running: - # Don't run when device is asleep or onroad - if not ui_state.started and device.awake: - self._send_request() - - for _ in range(int(self._period / self.SLEEP_INTERVAL)): - if not self._running: - break - time.sleep(self.SLEEP_INTERVAL) - - def _send_request(self): - if not self._dongle_id or self._dongle_id == UNREGISTERED_DONGLE_ID: - return - - try: - identity_token = get_token(self._dongle_id) - response = api_get(self._request_route, timeout=self.API_TIMEOUT, access_token=identity_token) - self._do_callbacks(response.text, 200 <= response.status_code < 300) - - except Exception as e: - cloudlog.error(f"Failed to send request to {self._request_route}: {e}") - self._do_callbacks("", False) - - def __del__(self): - self.stop() diff --git a/selfdrive/ui/lib/prime_state.py b/selfdrive/ui/lib/prime_state.py index f3adebf4b..fc72b4f9c 100644 --- a/selfdrive/ui/lib/prime_state.py +++ b/selfdrive/ui/lib/prime_state.py @@ -1,10 +1,13 @@ from enum import IntEnum import os -import json +import threading +import time +from openpilot.common.api import api_get from openpilot.common.params import Params from openpilot.common.swaglog import cloudlog -from openpilot.selfdrive.ui.lib.api_helpers import RequestRepeater +from openpilot.system.athena.registration import UNREGISTERED_DONGLE_ID +from openpilot.selfdrive.ui.lib.api_helpers import get_token class PrimeType(IntEnum): @@ -19,14 +22,17 @@ class PrimeType(IntEnum): class PrimeState: + FETCH_INTERVAL = 5.0 # seconds between API calls + API_TIMEOUT = 10.0 # seconds for API requests + SLEEP_INTERVAL = 0.5 # seconds to sleep between checks in the worker thread + def __init__(self): self._params = Params() + self._lock = threading.Lock() self.prime_type: PrimeType = self._load_initial_state() - dongle_id = self._params.get("DongleId") - self._request_repeater = RequestRepeater(dongle_id, f"v1.1/devices/{dongle_id}", 5, "ApiCache_Device") - self._request_repeater.add_request_done_callback(self._handle_reply) - self._request_repeater.load_cache() # sets prime_type from API response cache + self._running = False + self._thread = None def _load_initial_state(self) -> PrimeType: prime_type_str = os.getenv("PRIME_TYPE") or self._params.get("PrimeType") @@ -37,32 +43,61 @@ class PrimeState: pass return PrimeType.UNKNOWN - def _handle_reply(self, response: str, success: bool): - if not success: + def _fetch_prime_status(self) -> None: + dongle_id = self._params.get("DongleId") + if not dongle_id or dongle_id == UNREGISTERED_DONGLE_ID: return try: - data = json.loads(response) - 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) + 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) except Exception as e: cloudlog.error(f"Failed to fetch prime status: {e}") def set_type(self, prime_type: PrimeType) -> None: - if prime_type != self.prime_type: - self.prime_type = prime_type - self._params.put("PrimeType", int(prime_type)) - cloudlog.info(f"Prime type updated to {prime_type}") + with self._lock: + if prime_type != self.prime_type: + self.prime_type = prime_type + self._params.put("PrimeType", int(prime_type)) + cloudlog.info(f"Prime type updated to {prime_type}") + + def _worker_thread(self) -> None: + while self._running: + self._fetch_prime_status() + + for _ in range(int(self.FETCH_INTERVAL / self.SLEEP_INTERVAL)): + if not self._running: + break + time.sleep(self.SLEEP_INTERVAL) def start(self) -> None: - self._request_repeater.start() + if self._thread and self._thread.is_alive(): + return + self._running = True + self._thread = threading.Thread(target=self._worker_thread, daemon=True) + self._thread.start() + + def stop(self) -> None: + self._running = False + if self._thread and self._thread.is_alive(): + self._thread.join(timeout=1.0) def get_type(self) -> PrimeType: - return self.prime_type + with self._lock: + return self.prime_type def is_prime(self) -> bool: - return bool(self.prime_type > PrimeType.NONE) + with self._lock: + return bool(self.prime_type > PrimeType.NONE) def is_paired(self) -> bool: - return self.prime_type > PrimeType.UNPAIRED + with self._lock: + return self.prime_type > PrimeType.UNPAIRED + + def __del__(self): + self.stop() diff --git a/selfdrive/ui/mici/layouts/settings/firehose.py b/selfdrive/ui/mici/layouts/settings/firehose.py index ff75e6f8e..dcb4b6c84 100644 --- a/selfdrive/ui/mici/layouts/settings/firehose.py +++ b/selfdrive/ui/mici/layouts/settings/firehose.py @@ -1,15 +1,18 @@ +import threading +import time import pyray as rl -import json +from openpilot.common.api import api_get from openpilot.common.params import Params from openpilot.common.swaglog import cloudlog +from openpilot.selfdrive.ui.lib.api_helpers import get_token from openpilot.selfdrive.ui.ui_state import ui_state +from openpilot.system.athena.registration import UNREGISTERED_DONGLE_ID from openpilot.system.ui.lib.application import gui_app, FontWeight, FONT_SCALE from openpilot.system.ui.lib.wrap_text import wrap_text from openpilot.system.ui.lib.scroll_panel2 import GuiScrollPanel2 from openpilot.system.ui.lib.multilang import tr, trn, tr_noop from openpilot.system.ui.widgets import Widget, NavWidget -from openpilot.selfdrive.ui.lib.api_helpers import RequestRepeater TITLE = tr_noop("Firehose Mode") DESCRIPTION = tr_noop( @@ -31,37 +34,47 @@ FAQ_ITEMS = [ class FirehoseLayoutBase(Widget): + PARAM_KEY = "ApiCache_FirehoseStats" GREEN = rl.Color(46, 204, 113, 255) RED = rl.Color(231, 76, 60, 255) GRAY = rl.Color(68, 68, 68, 255) LIGHT_GRAY = rl.Color(228, 228, 228, 255) + UPDATE_INTERVAL = 30 # seconds def __init__(self): super().__init__() - self._segment_count = 0 + self._params = Params() + self._segment_count = self._get_segment_count() + self._scroll_panel = GuiScrollPanel2(horizontal=False) self._content_height = 0 - dongle_id = Params().get("DongleId") - self._request_repeater = RequestRepeater(dongle_id, f"v1/devices/{dongle_id}/firehose_stats", 30, "ApiCache_FirehoseStats") - self._request_repeater.add_request_done_callback(self._handle_reply) - self._request_repeater.load_cache() - self._request_repeater.start() - - def _handle_reply(self, response: str, success: bool): - if not success: - return + self._running = True + self._update_thread = threading.Thread(target=self._update_loop, daemon=True) + self._update_thread.start() + def __del__(self): + self._running = False try: - data = json.loads(response) - self._segment_count = data.get("firehose", 0) - except Exception as e: - cloudlog.error(f"Failed to fetch firehose stats: {e}") + if self._update_thread and self._update_thread.is_alive(): + self._update_thread.join(timeout=1.0) + except Exception: + pass def show_event(self): super().show_event() self._scroll_panel.set_offset(0) + def _get_segment_count(self) -> int: + stats = self._params.get(self.PARAM_KEY) + if not stats: + return 0 + try: + return int(stats.get("firehose", 0)) + except Exception: + cloudlog.exception(f"Failed to decode firehose stats: {stats}") + return 0 + def _render(self, rect: rl.Rectangle): # compute total content height for scrolling content_height = self._measure_content_height(rect) @@ -184,6 +197,26 @@ class FirehoseLayoutBase(Widget): else: return tr("INACTIVE: connect to an unmetered network"), self.RED + def _fetch_firehose_stats(self): + try: + dongle_id = self._params.get("DongleId") + if not dongle_id or dongle_id == UNREGISTERED_DONGLE_ID: + return + 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() + self._segment_count = data.get("firehose", 0) + self._params.put(self.PARAM_KEY, data) + except Exception as e: + cloudlog.error(f"Failed to fetch firehose stats: {e}") + + def _update_loop(self): + while self._running: + if not ui_state.started: + self._fetch_firehose_stats() + time.sleep(self.UPDATE_INTERVAL) + class FirehoseLayout(FirehoseLayoutBase, NavWidget): BACK_TOUCH_AREA_PERCENTAGE = 0.1