mirror of
https://github.com/firestar5683/StarPilot.git
synced 2026-08-21 08:14:00 +08:00
params: auto decode based on type (#35794)
* type * test * more * might as well use this * one more * live * athena * b * also * more * now * ah * pigeon
This commit is contained in:
@@ -128,7 +128,7 @@ class Car:
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
secoc_key = self.params.get("SecOCKey", encoding='utf8')
|
||||
secoc_key = self.params.get("SecOCKey")
|
||||
if secoc_key is not None:
|
||||
saved_secoc_key = bytes.fromhex(secoc_key.strip())
|
||||
if len(saved_secoc_key) == 16:
|
||||
|
||||
@@ -111,7 +111,7 @@ class TestAlerts:
|
||||
alert = copy.copy(self.offroad_alerts[a])
|
||||
set_offroad_alert(a, True)
|
||||
alert['extra'] = ''
|
||||
assert alert == params.get(a, encoding='utf8')
|
||||
assert alert == params.get(a)
|
||||
|
||||
# then delete it
|
||||
set_offroad_alert(a, False)
|
||||
@@ -125,6 +125,6 @@ class TestAlerts:
|
||||
alert = self.offroad_alerts[a]
|
||||
set_offroad_alert(a, True, extra_text="a"*i)
|
||||
|
||||
written_alert = params.get(a, encoding='utf8')
|
||||
written_alert = params.get(a)
|
||||
assert "a"*i == written_alert['extra']
|
||||
assert alert["text"] == written_alert['text']
|
||||
|
||||
@@ -147,7 +147,7 @@ class TestOnroad:
|
||||
while not sm.seen['carState']:
|
||||
sm.update(1000)
|
||||
|
||||
route = params.get("CurrentRoute", encoding="utf-8")
|
||||
route = params.get("CurrentRoute")
|
||||
assert route is not None
|
||||
|
||||
segs = list(Path(Paths.log_root()).glob(f"{route}--*"))
|
||||
|
||||
@@ -105,7 +105,7 @@ class TestUpdated:
|
||||
ret = None
|
||||
start_time = time.monotonic()
|
||||
while ret is None:
|
||||
ret = self.params.get(key, encoding='utf8')
|
||||
ret = self.params.get(key)
|
||||
if time.monotonic() - start_time > timeout:
|
||||
break
|
||||
time.sleep(0.01)
|
||||
@@ -162,7 +162,7 @@ class TestUpdated:
|
||||
|
||||
def _check_update_state(self, update_available):
|
||||
# make sure LastUpdateTime is recent
|
||||
last_update_time = self._read_param("LastUpdateTime", encoding="utf-8")
|
||||
last_update_time = self._read_param("LastUpdateTime")
|
||||
td = datetime.datetime.now(datetime.UTC).replace(tzinfo=None) - last_update_time
|
||||
assert td.total_seconds() < 10
|
||||
self.params.remove("LastUpdateTime")
|
||||
|
||||
@@ -210,5 +210,5 @@ class HomeLayout(Widget):
|
||||
|
||||
def _get_version_text(self) -> str:
|
||||
brand = "openpilot"
|
||||
description = self.params.get("UpdaterCurrentDescription", encoding='utf-8')
|
||||
description = self.params.get("UpdaterCurrentDescription")
|
||||
return f"{brand} {description}" if description else brand
|
||||
|
||||
@@ -41,8 +41,8 @@ class DeviceLayout(Widget):
|
||||
self._scroller = Scroller(items, line_separator=True, spacing=0)
|
||||
|
||||
def _initialize_items(self):
|
||||
dongle_id = self._params.get("DongleId", encoding="utf-8") or "N/A"
|
||||
serial = self._params.get("HardwareSerial") or "N/A"
|
||||
dongle_id = self._params.get("DongleId", default="N/A")
|
||||
serial = self._params.get("HardwareSerial", default="N/A")
|
||||
|
||||
items = [
|
||||
text_item("Dongle ID", dongle_id),
|
||||
|
||||
@@ -51,7 +51,7 @@ class FirehoseLayout(Widget):
|
||||
self.last_update_time = 0
|
||||
|
||||
def _get_segment_count(self) -> int:
|
||||
stats = self.params.get(self.PARAM_KEY, encoding='utf8')
|
||||
stats = self.params.get(self.PARAM_KEY)
|
||||
if not stats:
|
||||
return 0
|
||||
try:
|
||||
@@ -161,7 +161,7 @@ class FirehoseLayout(Widget):
|
||||
|
||||
def _fetch_firehose_stats(self):
|
||||
try:
|
||||
dongle_id = self.params.get("DongleId", encoding='utf8')
|
||||
dongle_id = self.params.get("DongleId")
|
||||
if not dongle_id or dongle_id == UNREGISTERED_DONGLE_ID:
|
||||
return
|
||||
identity_token = Api(dongle_id).get_token()
|
||||
|
||||
@@ -54,7 +54,7 @@ class TogglesLayout(Widget):
|
||||
buttons=["Aggressive", "Standard", "Relaxed"],
|
||||
button_width=255,
|
||||
callback=self._set_longitudinal_personality,
|
||||
selected_index=int(self._params.get("LongitudinalPersonality") or 0),
|
||||
selected_index=self._params.get("LongitudinalPersonality", default=0),
|
||||
icon="speed_limit.png"
|
||||
),
|
||||
toggle_item(
|
||||
|
||||
@@ -35,7 +35,7 @@ class PrimeState:
|
||||
self.start()
|
||||
|
||||
def _load_initial_state(self) -> PrimeType:
|
||||
prime_type_str = os.getenv("PRIME_TYPE") or self._params.get("PrimeType", encoding='utf8')
|
||||
prime_type_str = os.getenv("PRIME_TYPE") or self._params.get("PrimeType")
|
||||
try:
|
||||
if prime_type_str is not None:
|
||||
return PrimeType(int(prime_type_str))
|
||||
@@ -44,7 +44,7 @@ class PrimeState:
|
||||
return PrimeType.UNKNOWN
|
||||
|
||||
def _fetch_prime_status(self) -> None:
|
||||
dongle_id = self._params.get("DongleId", encoding='utf8')
|
||||
dongle_id = self._params.get("DongleId")
|
||||
if not dongle_id or dongle_id == UNREGISTERED_DONGLE_ID:
|
||||
return
|
||||
|
||||
|
||||
@@ -292,7 +292,7 @@ class UpdateAlert(AbstractAlert):
|
||||
def refresh(self) -> bool:
|
||||
update_available: bool = self.params.get_bool("UpdateAvailable")
|
||||
if update_available:
|
||||
self.release_notes = self.params.get("UpdaterNewReleaseNotes", encoding='utf-8')
|
||||
self.release_notes = self.params.get("UpdaterNewReleaseNotes")
|
||||
self._cached_content_height = 0
|
||||
|
||||
return update_available
|
||||
|
||||
@@ -23,7 +23,7 @@ class PairingDialog:
|
||||
|
||||
def _get_pairing_url(self) -> str:
|
||||
try:
|
||||
dongle_id = self.params.get("DongleId", encoding='utf8') or ""
|
||||
dongle_id = self.params.get("DongleId", default="")
|
||||
token = Api(dongle_id).get_token()
|
||||
except Exception as e:
|
||||
cloudlog.warning(f"Failed to get pairing token: {e}")
|
||||
|
||||
Reference in New Issue
Block a user