sunnypilot/selfdrive/ui/widgets/ssh_key.py

import pyray as rl
import requests
import threading
import copy
from collections.abc import Callable
from enum import Enum

from openpilot.common.params import Params
from openpilot.system.ui.lib.application import gui_app, FontWeight
from openpilot.system.ui.lib.multilang import tr, tr_noop
from openpilot.system.ui.lib.text_measure import measure_text_cached
from openpilot.system.ui.widgets import DialogResult
from openpilot.system.ui.widgets.button import Button, ButtonStyle
from openpilot.system.ui.widgets.confirm_dialog import alert_dialog
from openpilot.system.ui.widgets.keyboard import Keyboard
from openpilot.system.ui.widgets.list_view import (
  ItemAction,
  ListItem,
  BUTTON_HEIGHT,
  BUTTON_BORDER_RADIUS,
  BUTTON_FONT_SIZE,
  BUTTON_WIDTH,
)

VALUE_FONT_SIZE = 48


class SshKeyActionState(Enum):
  LOADING = tr_noop("LOADING")
  ADD = tr_noop("ADD")
  REMOVE = tr_noop("REMOVE")


class SshKeyAction(ItemAction):
  HTTP_TIMEOUT = 15  # seconds
  MAX_WIDTH = 500

  def __init__(self):
    super().__init__(self.MAX_WIDTH, True)

    self._keyboard = Keyboard(min_text_size=1)
    self._params = Params()
    self._error_message: str = ""
    self._text_font = gui_app.font(FontWeight.NORMAL)
    self._button = Button("", click_callback=self._handle_button_click, button_style=ButtonStyle.LIST_ACTION,
                          border_radius=BUTTON_BORDER_RADIUS, font_size=BUTTON_FONT_SIZE)

    self._refresh_state()

  def set_touch_valid_callback(self, touch_callback: Callable[[], bool]) -> None:
    super().set_touch_valid_callback(touch_callback)
    self._button.set_touch_valid_callback(touch_callback)

  def _refresh_state(self):
    self._username = self._params.get("GithubUsername")
    self._state = SshKeyActionState.REMOVE if self._params.get("GithubSshKeys") else SshKeyActionState.ADD

  def _render(self, rect: rl.Rectangle) -> bool:
    # Show error dialog if there's an error
    if self._error_message:
      message = copy.copy(self._error_message)
      gui_app.push_widget(alert_dialog(message))
      self._username = ""
      self._error_message = ""

    # Draw username if exists
    if self._username:
      text_size = measure_text_cached(self._text_font, self._username, VALUE_FONT_SIZE)
      rl.draw_text_ex(
        self._text_font,
        self._username,
        (rect.x + rect.width - BUTTON_WIDTH - text_size.x - 30, rect.y + (rect.height - text_size.y) / 2),
        VALUE_FONT_SIZE,
        1.0,
        rl.Color(170, 170, 170, 255),
      )

    # Draw button
    button_rect = rl.Rectangle(rect.x + rect.width - BUTTON_WIDTH, rect.y + (rect.height - BUTTON_HEIGHT) / 2, BUTTON_WIDTH, BUTTON_HEIGHT)
    self._button.set_rect(button_rect)
    self._button.set_text(tr(self._state.value))
    self._button.set_enabled(self._state != SshKeyActionState.LOADING)
    self._button.render(button_rect)
    return False

  def _handle_button_click(self):
    if self._state == SshKeyActionState.ADD:
      self._keyboard.reset()
      self._keyboard.set_title(tr("Enter your GitHub username"))
      self._keyboard.set_callback(self._on_username_submit)
      gui_app.push_widget(self._keyboard)
    elif self._state == SshKeyActionState.REMOVE:
      self._params.remove("GithubUsername")
      self._params.remove("GithubSshKeys")
      self._refresh_state()

  def _on_username_submit(self, result: DialogResult):
    if result != DialogResult.CONFIRM:
      return

    username = self._keyboard.text.strip()
    if not username:
      return

    self._state = SshKeyActionState.LOADING
    threading.Thread(target=lambda: self._fetch_ssh_key(username), daemon=True).start()

  def _fetch_ssh_key(self, username: str):
    try:
      url = f"https://github.com/{username}.keys"
      response = requests.get(url, timeout=self.HTTP_TIMEOUT)
      response.raise_for_status()
      keys = response.text.strip()
      if not keys:
        raise requests.exceptions.HTTPError(tr("No SSH keys found"))

      # Success - save keys
      self._params.put("GithubUsername", username)
      self._params.put("GithubSshKeys", keys)
      self._state = SshKeyActionState.REMOVE
      self._username = username

    except requests.exceptions.Timeout:
      self._error_message = tr("Request timed out")
      self._state = SshKeyActionState.ADD
    except Exception:
      self._error_message = tr("No SSH keys found for user '{}'").format(username)
      self._state = SshKeyActionState.ADD


def ssh_key_item(title: str | Callable[[], str], description: str | Callable[[], str]) -> ListItem:
  return ListItem(title=title, description=description, action_item=SshKeyAction())