onepilot/sunnypilot/sunnylink/sunnylink_state.py

"""
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 requests
import time
import json
import pyray as rl

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
from openpilot.system.ui.sunnypilot.lib.styles import style


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._session = requests.Session()  # reuse session to reduce SSL handshake overhead
    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._panel_open = False

    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, session=self._session)
      if response.status_code == 200:
        roles = response.text
        self._params.put("SunnylinkCache_Roles", roles, block=True)
        with self._lock:
          self._roles = _parse_roles(roles)
          sponsor_tier = self._get_highest_tier()
          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, session=self._session)
      if response.status_code == 200:
        users = response.text
        self._params.put("SunnylinkCache_Users", users, block=True)
        with self._lock:
          self._users = _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:
      with self._lock:
        panel_open = self._panel_open

      if panel_open:
        self._sm.update()
        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 get_sponsor_tier_color(self) -> rl.Color:
    tier = self.get_sponsor_tier()

    if tier == SponsorTier.GUARDIAN:
      return rl.Color(255, 215, 0, 255)
    elif tier == SponsorTier.BENEFACTOR:
      return rl.Color(60, 179, 113, 255)
    elif tier == SponsorTier.CONTRIBUTOR:
      return rl.Color(70, 130, 180, 255)
    elif tier == SponsorTier.SUPPORTER:
      return rl.Color(147, 112, 219, 255)
    else:
      return style.ITEM_TEXT_VALUE_COLOR

  def set_settings_open(self, _open: bool) -> None:
    with self._lock:
      self._panel_open = _open

  def __del__(self):
    self.stop()