ui(raylib): create BaseWindow (#35074)

* ui(raylib): create BaseWindow

* test without typing

* revert

* Revert "test without typing"

This reverts commit c8a5e1b.

* lines

Author: incognitojam

system/ui/lib/window.py

@@ -0,0 +1,57 @@
+import threading
+import time
+import os
+from typing import Generic, Protocol, TypeVar
+from openpilot.system.ui.lib.application import gui_app
+
+
+class RendererProtocol(Protocol):
+  def render(self): ...
+
+
+R = TypeVar("R", bound=RendererProtocol)
+
+
+class BaseWindow(Generic[R]):
+  def __init__(self, title: str):
+    self._title = title
+    self._renderer: R | None = None
+    self._stop_event = threading.Event()
+    self._thread = threading.Thread(target=self._run)
+    self._thread.start()
+
+    # wait for the renderer to be initialized
+    while self._renderer is None and self._thread.is_alive():
+      time.sleep(0.01)
+
+  def _create_renderer(self) -> R:
+    raise NotImplementedError("Subclasses of BaseWindow must implement _create_renderer()")
+
+  def _run(self):
+    if os.getenv("CI") is not None:
+      return
+    gui_app.init_window("Spinner")
+    self._renderer = self._create_renderer()
+    try:
+      for _ in gui_app.render():
+        if self._stop_event.is_set():
+          break
+        self._renderer.render()
+    finally:
+      gui_app.close()
+
+  def __enter__(self):
+    return self
+
+  def close(self):
+    if self._thread.is_alive():
+      self._stop_event.set()
+      self._thread.join(timeout=2.0)
+      if self._thread.is_alive():
+        print(f"WARNING: failed to join {self._title} thread")
+
+  def __del__(self):
+    self.close()
+
+  def __exit__(self, exc_type, exc_val, exc_tb):
+    self.close()

system/ui/spinner.py

@@ -6,6 +6,7 @@
 
 from openpilot.common.basedir import BASEDIR
 from openpilot.system.ui.lib.application import gui_app
+from openpilot.system.ui.lib.window import BaseWindow
 from openpilot.system.ui.text import wrap_text
 
 # Constants
@@ -85,16 +86,12 @@ def render(self):
                         FONT_SIZE, 0.0, rl.WHITE)
 
 
-class Spinner:
+class Spinner(BaseWindow[SpinnerRenderer]):
   def __init__(self):
-    self._renderer: SpinnerRenderer | None = None
-    self._stop_event = threading.Event()
-    self._thread = threading.Thread(target=self._run)
-    self._thread.start()
+    super().__init__("Spinner")
 
-    # wait for the renderer to be initialized
-    while self._renderer is None and self._thread.is_alive():
-      time.sleep(0.01)
+  def _create_renderer(self):
+    return SpinnerRenderer()
 
   def update(self, spinner_text: str):
     if self._renderer is not None:
@@ -103,35 +100,6 @@ def update(self, spinner_text: str):
   def update_progress(self, cur: float, total: float):
     self.update(str(round(100 * cur / total)))
 
-  def _run(self):
-    if os.getenv("CI") is not None:
-      return
-    gui_app.init_window("Spinner")
-    self._renderer = renderer = SpinnerRenderer()
-    try:
-      for _ in gui_app.render():
-        if self._stop_event.is_set():
-          break
-        renderer.render()
-    finally:
-      gui_app.close()
-
-  def __enter__(self):
-    return self
-
-  def close(self):
-    if self._thread.is_alive():
-      self._stop_event.set()
-      self._thread.join(timeout=2.0)
-      if self._thread.is_alive():
-        print("WARNING: failed to join spinner thread")
-
-  def __del__(self):
-    self.close()
-
-  def __exit__(self, exc_type, exc_val, exc_tb):
-    self.close()
-
 
 if __name__ == "__main__":
   with Spinner() as s:

system/ui/text.py

@@ -1,13 +1,12 @@
 #!/usr/bin/env python3
-import os
 import re
-import threading
 import time
 import pyray as rl
 from openpilot.system.hardware import HARDWARE, PC
 from openpilot.system.ui.lib.button import gui_button, ButtonStyle
 from openpilot.system.ui.lib.scroll_panel import GuiScrollPanel
 from openpilot.system.ui.lib.application import gui_app
+from openpilot.system.ui.lib.window import BaseWindow
 
 MARGIN = 50
 SPACING = 40
@@ -74,52 +73,18 @@ def render(self):
     return ret
 
 
-class TextWindow:
+class TextWindow(BaseWindow[TextWindowRenderer]):
   def __init__(self, text: str):
     self._text = text
+    super().__init__("Text")
 
-    self._renderer: TextWindowRenderer | None = None
-    self._stop_event = threading.Event()
-    self._thread = threading.Thread(target=self._run)
-    self._thread.start()
-
-    # wait for the renderer to be initialized
-    while self._renderer is None and self._thread.is_alive():
-      time.sleep(0.01)
+  def _create_renderer(self):
+    return TextWindowRenderer(self._text)
 
   def wait_for_exit(self):
     while self._thread.is_alive():
       time.sleep(0.01)
 
-  def _run(self):
-    if os.getenv("CI") is not None:
-      return
-    gui_app.init_window("Text")
-    self._renderer = renderer = TextWindowRenderer(self._text)
-    try:
-      for _ in gui_app.render():
-        if self._stop_event.is_set():
-          break
-        renderer.render()
-    finally:
-      gui_app.close()
-
-  def __enter__(self):
-    return self
-
-  def close(self):
-    if self._thread.is_alive():
-      self._stop_event.set()
-      self._thread.join(timeout=2.0)
-      if self._thread.is_alive():
-        print("WARNING: failed to join text window thread")
-
-  def __del__(self):
-    self.close()
-
-  def __exit__(self, exc_type, exc_val, exc_tb):
-    self.close()
-
 
 if __name__ == "__main__":
   with TextWindow(DEMO_TEXT):