openpilot_comma/system/ui/updater.py

#!/usr/bin/env python3
import sys
import subprocess
import threading
import pyray as rl
from enum import IntEnum

from openpilot.system.hardware import HARDWARE
from openpilot.system.ui.lib.application import gui_app, FontWeight
from openpilot.system.ui.lib.button import gui_button, ButtonStyle
from openpilot.system.ui.lib.label import gui_text_box, gui_label

# Constants
MARGIN = 50
BUTTON_HEIGHT = 160
BUTTON_WIDTH = 400
PROGRESS_BAR_HEIGHT = 72
TITLE_FONT_SIZE = 80
BODY_FONT_SIZE = 65
BACKGROUND_COLOR = rl.BLACK
PROGRESS_BG_COLOR = rl.Color(41, 41, 41, 255)
PROGRESS_COLOR = rl.Color(54, 77, 239, 255)


class Screen(IntEnum):
  PROMPT = 0
  WIFI = 1
  PROGRESS = 2


# Placeholder for a real WiFi manager implementation
class SimpleWifiManager:
  def __init__(self):
    pass

  def start(self):
    pass

  def stop(self):
    pass


class Updater:
  def __init__(self, updater_path, manifest_path):
    self.updater = updater_path
    self.manifest = manifest_path
    self.current_screen = Screen.PROMPT

    # Initialize simpler components
    self.wifi_manager = SimpleWifiManager()
    self.progress_value = 0
    self.progress_text = "Loading..."
    self.show_reboot_button = False
    self.process = None
    self.update_thread = None

  def install_update(self):
    self.current_screen = Screen.PROGRESS
    self.progress_value = 0
    self.progress_text = "Downloading..."
    self.show_reboot_button = False

    # Start the update process in a separate thread
    self.update_thread = threading.Thread(target=self._run_update_process)
    self.update_thread.daemon = True
    self.update_thread.start()

  def _run_update_process(self):
    # Create a subprocess to run the updater
    cmd = [self.updater, "--swap", self.manifest]
    self.process = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE,
                                   text=True, bufsize=1, universal_newlines=True)

    # Read output line by line
    for line in self.process.stdout:
      parts = line.strip().split(":")
      if len(parts) == 2:
        self.progress_text = parts[0]
        try:
          self.progress_value = int(float(parts[1]))
        except ValueError:
          pass

    # Wait for process to complete
    exit_code = self.process.wait()

    if exit_code == 0:
      # Update succeeded, trigger reboot
      HARDWARE.reboot()
    else:
      # Update failed, show reboot button
      self.progress_text = "Update failed"
      self.show_reboot_button = True

  def render_prompt_screen(self):
    # Title
    title_rect = rl.Rectangle(MARGIN + 50, 250, gui_app.width - MARGIN * 2 - 100, TITLE_FONT_SIZE)
    gui_label(title_rect, "Update Required", TITLE_FONT_SIZE, font_weight=FontWeight.BOLD)

    # Description
    desc_text = "An operating system update is required. Connect your device to Wi-Fi for the fastest update experience. \
                 The download size is approximately 1GB."
    desc_rect = rl.Rectangle(MARGIN + 50, 250 + TITLE_FONT_SIZE + 75, gui_app.width - MARGIN * 2 - 100, BODY_FONT_SIZE * 3)
    gui_text_box(desc_rect, desc_text, BODY_FONT_SIZE)

    # Buttons at the bottom
    button_y = gui_app.height - MARGIN - BUTTON_HEIGHT
    button_width = (gui_app.width - MARGIN * 3) // 2

    # WiFi button
    wifi_button_rect = rl.Rectangle(MARGIN, button_y, button_width, BUTTON_HEIGHT)
    if gui_button(wifi_button_rect, "Connect to Wi-Fi"):
      self.current_screen = Screen.WIFI
      self.wifi_manager.start()
      return  # Return to avoid processing other buttons after screen change

    # Install button
    install_button_rect = rl.Rectangle(MARGIN * 2 + button_width, button_y, button_width, BUTTON_HEIGHT)
    if gui_button(install_button_rect, "Install", button_style=ButtonStyle.PRIMARY):
      self.install_update()
      return  # Return to avoid further processing after action

  def render_wifi_screen(self):
    # Title and back button
    title_rect = rl.Rectangle(MARGIN + 50, MARGIN, gui_app.width - MARGIN * 2 - 100, 60)
    gui_label(title_rect, "Wi-Fi Networks", 60, font_weight=FontWeight.BOLD)

    back_button_rect = rl.Rectangle(MARGIN, gui_app.height - MARGIN - BUTTON_HEIGHT, BUTTON_WIDTH, BUTTON_HEIGHT)
    if gui_button(back_button_rect, "Back"):
      self.current_screen = Screen.PROMPT
      self.wifi_manager.stop()
      return  # Return to avoid processing other interactions after screen change

    # Draw placeholder for WiFi implementation
    placeholder_rect = rl.Rectangle(
      MARGIN,
      title_rect.y + title_rect.height + MARGIN,
      gui_app.width - MARGIN * 2,
      gui_app.height - title_rect.height - MARGIN * 3 - BUTTON_HEIGHT
    )

    # Draw rounded rectangle background
    rl.draw_rectangle_rounded(
      placeholder_rect,
      0.1,
      10,
      rl.Color(41, 41, 41, 255)
    )

    # Draw placeholder text
    placeholder_text = "WiFi Implementation Placeholder"
    text_size = rl.measure_text_ex(gui_app.font(), placeholder_text, 80, 1)
    text_pos = rl.Vector2(
      placeholder_rect.x + (placeholder_rect.width - text_size.x) / 2,
      placeholder_rect.y + (placeholder_rect.height - text_size.y) / 2
    )
    rl.draw_text_ex(gui_app.font(), placeholder_text, text_pos, 80, 1, rl.WHITE)

    # Draw instructions
    instructions_text = "Real WiFi functionality would be implemented here"
    instructions_size = rl.measure_text_ex(gui_app.font(), instructions_text, 40, 1)
    instructions_pos = rl.Vector2(
      placeholder_rect.x + (placeholder_rect.width - instructions_size.x) / 2,
      text_pos.y + text_size.y + 20
    )
    rl.draw_text_ex(gui_app.font(), instructions_text, instructions_pos, 40, 1, rl.GRAY)

  def render_progress_screen(self):
    title_rect = rl.Rectangle(MARGIN + 100, 330, gui_app.width - MARGIN * 2 - 200, 100)
    gui_label(title_rect, self.progress_text, 90, font_weight=FontWeight.SEMI_BOLD)

    # Progress bar
    bar_rect = rl.Rectangle(MARGIN + 100, 330 + 100 + 100, gui_app.width - MARGIN * 2 - 200, PROGRESS_BAR_HEIGHT)
    rl.draw_rectangle_rounded(bar_rect, 0.5, 10, PROGRESS_BG_COLOR)

    # Calculate the width of the progress chunk
    progress_width = (bar_rect.width * self.progress_value) / 100
    if progress_width > 0:
      progress_rect = rl.Rectangle(bar_rect.x, bar_rect.y, progress_width, bar_rect.height)
      rl.draw_rectangle_rounded(progress_rect, 0.5, 10, PROGRESS_COLOR)

    # Show reboot button if needed
    if self.show_reboot_button:
      reboot_rect = rl.Rectangle(MARGIN + 100, gui_app.height - MARGIN - BUTTON_HEIGHT, BUTTON_WIDTH, BUTTON_HEIGHT)
      if gui_button(reboot_rect, "Reboot"):
        # Return True to signal main loop to exit before rebooting
        HARDWARE.reboot()
        return

  def render(self):
    try:
      if self.current_screen == Screen.PROMPT:
        self.render_prompt_screen()
      elif self.current_screen == Screen.WIFI:
        self.render_wifi_screen()
      elif self.current_screen == Screen.PROGRESS:
        self.render_progress_screen()
    except Exception as e:
      import traceback
      print(f"Error in render: {e}")
      traceback.print_exc()  # Print full traceback for debugging


def main():
  if len(sys.argv) < 3:
    print("Usage: updater.py <updater_path> <manifest_path>")
    sys.exit(1)

  updater_path = sys.argv[1]
  manifest_path = sys.argv[2]

  try:
    gui_app.init_window("System Update")
    updater = Updater(updater_path, manifest_path)

    for _ in gui_app.render():
      try:
        updater.render()
      except Exception as e:
        import traceback
        print(f"Error in main render loop: {e}")
        traceback.print_exc()
  except Exception as e:
    print(f"Error initializing application: {e}")
  finally:
    # Make sure we clean up even if there's an error
    gui_app.close()


if __name__ == "__main__":
  main()