mawei/StarPilot
1
0
Fork 0
mirror of https://github.com/firestar5683/StarPilot.git synced 2026-07-15 14:22:11 +08:00
Code Issues Packages Projects Releases Wiki Activity

Revert "Fix raylib ui spamming API calls (#36700)" (#36744)

Browse Source 
This reverts commit 26261387f8.
This commit is contained in:
Maxime Desroches
2025-12-01 20:48:04 -08:00
committed by GitHub
parent 693c83f74c
commit dc654b439a
5 changed files with 108 additions and 131 deletions
Download Patch File Download Diff File Expand all files Collapse all files
common/params_keys.h
+1 -1
View File
@@ -10,7 +10,7 @@ inline static std::unordered_map<std::string, ParamKeyAttributes> 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}},
common/tests/test_params.py
+2 -2
View File
@@ -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)
selfdrive/ui/lib/api_helpers.py
+1 -92
View File
@@ -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()
selfdrive/ui/lib/prime_state.py
+55 -20
View File
@@ -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()
selfdrive/ui/mici/layouts/settings/firehose.py
+49 -16
View File
@@ -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