ui: sunnylink client-side implementation (#1488)

* param to control stock vs sp ui

* init styles

* SP Toggles

* Lint

* optimizations

* sp raylib preview

* fix callback

* fix ui preview

* sunnylink state

* introducing ui_state_sp for py

* poll from ui_state_sp

* cloudlog & ruff

* param to control stock vs sp ui

* better

* better padding

* this

* listitem -> listitemsp

* add show_description method

* remove padding from line separator.
like, WHY? 😩😩

* ui: `GuiApplicationExt`

* add to readme

* use gui_app.sunnypilot_ui()

* use gui_app.sunnypilot_ui()

* fetch only when connected to network

* final

---------

Co-authored-by: Jason Wen <haibin.wen3@gmail.com>

selfdrive/ui/sunnypilot/ui_state.py

@@ -6,6 +6,7 @@ See the LICENSE.md file in the root directory for more details.
 """
 from cereal import messaging, custom
 from openpilot.common.params import Params
+from openpilot.sunnypilot.sunnylink.sunnylink_state import SunnylinkState
 
 
 class UIStateSP:
@@ -16,8 +17,10 @@ class UIStateSP:
       "gpsLocation", "liveTorqueParameters", "carStateSP", "liveMapDataSP", "carParamsSP"
     ]
 
+    self.sunnylink_state = SunnylinkState()
+
   def update(self) -> None:
-    pass
+    self.sunnylink_state.start()
 
   def update_params(self) -> None:
     CP_SP_bytes = self.params.get("CarParamsSPPersistent")

sunnypilot/sunnylink/sunnylink_state.py

@@ -0,0 +1,205 @@
+"""
+Copyright (c) 2021-, Haibin Wen, sunnypilot, and a number of other contributors.
+
+This file is part of sunnypilot and is licensed under the MIT License.
+See the LICENSE.md file in the root directory for more details.
+"""
+from enum import IntEnum
+import threading
+import time
+import json
+
+from cereal import messaging
+from openpilot.common.params import Params
+from openpilot.common.swaglog import cloudlog
+from openpilot.sunnypilot.sunnylink.api import UNREGISTERED_SUNNYLINK_DONGLE_ID, SunnylinkApi
+
+
+class RoleType(IntEnum):
+  READONLY = 0
+  SPONSOR = 1
+  ADMIN = 2
+
+
+class SponsorTier(IntEnum):
+  FREE = 0
+  NOVICE = 1
+  SUPPORTER = 2
+  CONTRIBUTOR = 3
+  BENEFACTOR = 4
+  GUARDIAN = 5
+
+
+class User:
+  device_id: str
+  user_id: str
+  created_at: int
+  updated_at: int
+  token_hash: str
+
+  def __init__(self, json_data):
+    self.device_id = json_data.get("device_id")
+    self.user_id = json_data.get("user_id")
+    self.created_at = json_data.get("created_at")
+    self.updated_at = json_data.get("updated_at")
+    self.token_hash = json_data.get("token_hash")
+
+
+class Role:
+  role_type: str
+  role_tier: str
+
+  def __init__(self, json_data):
+    self.role_type = json_data.get("role_type")
+    self.role_tier = json_data.get("role_tier")
+
+
+def _parse_roles(roles: str) -> list[Role]:
+  lst_roles = []
+  try:
+    roles_list = json.loads(roles)
+    for r in roles_list:
+      try:
+        role = Role(r)
+        lst_roles.append(role)
+      except Exception as e:
+        cloudlog.exception(f"Failed to parse role {r}: {e}")
+    return lst_roles
+  except Exception as e:
+    cloudlog.exception(f"Error parsing roles: {e}")
+    return []
+
+
+def _parse_users(users: str) -> list[User]:
+  lst_users = []
+  try:
+    users_list = json.loads(users)
+    for u in users_list:
+      try:
+        user = User(u)
+        lst_users.append(user)
+      except Exception as e:
+        cloudlog.exception(f"Failed to parse user {u}: {e}")
+    return lst_users
+  except Exception as e:
+    cloudlog.exception(f"Error parsing users: {e}")
+    return []
+
+
+class SunnylinkState:
+  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
+  NOT_PAIRED_USERNAMES = ["unregisteredsponsor", "temporarysponsor"]
+
+  def __init__(self):
+    self._params = Params()
+    self._lock = threading.Lock()
+    self._running = False
+    self._thread = None
+    self._sm = messaging.SubMaster(['deviceState'])
+
+    self._roles: list[Role] = []
+    self._users: list[User] = []
+    self.sponsor_tier: SponsorTier = SponsorTier.FREE
+    self.sunnylink_dongle_id = self._params.get("SunnylinkDongleId")
+    self._api = SunnylinkApi(self.sunnylink_dongle_id)
+
+    self._load_initial_state()
+
+  def _load_initial_state(self) -> None:
+    roles_cache = self._params.get("SunnylinkCache_Roles")
+    users_cache = self._params.get("SunnylinkCache_Users")
+    if roles_cache is not None:
+      self._roles = _parse_roles(roles_cache)
+      self.sponsor_tier = self._get_highest_tier()
+    if users_cache is not None:
+      self._users = _parse_users(users_cache)
+
+  def _get_highest_tier(self) -> SponsorTier:
+    role_tier = SponsorTier.FREE
+    for role in self._roles:
+      try:
+        if RoleType[role.role_type.upper()] == RoleType.SPONSOR:
+          role_tier = max(role_tier, SponsorTier[role.role_tier.upper()])
+      except Exception as e:
+        cloudlog.exception(f"Error parsing role {role}: {e} for dongle id {self.sunnylink_dongle_id}")
+    return role_tier
+
+  def _fetch_roles(self) -> None:
+    if not self.sunnylink_dongle_id or self.sunnylink_dongle_id == UNREGISTERED_SUNNYLINK_DONGLE_ID:
+      return
+
+    try:
+      token = self._api.get_token()
+      response = self._api.api_get(f"device/{self.sunnylink_dongle_id}/roles", method='GET', access_token=token)
+      if response.status_code == 200:
+        self._roles = _parse_roles(response.text)
+        self._params.put("SunnylinkCache_Roles", response.text)
+        sponsor_tier = self._get_highest_tier()
+        with self._lock:
+          if sponsor_tier != self.sponsor_tier:
+            self.sponsor_tier = sponsor_tier
+            cloudlog.info(f"Sunnylink sponsor tier updated to {sponsor_tier.name}")
+    except Exception as e:
+      cloudlog.exception(f"Failed to fetch sunnylink roles: {e} for dongle id {self.sunnylink_dongle_id}")
+
+  def _fetch_users(self) -> None:
+    if not self.sunnylink_dongle_id or self.sunnylink_dongle_id == UNREGISTERED_SUNNYLINK_DONGLE_ID:
+      return
+
+    try:
+      token = self._api.get_token()
+      response = self._api.api_get(f"device/{self.sunnylink_dongle_id}/users", method='GET', access_token=token)
+      if response.status_code == 200:
+        users = response.text
+        self._params.put("SunnylinkCache_Users", users)
+        with self._lock:
+          _parse_users(users)
+    except Exception as e:
+      cloudlog.exception(f"Failed to fetch sunnylink users: {e} for dongle id {self.sunnylink_dongle_id}")
+
+  def _worker_thread(self) -> None:
+    while self._running:
+      if self.is_connected():
+        self._fetch_roles()
+        self._fetch_users()
+
+      for _ in range(int(self.FETCH_INTERVAL / self.SLEEP_INTERVAL)):
+        if not self._running:
+          break
+        time.sleep(self.SLEEP_INTERVAL)
+
+  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)
+
+  def get_sponsor_tier(self) -> SponsorTier:
+    with self._lock:
+      return self.sponsor_tier
+
+  def is_sponsor(self) -> bool:
+    with self._lock:
+      is_sponsor = any(role.role_type.upper() == RoleType.SPONSOR.name and role.role_tier.upper() != SponsorTier.FREE.name
+                       for role in self._roles)
+      return is_sponsor
+
+  def is_paired(self) -> bool:
+    with self._lock:
+      is_paired = any(user.user_id not in self.NOT_PAIRED_USERNAMES for user in self._users)
+      return is_paired
+
+  def is_connected(self) -> bool:
+    network_type = self._sm["deviceState"].networkType
+    return bool(network_type != 0)
+
+  def __del__(self):
+    self.stop()