From 26261387f8b4188951d107567a067d22cf87e5e3 Mon Sep 17 00:00:00 2001 From: Shane Smiskol Date: Thu, 27 Nov 2025 01:37:16 -0800 Subject: [PATCH] Fix raylib ui spamming API calls (#36700) * intern * start * move * common caching * use constant for slep * works * add gating back * clean up * more * match cache logic * hate this circular * not needed since sync * no need for lock? * even qt had something like _load_initial_state for tests, keep * clean up * clean up * clean up * loading json as string works, else it will fail to parse json, catch that and log, and next api call will overwrite * move over firehose * clean up * fix test * no * flip * more * match qt * consistent * clean up * cmt * fix test! --- 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, 131 insertions(+), 108 deletions(-) diff --git a/common/params_keys.h b/common/params_keys.h index bf410825f1..fc7f410e40 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, JSON}}, + {"ApiCache_FirehoseStats", {PERSISTENT, STRING}}, {"AssistNowToken", {PERSISTENT, STRING}}, {"AthenadPid", {PERSISTENT, INT}}, {"AthenadUploadQueue", {PERSISTENT, JSON}}, diff --git a/common/tests/test_params.py b/common/tests/test_params.py index 592bf2c4b2..e0f213e02c 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("ApiCache_FirehoseStats", {"a": 0}) - assert self.params.get("ApiCache_FirehoseStats") == {"a": 0} + self.params.put("LiveParameters", {"a": 0}) + assert self.params.get("LiveParameters") == {"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 8ed1c22a63..31e844dd0a 100644 --- a/selfdrive/ui/lib/api_helpers.py +++ b/selfdrive/ui/lib/api_helpers.py @@ -1,7 +1,13 @@ import time +import threading +from collections.abc import Callable from functools import lru_cache -from openpilot.common.api import Api + +from openpilot.common.api import Api, api_get +from openpilot.common.params import Params +from openpilot.common.swaglog import cloudlog from openpilot.common.time_helpers import system_time_valid +from openpilot.system.athena.registration import UNREGISTERED_DONGLE_ID TOKEN_EXPIRY_HOURS = 2 @@ -16,3 +22,88 @@ 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 fc72b4f9c6..f3adebf4b6 100644 --- a/selfdrive/ui/lib/prime_state.py +++ b/selfdrive/ui/lib/prime_state.py @@ -1,13 +1,10 @@ from enum import IntEnum import os -import threading -import time +import json -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 -from openpilot.selfdrive.ui.lib.api_helpers import get_token +from openpilot.selfdrive.ui.lib.api_helpers import RequestRepeater class PrimeType(IntEnum): @@ -22,17 +19,14 @@ 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() - self._running = False - self._thread = None + 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 def _load_initial_state(self) -> PrimeType: prime_type_str = os.getenv("PRIME_TYPE") or self._params.get("PrimeType") @@ -43,61 +37,32 @@ class PrimeState: pass return PrimeType.UNKNOWN - def _fetch_prime_status(self) -> None: - dongle_id = self._params.get("DongleId") - if not dongle_id or dongle_id == UNREGISTERED_DONGLE_ID: + def _handle_reply(self, response: str, success: bool): + if not success: return try: - 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) + 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) except Exception as e: cloudlog.error(f"Failed to fetch prime status: {e}") def set_type(self, prime_type: PrimeType) -> None: - 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) + 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 start(self) -> None: - 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) + self._request_repeater.start() def get_type(self) -> PrimeType: - with self._lock: - return self.prime_type + return self.prime_type def is_prime(self) -> bool: - with self._lock: - return bool(self.prime_type > PrimeType.NONE) + return bool(self.prime_type > PrimeType.NONE) def is_paired(self) -> bool: - with self._lock: - return self.prime_type > PrimeType.UNPAIRED - - def __del__(self): - self.stop() + return self.prime_type > PrimeType.UNPAIRED diff --git a/selfdrive/ui/mici/layouts/settings/firehose.py b/selfdrive/ui/mici/layouts/settings/firehose.py index f2f8bcb1cb..5cc43322d0 100644 --- a/selfdrive/ui/mici/layouts/settings/firehose.py +++ b/selfdrive/ui/mici/layouts/settings/firehose.py @@ -1,18 +1,15 @@ -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( @@ -34,47 +31,37 @@ 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._params = Params() - self._segment_count = self._get_segment_count() - + self._segment_count = 0 self._scroll_panel = GuiScrollPanel2(horizontal=False) self._content_height = 0 - self._running = True - self._update_thread = threading.Thread(target=self._update_loop, daemon=True) - self._update_thread.start() + 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 - def __del__(self): - self._running = False try: - if self._update_thread and self._update_thread.is_alive(): - self._update_thread.join(timeout=1.0) - except Exception: - pass + data = json.loads(response) + self._segment_count = data.get("firehose", 0) + except Exception as e: + cloudlog.error(f"Failed to fetch firehose stats: {e}") 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) @@ -197,26 +184,6 @@ 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