openpilot v0.9.9 release (#35334)

* openpilot v0.9.9 release

date: 2025-06-05T19:54:08
master commit: 8aadf02b2fd91f4e1285e18c2c7feb32d93b66f5

* AGNOS 12.4 (#35558)

agnos12.4

---------

Co-authored-by: Vehicle Researcher <user@comma.ai>
Co-authored-by: Maxime Desroches <desroches.maxime@gmail.com>
This commit is contained in:
Adeeb Shihadeh
2025-06-17 16:32:08 -07:00
committed by GitHub
parent dd778596b7
commit 100f89a161
1240 changed files with 310104 additions and 30549 deletions
+2 -1
View File
@@ -3,7 +3,8 @@
The user interfaces here are built with [raylib](https://www.raylib.com/).
Quick start:
* set `DEBUG_FPS=1` to show the FPS
* set `SHOW_FPS=1` to show the FPS
* set `STRICT_MODE=1` to kill the app if it drops too much below 60fps
* set `SCALE=1.5` to scale the entire UI by 1.5x
* https://www.raylib.com/cheatsheet/cheatsheet.html
* https://electronstudio.github.io/raylib-python-cffi/README.html#quickstart
+160 -42
View File
@@ -3,31 +3,37 @@ import os
import time
import pyray as rl
from enum import IntEnum
from openpilot.common.basedir import BASEDIR
from importlib.resources import as_file, files
from openpilot.common.swaglog import cloudlog
from openpilot.system.hardware import HARDWARE
DEFAULT_FPS = 60
FPS_LOG_INTERVAL = 5 # Seconds between logging FPS drops
FPS_DROP_THRESHOLD = 0.9 # FPS drop threshold for triggering a warning
FPS_CRITICAL_THRESHOLD = 0.5 # Critical threshold for triggering strict actions
DEBUG_FPS = os.getenv("DEBUG_FPS") == '1'
ENABLE_VSYNC = os.getenv("ENABLE_VSYNC") == "1"
SHOW_FPS = os.getenv("SHOW_FPS") == '1'
STRICT_MODE = os.getenv("STRICT_MODE") == '1'
SCALE = float(os.getenv("SCALE", "1.0"))
DEFAULT_TEXT_SIZE = 60
DEFAULT_TEXT_COLOR = rl.Color(200, 200, 200, 255)
FONT_DIR = os.path.join(BASEDIR, "selfdrive/assets/fonts")
DEFAULT_TEXT_COLOR = rl.WHITE
ASSETS_DIR = files("openpilot.selfdrive").joinpath("assets")
FONT_DIR = ASSETS_DIR.joinpath("fonts")
class FontWeight(IntEnum):
BLACK = 0
BOLD = 1
EXTRA_BOLD = 2
EXTRA_LIGHT = 3
THIN = 0
EXTRA_LIGHT = 1
LIGHT = 2
NORMAL = 3
MEDIUM = 4
NORMAL = 5
SEMI_BOLD = 6
THIN = 7
SEMI_BOLD = 5
BOLD = 6
EXTRA_BOLD = 7
BLACK = 8
class GuiApplication:
@@ -35,58 +41,133 @@ class GuiApplication:
self._fonts: dict[FontWeight, rl.Font] = {}
self._width = width
self._height = height
self._textures: list[rl.Texture] = []
self._scale = SCALE
self._scaled_width = int(self._width * self._scale)
self._scaled_height = int(self._height * self._scale)
self._render_texture: rl.RenderTexture | None = None
self._textures: dict[str, rl.Texture] = {}
self._target_fps: int = DEFAULT_FPS
self._last_fps_log_time: float = time.monotonic()
self._window_close_requested = False
self._trace_log_callback = None
def init_window(self, title: str, fps: int=DEFAULT_FPS):
def request_close(self):
self._window_close_requested = True
def init_window(self, title: str, fps: int = DEFAULT_FPS):
atexit.register(self.close) # Automatically call close() on exit
rl.set_config_flags(rl.ConfigFlags.FLAG_MSAA_4X_HINT | rl.ConfigFlags.FLAG_VSYNC_HINT)
rl.init_window(self._width, self._height, title)
HARDWARE.set_display_power(True)
HARDWARE.set_screen_brightness(65)
self._set_log_callback()
rl.set_trace_log_level(rl.TraceLogLevel.LOG_ALL)
flags = rl.ConfigFlags.FLAG_MSAA_4X_HINT
if ENABLE_VSYNC:
flags |= rl.ConfigFlags.FLAG_VSYNC_HINT
rl.set_config_flags(flags)
rl.init_window(self._scaled_width, self._scaled_height, title)
if self._scale != 1.0:
rl.set_mouse_scale(1 / self._scale, 1 / self._scale)
self._render_texture = rl.load_render_texture(self._width, self._height)
rl.set_texture_filter(self._render_texture.texture, rl.TextureFilter.TEXTURE_FILTER_BILINEAR)
rl.set_target_fps(fps)
self._target_fps = fps
self._set_styles()
self._load_fonts()
def load_texture_from_image(self, file_name: str, width: int, height: int):
def texture(self, asset_path: str, width: int, height: int, alpha_premultiply=False, keep_aspect_ratio=True):
cache_key = f"{asset_path}_{width}_{height}_{alpha_premultiply}{keep_aspect_ratio}"
if cache_key in self._textures:
return self._textures[cache_key]
with as_file(ASSETS_DIR.joinpath(asset_path)) as fspath:
texture_obj = self._load_texture_from_image(fspath.as_posix(), width, height, alpha_premultiply, keep_aspect_ratio)
self._textures[cache_key] = texture_obj
return texture_obj
def _load_texture_from_image(self, image_path: str, width: int, height: int, alpha_premultiply=False, keep_aspect_ratio=True):
"""Load and resize a texture, storing it for later automatic unloading."""
image = rl.load_image(file_name)
rl.image_resize(image, width, height)
image = rl.load_image(image_path)
if alpha_premultiply:
rl.image_alpha_premultiply(image)
# Resize with aspect ratio preservation if requested
if keep_aspect_ratio:
orig_width = image.width
orig_height = image.height
scale_width = width / orig_width
scale_height = height / orig_height
# Calculate new dimensions
scale = min(scale_width, scale_height)
new_width = int(orig_width * scale)
new_height = int(orig_height * scale)
rl.image_resize(image, new_width, new_height)
else:
rl.image_resize(image, width, height)
texture = rl.load_texture_from_image(image)
# Set texture filtering to smooth the result
rl.set_texture_filter(texture, rl.TextureFilter.TEXTURE_FILTER_BILINEAR)
rl.unload_image(image)
self._textures.append(texture)
return texture
def close(self):
for texture in self._textures:
if not rl.is_window_ready():
return
for texture in self._textures.values():
rl.unload_texture(texture)
self._textures = {}
for font in self._fonts.values():
rl.unload_font(font)
self._fonts = {}
if self._render_texture is not None:
rl.unload_render_texture(self._render_texture)
self._render_texture = None
rl.close_window()
def render(self):
while not rl.window_should_close():
rl.begin_drawing()
rl.clear_background(rl.BLACK)
try:
while not (self._window_close_requested or rl.window_should_close()):
if self._render_texture:
rl.begin_texture_mode(self._render_texture)
rl.clear_background(rl.BLACK)
else:
rl.begin_drawing()
rl.clear_background(rl.BLACK)
yield
yield
if DEBUG_FPS:
rl.draw_fps(10, 10)
if self._render_texture:
rl.end_texture_mode()
rl.begin_drawing()
rl.clear_background(rl.BLACK)
src_rect = rl.Rectangle(0, 0, float(self._width), -float(self._height))
dst_rect = rl.Rectangle(0, 0, float(self._scaled_width), float(self._scaled_height))
rl.draw_texture_pro(self._render_texture.texture, src_rect, dst_rect, rl.Vector2(0, 0), 0.0, rl.WHITE)
rl.end_drawing()
self._monitor_fps()
if SHOW_FPS:
rl.draw_fps(10, 10)
def font(self, font_wight: FontWeight=FontWeight.NORMAL):
return self._fonts[font_wight]
rl.end_drawing()
self._monitor_fps()
except KeyboardInterrupt:
pass
def font(self, font_weight: FontWeight = FontWeight.NORMAL):
return self._fonts[font_weight]
@property
def width(self):
@@ -98,21 +179,36 @@ class GuiApplication:
def _load_fonts(self):
font_files = (
"Inter-Black.ttf",
"Inter-Thin.ttf",
"Inter-ExtraLight.ttf",
"Inter-Light.ttf",
"Inter-Regular.ttf",
"Inter-Medium.ttf",
"Inter-SemiBold.ttf",
"Inter-Bold.ttf",
"Inter-ExtraBold.ttf",
"Inter-ExtraLight.ttf",
"Inter-Medium.ttf",
"Inter-Regular.ttf",
"Inter-SemiBold.ttf",
"Inter-Thin.ttf"
)
"Inter-Black.ttf",
)
# Create a character set from our keyboard layouts
from openpilot.system.ui.widgets.keyboard import KEYBOARD_LAYOUTS
from openpilot.selfdrive.ui.onroad.hud_renderer import CRUISE_DISABLED_CHAR
all_chars = set()
for layout in KEYBOARD_LAYOUTS.values():
all_chars.update(key for row in layout for key in row)
all_chars = "".join(all_chars)
all_chars += CRUISE_DISABLED_CHAR
codepoint_count = rl.ffi.new("int *", 1)
codepoints = rl.load_codepoints(all_chars, codepoint_count)
for index, font_file in enumerate(font_files):
font = rl.load_font_ex(os.path.join(FONT_DIR, font_file), 120, None, 0)
rl.set_texture_filter(font.texture, rl.TextureFilter.TEXTURE_FILTER_BILINEAR)
self._fonts[index] = font
with as_file(FONT_DIR.joinpath(font_file)) as fspath:
font = rl.load_font_ex(fspath.as_posix(), 200, codepoints, codepoint_count[0])
rl.set_texture_filter(font.texture, rl.TextureFilter.TEXTURE_FILTER_BILINEAR)
self._fonts[index] = font
rl.unload_codepoints(codepoints)
rl.gui_set_font(self._fonts[FontWeight.NORMAL])
def _set_styles(self):
@@ -120,9 +216,31 @@ class GuiApplication:
rl.gui_set_style(rl.GuiControl.DEFAULT, rl.GuiDefaultProperty.TEXT_SIZE, DEFAULT_TEXT_SIZE)
rl.gui_set_style(rl.GuiControl.DEFAULT, rl.GuiDefaultProperty.BACKGROUND_COLOR, rl.color_to_int(rl.BLACK))
rl.gui_set_style(rl.GuiControl.DEFAULT, rl.GuiControlProperty.TEXT_COLOR_NORMAL, rl.color_to_int(DEFAULT_TEXT_COLOR))
rl.gui_set_style(rl.GuiControl.DEFAULT, rl.GuiDefaultProperty.BACKGROUND_COLOR, rl.color_to_int(rl.Color(30, 30, 30, 255)))
rl.gui_set_style(rl.GuiControl.DEFAULT, rl.GuiControlProperty.BASE_COLOR_NORMAL, rl.color_to_int(rl.Color(50, 50, 50, 255)))
def _set_log_callback(self):
@rl.ffi.callback("void(int, char *, void *)")
def trace_log_callback(log_level, text, args):
try:
text_str = rl.ffi.string(text).decode('utf-8')
except (TypeError, UnicodeDecodeError):
text_str = str(text)
if log_level == rl.TraceLogLevel.LOG_ERROR:
cloudlog.error(f"raylib: {text_str}")
elif log_level == rl.TraceLogLevel.LOG_WARNING:
cloudlog.warning(f"raylib: {text_str}")
elif log_level == rl.TraceLogLevel.LOG_INFO:
cloudlog.info(f"raylib: {text_str}")
elif log_level == rl.TraceLogLevel.LOG_DEBUG:
cloudlog.debug(f"raylib: {text_str}")
else:
cloudlog.error(f"raylib: Unknown level {log_level}: {text_str}")
# Store callback reference
self._trace_log_callback = trace_log_callback
rl.set_trace_log_callback(self._trace_log_callback)
def _monitor_fps(self):
fps = rl.get_fps()
+77 -10
View File
@@ -8,11 +8,21 @@ class ButtonStyle(IntEnum):
PRIMARY = 1 # For main actions
DANGER = 2 # For critical actions, like reboot or delete
TRANSPARENT = 3 # For buttons with transparent background and border
ACTION = 4
class TextAlignment(IntEnum):
LEFT = 0
CENTER = 1
RIGHT = 2
ICON_PADDING = 15
DEFAULT_BUTTON_FONT_SIZE = 60
BUTTON_ENABLED_TEXT_COLOR = rl.Color(228, 228, 228, 255)
BUTTON_DISABLED_TEXT_COLOR = rl.Color(228, 228, 228, 51)
ACTION_BUTTON_FONT_SIZE = 48
ACTION_BUTTON_TEXT_COLOR = rl.Color(0, 0, 0, 255)
BUTTON_BACKGROUND_COLORS = {
@@ -20,6 +30,7 @@ BUTTON_BACKGROUND_COLORS = {
ButtonStyle.PRIMARY: rl.Color(70, 91, 234, 255),
ButtonStyle.DANGER: rl.Color(255, 36, 36, 255),
ButtonStyle.TRANSPARENT: rl.BLACK,
ButtonStyle.ACTION: rl.Color(189, 189, 189, 255),
}
BUTTON_PRESSED_BACKGROUND_COLORS = {
@@ -27,8 +38,11 @@ BUTTON_PRESSED_BACKGROUND_COLORS = {
ButtonStyle.PRIMARY: rl.Color(48, 73, 244, 255),
ButtonStyle.DANGER: rl.Color(255, 36, 36, 255),
ButtonStyle.TRANSPARENT: rl.BLACK,
ButtonStyle.ACTION: rl.Color(130, 130, 130, 255),
}
_pressed_buttons: set[str] = set() # Track mouse press state globally
def gui_button(
rect: rl.Rectangle,
@@ -38,16 +52,42 @@ def gui_button(
button_style: ButtonStyle = ButtonStyle.NORMAL,
is_enabled: bool = True,
border_radius: int = 10, # Corner rounding in pixels
text_alignment: TextAlignment = TextAlignment.CENTER,
text_padding: int = 20, # Padding for left/right alignment
icon=None,
) -> int:
button_id = f"{rect.x}_{rect.y}_{rect.width}_{rect.height}"
result = 0
if button_style in (ButtonStyle.PRIMARY, ButtonStyle.DANGER) and not is_enabled:
button_style = ButtonStyle.NORMAL
if button_style == ButtonStyle.ACTION and font_size == DEFAULT_BUTTON_FONT_SIZE:
font_size = ACTION_BUTTON_FONT_SIZE
# Set background color based on button type
bg_color = BUTTON_BACKGROUND_COLORS[button_style]
if is_enabled and rl.check_collision_point_rec(rl.get_mouse_position(), rect):
if rl.is_mouse_button_down(rl.MouseButton.MOUSE_BUTTON_LEFT):
mouse_over = is_enabled and rl.check_collision_point_rec(rl.get_mouse_position(), rect)
is_pressed = button_id in _pressed_buttons
if mouse_over:
if rl.is_mouse_button_pressed(rl.MouseButton.MOUSE_BUTTON_LEFT):
# Only this button enters pressed state
_pressed_buttons.add(button_id)
is_pressed = True
# Use pressed color when mouse is down over this button
if is_pressed and rl.is_mouse_button_down(rl.MouseButton.MOUSE_BUTTON_LEFT):
bg_color = BUTTON_PRESSED_BACKGROUND_COLORS[button_style]
elif rl.is_mouse_button_released(rl.MouseButton.MOUSE_BUTTON_LEFT):
# Handle button click
if rl.is_mouse_button_released(rl.MouseButton.MOUSE_BUTTON_LEFT) and is_pressed:
result = 1
_pressed_buttons.remove(button_id)
# Clean up pressed state if mouse is released anywhere
if rl.is_mouse_button_released(rl.MouseButton.MOUSE_BUTTON_LEFT) and button_id in _pressed_buttons:
_pressed_buttons.remove(button_id)
# Draw the button with rounded corners
roundness = border_radius / (min(rect.width, rect.height) / 2)
@@ -57,15 +97,42 @@ def gui_button(
rl.draw_rectangle_rounded(rect, roundness, 20, rl.BLACK)
rl.draw_rectangle_rounded_lines_ex(rect, roundness, 20, 2, rl.WHITE)
# Handle icon and text positioning
font = gui_app.font(font_weight)
# Center text in the button
text_size = rl.measure_text_ex(font, text, font_size, 0)
text_pos = rl.Vector2(
rect.x + (rect.width - text_size.x) // 2, rect.y + (rect.height - text_size.y) // 2
)
text_pos = rl.Vector2(0, rect.y + (rect.height - text_size.y) // 2) # Vertical centering
# Draw the button text
text_color = BUTTON_ENABLED_TEXT_COLOR if is_enabled else BUTTON_DISABLED_TEXT_COLOR
rl.draw_text_ex(font, text, text_pos, font_size, 0, text_color)
# Draw icon if provided
if icon:
icon_y = rect.y + (rect.height - icon.height) / 2
if text:
if text_alignment == TextAlignment.LEFT:
icon_x = rect.x + text_padding
text_pos.x = icon_x + icon.width + ICON_PADDING
elif text_alignment == TextAlignment.CENTER:
total_width = icon.width + ICON_PADDING + text_size.x
icon_x = rect.x + (rect.width - total_width) / 2
text_pos.x = icon_x + icon.width + ICON_PADDING
else: # RIGHT
text_pos.x = rect.x + rect.width - text_size.x - text_padding
icon_x = text_pos.x - ICON_PADDING - icon.width
else:
# Center icon when no text
icon_x = rect.x + (rect.width - icon.width) / 2
rl.draw_texture_v(icon, rl.Vector2(icon_x, icon_y), rl.WHITE if is_enabled else rl.Color(255, 255, 255, 100))
else:
# No icon, position text normally
if text_alignment == TextAlignment.LEFT:
text_pos.x = rect.x + text_padding
elif text_alignment == TextAlignment.CENTER:
text_pos.x = rect.x + (rect.width - text_size.x) // 2
elif text_alignment == TextAlignment.RIGHT:
text_pos.x = rect.x + rect.width - text_size.x - text_padding
# Draw the button text if any
if text:
text_color = ACTION_BUTTON_TEXT_COLOR if button_style == ButtonStyle.ACTION else BUTTON_ENABLED_TEXT_COLOR if is_enabled else BUTTON_DISABLED_TEXT_COLOR
rl.draw_text_ex(font, text, text_pos, font_size, 0, text_color)
return result
+177
View File
@@ -0,0 +1,177 @@
import os
import cffi
from dataclasses import dataclass
from typing import Any
from openpilot.common.swaglog import cloudlog
# EGL constants
EGL_LINUX_DMA_BUF_EXT = 0x3270
EGL_WIDTH = 0x3057
EGL_HEIGHT = 0x3056
EGL_LINUX_DRM_FOURCC_EXT = 0x3271
EGL_DMA_BUF_PLANE0_FD_EXT = 0x3272
EGL_DMA_BUF_PLANE0_OFFSET_EXT = 0x3273
EGL_DMA_BUF_PLANE0_PITCH_EXT = 0x3274
EGL_DMA_BUF_PLANE1_FD_EXT = 0x3275
EGL_DMA_BUF_PLANE1_OFFSET_EXT = 0x3276
EGL_DMA_BUF_PLANE1_PITCH_EXT = 0x3277
EGL_NONE = 0x3038
GL_TEXTURE0 = 0x84C0
GL_TEXTURE_EXTERNAL_OES = 0x8D65
# DRM Format for NV12
DRM_FORMAT_NV12 = 842094158
@dataclass
class EGLImage:
"""Container for EGL image and associated resources"""
egl_image: Any
fd: int
@dataclass
class EGLState:
"""Container for all EGL-related state"""
initialized: bool = False
ffi: Any = None
egl_lib: Any = None
gles_lib: Any = None
# EGL display connection - shared across all users
display: Any = None
# Constants
NO_CONTEXT: Any = None
NO_DISPLAY: Any = None
NO_IMAGE_KHR: Any = None
# Function pointers
get_current_display: Any = None
create_image_khr: Any = None
destroy_image_khr: Any = None
image_target_texture: Any = None
get_error: Any = None
bind_texture: Any = None
active_texture: Any = None
# Create a single instance of the state
_egl = EGLState()
def init_egl() -> bool:
"""Initialize EGL and load necessary functions"""
global _egl
# Don't re-initialize if already done
if _egl.initialized:
return True
try:
_egl.ffi = cffi.FFI()
_egl.ffi.cdef("""
typedef int EGLint;
typedef unsigned int EGLBoolean;
typedef unsigned int EGLenum;
typedef unsigned int GLenum;
typedef void *EGLContext;
typedef void *EGLDisplay;
typedef void *EGLClientBuffer;
typedef void *EGLImageKHR;
typedef void *GLeglImageOES;
EGLDisplay eglGetCurrentDisplay(void);
EGLint eglGetError(void);
EGLImageKHR eglCreateImageKHR(EGLDisplay dpy, EGLContext ctx,
EGLenum target, EGLClientBuffer buffer,
const EGLint *attrib_list);
EGLBoolean eglDestroyImageKHR(EGLDisplay dpy, EGLImageKHR image);
void glEGLImageTargetTexture2DOES(GLenum target, GLeglImageOES image);
void glBindTexture(GLenum target, unsigned int texture);
void glActiveTexture(GLenum texture);
""")
# Load libraries
_egl.egl_lib = _egl.ffi.dlopen("libEGL.so")
_egl.gles_lib = _egl.ffi.dlopen("libGLESv2.so")
# Cast NULL pointers
_egl.NO_CONTEXT = _egl.ffi.cast("void *", 0)
_egl.NO_DISPLAY = _egl.ffi.cast("void *", 0)
_egl.NO_IMAGE_KHR = _egl.ffi.cast("void *", 0)
# Bind functions
_egl.get_current_display = _egl.egl_lib.eglGetCurrentDisplay
_egl.create_image_khr = _egl.egl_lib.eglCreateImageKHR
_egl.destroy_image_khr = _egl.egl_lib.eglDestroyImageKHR
_egl.image_target_texture = _egl.gles_lib.glEGLImageTargetTexture2DOES
_egl.get_error = _egl.egl_lib.eglGetError
_egl.bind_texture = _egl.gles_lib.glBindTexture
_egl.active_texture = _egl.gles_lib.glActiveTexture
# Initialize EGL display once here
_egl.display = _egl.get_current_display()
if _egl.display == _egl.NO_DISPLAY:
raise RuntimeError("Failed to get EGL display")
_egl.initialized = True
return True
except Exception as e:
cloudlog.exception(f"EGL initialization failed: {e}")
_egl.initialized = False
return False
def create_egl_image(width: int, height: int, stride: int, fd: int, uv_offset: int) -> EGLImage | None:
assert _egl.initialized, "EGL not initialized"
# Duplicate fd since EGL needs it
dup_fd = os.dup(fd)
# Create image attributes for EGL
img_attrs = [
EGL_WIDTH, width,
EGL_HEIGHT, height,
EGL_LINUX_DRM_FOURCC_EXT, DRM_FORMAT_NV12,
EGL_DMA_BUF_PLANE0_FD_EXT, dup_fd,
EGL_DMA_BUF_PLANE0_OFFSET_EXT, 0,
EGL_DMA_BUF_PLANE0_PITCH_EXT, stride,
EGL_DMA_BUF_PLANE1_FD_EXT, dup_fd,
EGL_DMA_BUF_PLANE1_OFFSET_EXT, uv_offset,
EGL_DMA_BUF_PLANE1_PITCH_EXT, stride,
EGL_NONE
]
attr_array = _egl.ffi.new("int[]", img_attrs)
egl_image = _egl.create_image_khr(_egl.display, _egl.NO_CONTEXT, EGL_LINUX_DMA_BUF_EXT, _egl.ffi.NULL, attr_array)
if egl_image == _egl.NO_IMAGE_KHR:
cloudlog.error(f"Failed to create EGL image: {_egl.get_error()}")
os.close(dup_fd)
return None
return EGLImage(egl_image=egl_image, fd=dup_fd)
def destroy_egl_image(egl_image: EGLImage) -> None:
assert _egl.initialized, "EGL not initialized"
_egl.destroy_image_khr(_egl.display, egl_image.egl_image)
# Close the duplicated fd we created in create_egl_image()
# We need to handle OSError since the fd might already be closed
try:
os.close(egl_image.fd)
except OSError:
pass
def bind_egl_image_to_texture(texture_id: int, egl_image: EGLImage) -> None:
assert _egl.initialized, "EGL not initialized"
_egl.active_texture(GL_TEXTURE0)
_egl.bind_texture(GL_TEXTURE_EXTERNAL_OES, texture_id)
_egl.image_target_texture(GL_TEXTURE_EXTERNAL_OES, egl_image.egl_image)
+232
View File
@@ -0,0 +1,232 @@
import pyray as rl
import time
from openpilot.system.ui.lib.application import gui_app
PASSWORD_MASK_CHAR = ""
PASSWORD_MASK_DELAY = 1.5 # Seconds to show character before masking
class InputBox:
def __init__(self, max_text_size=255, password_mode=False):
self._max_text_size = max_text_size
self._input_text = ""
self._cursor_position = 0
self._password_mode = password_mode
self._blink_counter = 0
self._show_cursor = False
self._last_key_pressed = 0
self._key_press_time = 0
self._repeat_delay = 30
self._repeat_rate = 4
self._text_offset = 0
self._visible_width = 0
self._last_char_time = 0 # Track when last character was added
self._masked_length = 0 # How many characters are currently masked
@property
def text(self):
return self._input_text
@text.setter
def text(self, value):
self._input_text = value[: self._max_text_size]
self._cursor_position = len(self._input_text)
self._update_text_offset()
def set_password_mode(self, password_mode):
self._password_mode = password_mode
def clear(self):
self._input_text = ''
self._cursor_position = 0
self._text_offset = 0
def set_cursor_position(self, position):
"""Set the cursor position and reset the blink counter."""
if 0 <= position <= len(self._input_text):
self._cursor_position = position
self._blink_counter = 0
self._show_cursor = True
self._update_text_offset()
def _update_text_offset(self):
"""Ensure the cursor is visible by adjusting text offset."""
if self._visible_width == 0:
return
font = gui_app.font()
display_text = self._get_display_text()
padding = 10
if self._cursor_position > 0:
cursor_x = rl.measure_text_ex(font, display_text[: self._cursor_position], self._font_size, 0).x
else:
cursor_x = 0
visible_width = self._visible_width - (padding * 2)
# Adjust offset if cursor would be outside visible area
if cursor_x < self._text_offset:
self._text_offset = max(0, cursor_x - padding)
elif cursor_x > self._text_offset + visible_width:
self._text_offset = cursor_x - visible_width + padding
def add_char_at_cursor(self, char):
"""Add a character at the current cursor position."""
if len(self._input_text) < self._max_text_size:
self._input_text = self._input_text[: self._cursor_position] + char + self._input_text[self._cursor_position :]
self.set_cursor_position(self._cursor_position + 1)
if self._password_mode:
self._last_char_time = time.time()
return True
return False
def delete_char_before_cursor(self):
"""Delete the character before the cursor position (backspace)."""
if self._cursor_position > 0:
self._input_text = self._input_text[: self._cursor_position - 1] + self._input_text[self._cursor_position :]
self.set_cursor_position(self._cursor_position - 1)
return True
return False
def delete_char_at_cursor(self):
"""Delete the character at the cursor position (delete)."""
if self._cursor_position < len(self._input_text):
self._input_text = self._input_text[: self._cursor_position] + self._input_text[self._cursor_position + 1 :]
self.set_cursor_position(self._cursor_position)
return True
return False
def render(self, rect, color=rl.BLACK, border_color=rl.DARKGRAY, text_color=rl.WHITE, font_size=80):
# Store dimensions for text offset calculations
self._visible_width = rect.width
self._font_size = font_size
# Handle mouse input
self._handle_mouse_input(rect, font_size)
# Draw input box
rl.draw_rectangle_rec(rect, color)
# Process keyboard input
self._handle_keyboard_input()
# Update cursor blink
self._blink_counter += 1
if self._blink_counter >= 30:
self._show_cursor = not self._show_cursor
self._blink_counter = 0
# Display text
font = gui_app.font()
display_text = self._get_display_text()
padding = 10
# Clip text within input box bounds
buffer = 2
rl.begin_scissor_mode(int(rect.x + padding - buffer), int(rect.y), int(rect.width - padding * 2 + buffer * 2), int(rect.height))
rl.draw_text_ex(
font,
display_text,
rl.Vector2(int(rect.x + padding - self._text_offset), int(rect.y + rect.height / 2 - font_size / 2)),
font_size,
0,
text_color,
)
# Draw cursor
if self._show_cursor:
cursor_x = rect.x + padding
if len(display_text) > 0 and self._cursor_position > 0:
cursor_x += rl.measure_text_ex(font, display_text[: self._cursor_position], font_size, 0).x
# Apply text offset to cursor position
cursor_x -= self._text_offset
cursor_height = font_size + 4
cursor_y = rect.y + rect.height / 2 - cursor_height / 2
rl.draw_line(int(cursor_x), int(cursor_y), int(cursor_x), int(cursor_y + cursor_height), rl.WHITE)
rl.end_scissor_mode()
def _get_display_text(self):
"""Get text to display, applying password masking with delay if needed."""
if not self._password_mode:
return self._input_text
# Show character at last edited position if within delay window
masked_text = PASSWORD_MASK_CHAR * len(self._input_text)
recent_edit = time.time() - self._last_char_time < PASSWORD_MASK_DELAY
if recent_edit and self._input_text:
last_pos = max(0, self._cursor_position - 1)
if last_pos < len(self._input_text):
return masked_text[:last_pos] + self._input_text[last_pos] + masked_text[last_pos + 1 :]
return masked_text
def _handle_mouse_input(self, rect, font_size):
"""Handle mouse clicks to position cursor."""
mouse_pos = rl.get_mouse_position()
if rl.is_mouse_button_pressed(rl.MOUSE_LEFT_BUTTON) and rl.check_collision_point_rec(mouse_pos, rect):
# Calculate cursor position from click
if len(self._input_text) > 0:
font = gui_app.font()
display_text = self._get_display_text()
# Find the closest character position to the click
relative_x = mouse_pos.x - (rect.x + 10) + self._text_offset
best_pos = 0
min_distance = float('inf')
for i in range(len(self._input_text) + 1):
char_width = rl.measure_text_ex(font, display_text[:i], font_size, 0).x
distance = abs(relative_x - char_width)
if distance < min_distance:
min_distance = distance
best_pos = i
self.set_cursor_position(best_pos)
else:
self.set_cursor_position(0)
def _handle_keyboard_input(self):
# Handle navigation keys
key = rl.get_key_pressed()
if key != 0:
self._process_key(key)
if key in (rl.KEY_LEFT, rl.KEY_RIGHT, rl.KEY_BACKSPACE, rl.KEY_DELETE):
self._last_key_pressed = key
self._key_press_time = 0
# Handle repeats for held keys
elif self._last_key_pressed != 0:
if rl.is_key_down(self._last_key_pressed):
self._key_press_time += 1
if self._key_press_time > self._repeat_delay and self._key_press_time % self._repeat_rate == 0:
self._process_key(self._last_key_pressed)
else:
self._last_key_pressed = 0
# Handle text input
char = rl.get_char_pressed()
if char != 0 and char >= 32: # Filter out control characters
self.add_char_at_cursor(chr(char))
def _process_key(self, key):
if key == rl.KEY_LEFT:
if self._cursor_position > 0:
self.set_cursor_position(self._cursor_position - 1)
elif key == rl.KEY_RIGHT:
if self._cursor_position < len(self._input_text):
self.set_cursor_position(self._cursor_position + 1)
elif key == rl.KEY_BACKSPACE:
self.delete_char_before_cursor()
elif key == rl.KEY_DELETE:
self.delete_char_at_cursor()
elif key == rl.KEY_HOME:
self.set_cursor_position(0)
elif key == rl.KEY_END:
self.set_cursor_position(len(self._input_text))
+27 -6
View File
@@ -10,13 +10,27 @@ def gui_label(
color: rl.Color = DEFAULT_TEXT_COLOR,
font_weight: FontWeight = FontWeight.NORMAL,
alignment: int = rl.GuiTextAlignment.TEXT_ALIGN_LEFT,
alignment_vertical: int = rl.GuiTextAlignmentVertical.TEXT_ALIGN_MIDDLE
alignment_vertical: int = rl.GuiTextAlignmentVertical.TEXT_ALIGN_MIDDLE,
elide_right: bool = True
):
# Set font based on the provided weight
font = gui_app.font(font_weight)
# Measure text size
text_size = rl.measure_text_ex(font, text, font_size, 0)
display_text = text
# Elide text to fit within the rectangle
if elide_right and text_size.x > rect.width:
ellipsis = "..."
left, right = 0, len(text)
while left < right:
mid = (left + right) // 2
candidate = text[:mid] + ellipsis
candidate_size = rl.measure_text_ex(font, candidate, font_size, 0)
if candidate_size.x <= rect.width:
left = mid + 1
else:
right = mid
display_text = text[: left - 1] + ellipsis if left > 0 else ellipsis
text_size = rl.measure_text_ex(font, display_text, font_size, 0)
# Calculate horizontal position based on alignment
text_x = rect.x + {
@@ -33,7 +47,7 @@ def gui_label(
}.get(alignment_vertical, 0)
# Draw the text in the specified rectangle
rl.draw_text_ex(font, text, rl.Vector2(text_x, text_y), font_size, 0, color)
rl.draw_text_ex(font, display_text, rl.Vector2(text_x, text_y), font_size, 0, color)
def gui_text_box(
@@ -42,7 +56,8 @@ def gui_text_box(
font_size: int = DEFAULT_TEXT_SIZE,
color: rl.Color = DEFAULT_TEXT_COLOR,
alignment: int = rl.GuiTextAlignment.TEXT_ALIGN_LEFT,
alignment_vertical: int = rl.GuiTextAlignmentVertical.TEXT_ALIGN_TOP
alignment_vertical: int = rl.GuiTextAlignmentVertical.TEXT_ALIGN_TOP,
font_weight: FontWeight = FontWeight.NORMAL,
):
styles = [
(rl.GuiControl.DEFAULT, rl.GuiControlProperty.TEXT_COLOR_NORMAL, rl.color_to_int(color)),
@@ -52,6 +67,12 @@ def gui_text_box(
(rl.GuiControl.DEFAULT, rl.GuiDefaultProperty.TEXT_ALIGNMENT_VERTICAL, alignment_vertical),
(rl.GuiControl.DEFAULT, rl.GuiDefaultProperty.TEXT_WRAP_MODE, rl.GuiTextWrapMode.TEXT_WRAP_WORD)
]
if font_weight != FontWeight.NORMAL:
rl.gui_set_font(gui_app.font(font_weight))
with GuiStyleContext(styles):
rl.gui_label(rect, text)
if font_weight != FontWeight.NORMAL:
rl.gui_set_font(gui_app.font(FontWeight.NORMAL))
+380
View File
@@ -0,0 +1,380 @@
import os
import pyray as rl
from dataclasses import dataclass
from collections.abc import Callable
from abc import ABC, abstractmethod
from openpilot.system.ui.lib.scroll_panel import GuiScrollPanel
from openpilot.system.ui.lib.application import gui_app, FontWeight
from openpilot.system.ui.lib.text_measure import measure_text_cached
from openpilot.system.ui.lib.wrap_text import wrap_text
from openpilot.system.ui.lib.button import gui_button
from openpilot.system.ui.lib.toggle import Toggle
from openpilot.system.ui.lib.toggle import WIDTH as TOGGLE_WIDTH, HEIGHT as TOGGLE_HEIGHT
LINE_PADDING = 40
LINE_COLOR = rl.GRAY
ITEM_PADDING = 20
ITEM_SPACING = 80
ITEM_BASE_HEIGHT = 170
ITEM_TEXT_FONT_SIZE = 50
ITEM_TEXT_COLOR = rl.WHITE
ITEM_DESC_TEXT_COLOR = rl.Color(128, 128, 128, 255)
ITEM_DESC_FONT_SIZE = 40
ITEM_DESC_V_OFFSET = 130
RIGHT_ITEM_PADDING = 20
ICON_SIZE = 80
BUTTON_WIDTH = 250
BUTTON_HEIGHT = 100
BUTTON_BORDER_RADIUS = 50
BUTTON_FONT_SIZE = 35
BUTTON_FONT_WEIGHT = FontWeight.MEDIUM
# Abstract base class for right-side items
class RightItem(ABC):
def __init__(self, width: int = 100):
self.width = width
self.enabled = True
@abstractmethod
def draw(self, rect: rl.Rectangle) -> bool:
pass
@abstractmethod
def get_width(self) -> int:
pass
class ToggleRightItem(RightItem):
def __init__(self, initial_state: bool = False, width: int = TOGGLE_WIDTH):
super().__init__(width)
self.toggle = Toggle(initial_state=initial_state)
self.state = initial_state
self.enabled = True
def draw(self, rect: rl.Rectangle) -> bool:
if self.toggle.render(rl.Rectangle(rect.x, rect.y + (rect.height - TOGGLE_HEIGHT) / 2, self.width, TOGGLE_HEIGHT)):
self.state = not self.state
return True
return False
def get_width(self) -> int:
return self.width
def set_state(self, state: bool):
self.state = state
self.toggle.set_state(state)
def get_state(self) -> bool:
return self.state
def set_enabled(self, enabled: bool):
self.enabled = enabled
class ButtonRightItem(RightItem):
def __init__(self, text: str, width: int = BUTTON_WIDTH):
super().__init__(width)
self.text = text
self.enabled = True
def draw(self, rect: rl.Rectangle) -> bool:
return (
gui_button(
rl.Rectangle(rect.x, rect.y + (rect.height - BUTTON_HEIGHT) / 2, BUTTON_WIDTH, BUTTON_HEIGHT),
self.text,
border_radius=BUTTON_BORDER_RADIUS,
font_weight=BUTTON_FONT_WEIGHT,
font_size=BUTTON_FONT_SIZE,
is_enabled=self.enabled,
)
== 1
)
def get_width(self) -> int:
return self.width
def set_enabled(self, enabled: bool):
self.enabled = enabled
class TextRightItem(RightItem):
def __init__(self, text: str, color: rl.Color = ITEM_TEXT_COLOR, font_size: int = ITEM_TEXT_FONT_SIZE):
self.text = text
self.color = color
self.font_size = font_size
font = gui_app.font(FontWeight.NORMAL)
text_width = measure_text_cached(font, text, font_size).x
super().__init__(int(text_width + 20))
def draw(self, rect: rl.Rectangle) -> bool:
font = gui_app.font(FontWeight.NORMAL)
text_size = measure_text_cached(font, self.text, self.font_size)
# Center the text in the allocated rectangle
text_x = rect.x + (rect.width - text_size.x) / 2
text_y = rect.y + (rect.height - text_size.y) / 2
rl.draw_text_ex(font, self.text, rl.Vector2(text_x, text_y), self.font_size, 0, self.color)
return False
def get_width(self) -> int:
return self.width
def set_text(self, text: str):
self.text = text
font = gui_app.font(FontWeight.NORMAL)
text_width = measure_text_cached(font, text, self.font_size).x
self.width = int(text_width + 20)
@dataclass
class ListItem:
title: str
icon: str | None = None
description: str | None = None
description_visible: bool = False
rect: "rl.Rectangle | None" = None
callback: Callable | None = None
right_item: RightItem | None = None
# Cached properties for performance
_wrapped_description: str | None = None
_description_height: float = 0
def get_right_item(self) -> RightItem | None:
return self.right_item
def get_item_height(self, font: rl.Font, max_width: int) -> float:
if self.description_visible and self.description:
if not self._wrapped_description:
wrapped_lines = wrap_text(font, self.description, ITEM_DESC_FONT_SIZE, max_width)
self._wrapped_description = "\n".join(wrapped_lines)
self._description_height = len(wrapped_lines) * 20 + 10 # Line height + padding
return ITEM_BASE_HEIGHT + self._description_height - (ITEM_BASE_HEIGHT - ITEM_DESC_V_OFFSET) + ITEM_SPACING
return ITEM_BASE_HEIGHT
def get_content_width(self, total_width: int) -> int:
if self.right_item:
return total_width - self.right_item.get_width() - RIGHT_ITEM_PADDING
return total_width
def get_right_item_rect(self, item_rect: rl.Rectangle) -> rl.Rectangle:
if not self.right_item:
return rl.Rectangle(0, 0, 0, 0)
right_width = self.right_item.get_width()
right_x = item_rect.x + item_rect.width - right_width
right_y = item_rect.y
return rl.Rectangle(right_x, right_y, right_width, ITEM_BASE_HEIGHT)
class ListView:
def __init__(self, items: list[ListItem]):
self._items: list[ListItem] = items
self._last_dim: tuple[float, float] = (0, 0)
self.scroll_panel = GuiScrollPanel()
self._font_normal = gui_app.font(FontWeight.NORMAL)
# Interaction state
self._hovered_item: int = -1
self._last_mouse_pos = rl.Vector2(0, 0)
self._total_height: float = 0
self._visible_range = (0, 0)
def invalid_height_cache(self):
self._last_dim = (0, 0)
def render(self, rect: rl.Rectangle):
if self._last_dim != (rect.width, rect.height):
self._update_item_rects(rect)
self._last_dim = (rect.width, rect.height)
# Update layout and handle scrolling
content_rect = rl.Rectangle(rect.x, rect.y, rect.width, self._total_height)
scroll_offset = self.scroll_panel.handle_scroll(rect, content_rect)
# Handle mouse interaction
if self.scroll_panel.is_click_valid():
self._handle_mouse_interaction(rect, scroll_offset)
# Set scissor mode for clipping
rl.begin_scissor_mode(int(rect.x), int(rect.y), int(rect.width), int(rect.height))
# Calculate visible range for performance
self._calculate_visible_range(rect, -scroll_offset.y)
# Render only visible items
for i in range(self._visible_range[0], min(self._visible_range[1], len(self._items))):
item = self._items[i]
if item.rect:
adjusted_rect = rl.Rectangle(item.rect.x, item.rect.y + scroll_offset.y, item.rect.width, item.rect.height)
self._render_item(item, adjusted_rect, i)
if i != len(self._items) - 1:
rl.draw_line_ex(
rl.Vector2(adjusted_rect.x + LINE_PADDING, adjusted_rect.y + adjusted_rect.height - 1),
rl.Vector2(
adjusted_rect.x + adjusted_rect.width - LINE_PADDING * 2, adjusted_rect.y + adjusted_rect.height - 1
),
1.0,
LINE_COLOR,
)
rl.end_scissor_mode()
def _render_item(self, item: ListItem, rect: rl.Rectangle, index: int):
content_x = rect.x + ITEM_PADDING
text_x = content_x
# Calculate available width for main content
content_width = item.get_content_width(int(rect.width - ITEM_PADDING * 2))
# Draw icon if present
if item.icon:
icon_texture = gui_app.texture(os.path.join("icons", item.icon), ICON_SIZE, ICON_SIZE)
rl.draw_texture(
icon_texture, int(content_x), int(rect.y + (ITEM_BASE_HEIGHT - icon_texture.width) // 2), rl.WHITE
)
text_x += ICON_SIZE + ITEM_PADDING
# Draw main text
text_size = measure_text_cached(self._font_normal, item.title, ITEM_TEXT_FONT_SIZE)
item_y = rect.y + (ITEM_BASE_HEIGHT - text_size.y) // 2
rl.draw_text_ex(self._font_normal, item.title, rl.Vector2(text_x, item_y), ITEM_TEXT_FONT_SIZE, 0, ITEM_TEXT_COLOR)
# Draw description if visible (adjust width for right item)
if item.description_visible and item._wrapped_description:
desc_y = rect.y + ITEM_DESC_V_OFFSET
desc_max_width = int(content_width - (text_x - content_x))
# Re-wrap description if needed due to right item
if (item.right_item and item.description) and not item._wrapped_description:
wrapped_lines = wrap_text(self._font_normal, item.description, ITEM_DESC_FONT_SIZE, desc_max_width)
item._wrapped_description = "\n".join(wrapped_lines)
rl.draw_text_ex(
self._font_normal,
item._wrapped_description,
rl.Vector2(text_x, desc_y),
ITEM_DESC_FONT_SIZE,
0,
ITEM_DESC_TEXT_COLOR,
)
# Draw right item if present
if item.right_item:
right_rect = item.get_right_item_rect(rect)
# Adjust for scroll offset
right_rect.y = right_rect.y
if item.right_item.draw(right_rect):
# Right item was clicked/activated
if item.callback:
item.callback()
def _update_item_rects(self, container_rect: rl.Rectangle) -> None:
current_y: float = 0.0
self._total_height = 0
for item in self._items:
content_width = item.get_content_width(int(container_rect.width - ITEM_PADDING * 2))
item_height = item.get_item_height(self._font_normal, content_width)
item.rect = rl.Rectangle(container_rect.x, container_rect.y + current_y, container_rect.width, item_height)
current_y += item_height
self._total_height += item_height
def _calculate_visible_range(self, rect: rl.Rectangle, scroll_offset: float):
if not self._items:
self._visible_range = (0, 0)
return
visible_top = scroll_offset
visible_bottom = scroll_offset + rect.height
start_idx = 0
end_idx = len(self._items)
# Find first visible item
for i, item in enumerate(self._items):
if item.rect and item.rect.y + item.rect.height >= visible_top:
start_idx = max(0, i - 1)
break
# Find last visible item
for i in range(start_idx, len(self._items)):
item = self._items[i]
if item.rect and item.rect.y > visible_bottom:
end_idx = min(len(self._items), i + 2)
break
self._visible_range = (start_idx, end_idx)
def _handle_mouse_interaction(self, rect: rl.Rectangle, scroll_offset: rl.Vector2):
mouse_pos = rl.get_mouse_position()
self._hovered_item = -1
if not rl.check_collision_point_rec(mouse_pos, rect):
return
content_mouse_y = mouse_pos.y - rect.y - scroll_offset.y
for i, item in enumerate(self._items):
if item.rect:
# Check if mouse is within this item's bounds in content space
if (
mouse_pos.x >= rect.x
and mouse_pos.x <= rect.x + rect.width
and content_mouse_y >= item.rect.y
and content_mouse_y <= item.rect.y + item.rect.height
):
item_screen_y = item.rect.y + scroll_offset.y
if item_screen_y < rect.height and item_screen_y + item.rect.height > 0:
self._hovered_item = i
break
# Handle click on main item (not right item)
if rl.is_mouse_button_released(rl.MouseButton.MOUSE_BUTTON_LEFT) and self._hovered_item >= 0:
item = self._items[self._hovered_item]
# Check if click was on right item area
if item.right_item and item.rect:
adjusted_rect = rl.Rectangle(item.rect.x, item.rect.y + scroll_offset.y, item.rect.width, item.rect.height)
right_rect = item.get_right_item_rect(adjusted_rect)
if rl.check_collision_point_rec(mouse_pos, right_rect):
# Click was handled by right item, don't process main item click
return
# Toggle description visibility if item has description
if item.description:
item.description_visible = not item.description_visible
# Force layout update when description visibility changes
self._last_dim = (0, 0)
# Call item callback
if item.callback:
item.callback()
# Factory functions
def simple_item(title: str, callback: Callable | None = None) -> ListItem:
return ListItem(title=title, callback=callback)
def toggle_item(
title: str, description: str = None, initial_state: bool = False, callback: Callable | None = None, icon: str = ""
) -> ListItem:
toggle = ToggleRightItem(initial_state=initial_state)
return ListItem(title=title, description=description, right_item=toggle, icon=icon, callback=callback)
def button_item(title: str, button_text: str, description: str = None, callback: Callable | None = None) -> ListItem:
button = ButtonRightItem(text=button_text)
return ListItem(title=title, description=description, right_item=button, callback=callback)
def text_item(title: str, value: str, description: str = None, callback: Callable | None = None) -> ListItem:
text_item = TextRightItem(text=value, color=rl.Color(170, 170, 170, 255))
return ListItem(title=title, description=description, right_item=text_item, callback=callback)
+146 -25
View File
@@ -1,32 +1,61 @@
import pyray as rl
from enum import IntEnum
# Scroll constants for smooth scrolling behavior
MOUSE_WHEEL_SCROLL_SPEED = 30
INERTIA_FRICTION = 0.95 # The rate at which the inertia slows down
MIN_VELOCITY = 0.1 # Minimum velocity before stopping the inertia
INERTIA_FRICTION = 0.92 # The rate at which the inertia slows down
MIN_VELOCITY = 0.5 # Minimum velocity before stopping the inertia
DRAG_THRESHOLD = 5 # Pixels of movement to consider it a drag, not a click
BOUNCE_FACTOR = 0.2 # Elastic bounce when scrolling past boundaries
BOUNCE_RETURN_SPEED = 0.15 # How quickly it returns from the bounce
MAX_BOUNCE_DISTANCE = 150 # Maximum distance for bounce effect
FLICK_MULTIPLIER = 1.8 # Multiplier for flick gestures
VELOCITY_HISTORY_SIZE = 5 # Track velocity over multiple frames for smoother motion
class ScrollState(IntEnum):
IDLE = 0
DRAGGING_CONTENT = 1
DRAGGING_SCROLLBAR = 2
BOUNCING = 3
class GuiScrollPanel:
def __init__(self, show_vertical_scroll_bar: bool = False):
self._scroll_state: ScrollState = ScrollState.IDLE
self._last_mouse_y: float = 0.0
self._start_mouse_y: float = 0.0 # Track the initial mouse position for drag detection
self._offset = rl.Vector2(0, 0)
self._view = rl.Rectangle(0, 0, 0, 0)
self._show_vertical_scroll_bar: bool = show_vertical_scroll_bar
self._velocity_y = 0.0 # Velocity for inertia
self._is_dragging: bool = False
self._bounce_offset: float = 0.0
self._last_frame_time = rl.get_time()
self._velocity_history: list[float] = []
self._last_drag_time: float = 0.0
self._content_rect: rl.Rectangle | None = None
self._bounds_rect: rl.Rectangle | None = None
def handle_scroll(self, bounds: rl.Rectangle, content: rl.Rectangle) -> rl.Vector2:
mouse_pos = rl.get_mouse_position()
# Store rectangles for reference
self._content_rect = content
self._bounds_rect = bounds
# Handle dragging logic
# Calculate time delta
current_time = rl.get_time()
delta_time = current_time - self._last_frame_time
self._last_frame_time = current_time
# Prevent large jumps
delta_time = min(delta_time, 0.05)
mouse_pos = rl.get_mouse_position()
max_scroll_y = max(content.height - bounds.height, 0)
# Start dragging on mouse press
if rl.check_collision_point_rec(mouse_pos, bounds) and rl.is_mouse_button_pressed(rl.MouseButton.MOUSE_BUTTON_LEFT):
if self._scroll_state == ScrollState.IDLE:
if self._scroll_state == ScrollState.IDLE or self._scroll_state == ScrollState.BOUNCING:
self._scroll_state = ScrollState.DRAGGING_CONTENT
if self._show_vertical_scroll_bar:
scrollbar_width = rl.gui_get_style(rl.GuiControl.LISTVIEW, rl.GuiListViewProperty.SCROLLBAR_WIDTH)
@@ -35,41 +64,133 @@ class GuiScrollPanel:
self._scroll_state = ScrollState.DRAGGING_SCROLLBAR
self._last_mouse_y = mouse_pos.y
self._velocity_y = 0.0 # Reset velocity when drag starts
self._start_mouse_y = mouse_pos.y
self._last_drag_time = current_time
self._velocity_history = []
self._velocity_y = 0.0
self._bounce_offset = 0.0
self._is_dragging = False
if self._scroll_state != ScrollState.IDLE:
# Handle active dragging
if self._scroll_state == ScrollState.DRAGGING_CONTENT or self._scroll_state == ScrollState.DRAGGING_SCROLLBAR:
if rl.is_mouse_button_down(rl.MouseButton.MOUSE_BUTTON_LEFT):
delta_y = mouse_pos.y - self._last_mouse_y
# Track velocity for inertia
time_since_last_drag = current_time - self._last_drag_time
if time_since_last_drag > 0:
drag_velocity = delta_y / time_since_last_drag / 60.0
self._velocity_history.append(drag_velocity)
if len(self._velocity_history) > VELOCITY_HISTORY_SIZE:
self._velocity_history.pop(0)
self._last_drag_time = current_time
# Detect actual dragging
total_drag = abs(mouse_pos.y - self._start_mouse_y)
if total_drag > DRAG_THRESHOLD:
self._is_dragging = True
if self._scroll_state == ScrollState.DRAGGING_CONTENT:
# Add resistance at boundaries
if (self._offset.y > 0 and delta_y > 0) or (self._offset.y < -max_scroll_y and delta_y < 0):
delta_y *= BOUNCE_FACTOR
self._offset.y += delta_y
else:
delta_y = -delta_y
elif self._scroll_state == ScrollState.DRAGGING_SCROLLBAR:
scroll_ratio = content.height / bounds.height
self._offset.y -= delta_y * scroll_ratio
self._last_mouse_y = mouse_pos.y
self._velocity_y = delta_y # Update velocity during drag
else:
self._scroll_state = ScrollState.IDLE
# Handle mouse wheel scrolling
elif rl.is_mouse_button_released(rl.MouseButton.MOUSE_BUTTON_LEFT):
# Calculate flick velocity
if self._velocity_history:
total_weight = 0
weighted_velocity = 0.0
for i, v in enumerate(self._velocity_history):
weight = i + 1
weighted_velocity += v * weight
total_weight += weight
if total_weight > 0:
avg_velocity = weighted_velocity / total_weight
self._velocity_y = avg_velocity * FLICK_MULTIPLIER
# Check bounds
if self._offset.y > 0 or self._offset.y < -max_scroll_y:
self._scroll_state = ScrollState.BOUNCING
else:
self._scroll_state = ScrollState.IDLE
# Handle mouse wheel
wheel_move = rl.get_mouse_wheel_move()
if self._show_vertical_scroll_bar:
self._offset.y += wheel_move * (MOUSE_WHEEL_SCROLL_SPEED - 20)
rl.gui_scroll_panel(bounds, rl.ffi.NULL, content, self._offset, self._view)
else:
self._offset.y += wheel_move * MOUSE_WHEEL_SCROLL_SPEED
if wheel_move != 0:
self._velocity_y = 0.0
if self._show_vertical_scroll_bar:
self._offset.y += wheel_move * (MOUSE_WHEEL_SCROLL_SPEED - 20)
rl.gui_scroll_panel(bounds, rl.ffi.NULL, content, self._offset, self._view)
else:
self._offset.y += wheel_move * MOUSE_WHEEL_SCROLL_SPEED
if self._offset.y > 0 or self._offset.y < -max_scroll_y:
self._scroll_state = ScrollState.BOUNCING
# Apply inertia (continue scrolling after mouse release)
if self._scroll_state == ScrollState.IDLE:
self._offset.y += self._velocity_y
self._velocity_y *= INERTIA_FRICTION # Slow down velocity over time
if abs(self._velocity_y) > MIN_VELOCITY:
self._offset.y += self._velocity_y
self._velocity_y *= INERTIA_FRICTION
# Stop scrolling when velocity is low
if abs(self._velocity_y) < MIN_VELOCITY:
if self._offset.y > 0 or self._offset.y < -max_scroll_y:
self._scroll_state = ScrollState.BOUNCING
else:
self._velocity_y = 0.0
# Ensure scrolling doesn't go beyond bounds
max_scroll_y = max(content.height - bounds.height, 0)
self._offset.y = max(min(self._offset.y, 0), -max_scroll_y)
# Handle bouncing effect
elif self._scroll_state == ScrollState.BOUNCING:
target_y = 0.0
if self._offset.y < -max_scroll_y:
target_y = -max_scroll_y
distance = target_y - self._offset.y
bounce_step = distance * BOUNCE_RETURN_SPEED
self._offset.y += bounce_step
self._velocity_y *= INERTIA_FRICTION * 0.8
if abs(distance) < 0.5 and abs(self._velocity_y) < MIN_VELOCITY:
self._offset.y = target_y
self._velocity_y = 0.0
self._scroll_state = ScrollState.IDLE
# Limit bounce distance
if self._scroll_state != ScrollState.DRAGGING_CONTENT:
if self._offset.y > MAX_BOUNCE_DISTANCE:
self._offset.y = MAX_BOUNCE_DISTANCE
elif self._offset.y < -(max_scroll_y + MAX_BOUNCE_DISTANCE):
self._offset.y = -(max_scroll_y + MAX_BOUNCE_DISTANCE)
return self._offset
def is_click_valid(self) -> bool:
# Check if this is a click rather than a drag
return (
self._scroll_state == ScrollState.IDLE
and not self._is_dragging
and rl.is_mouse_button_released(rl.MouseButton.MOUSE_BUTTON_LEFT)
)
def get_normalized_scroll_position(self) -> float:
"""Returns the current scroll position as a value from 0.0 to 1.0"""
if not self._content_rect or not self._bounds_rect:
return 0.0
max_scroll_y = max(self._content_rect.height - self._bounds_rect.height, 0)
if max_scroll_y == 0:
return 0.0
normalized = -self._offset.y / max_scroll_y
return max(0.0, min(1.0, normalized))
+315
View File
@@ -0,0 +1,315 @@
import pyray as rl
import numpy as np
from typing import Any
MAX_GRADIENT_COLORS = 15
FRAGMENT_SHADER = """
#version 300 es
precision mediump float;
in vec2 fragTexCoord;
out vec4 finalColor;
uniform vec2 points[100];
uniform int pointCount;
uniform vec4 fillColor;
uniform vec2 resolution;
uniform int useGradient;
uniform vec2 gradientStart;
uniform vec2 gradientEnd;
uniform vec4 gradientColors[15];
uniform float gradientStops[15];
uniform int gradientColorCount;
vec4 getGradientColor(vec2 pos) {
vec2 gradientDir = gradientEnd - gradientStart;
float gradientLength = length(gradientDir);
if (gradientLength < 0.001) return gradientColors[0];
vec2 normalizedDir = gradientDir / gradientLength;
float t = clamp(dot(pos - gradientStart, normalizedDir) / gradientLength, 0.0, 1.0);
if (gradientColorCount <= 1) return gradientColors[0];
for (int i = 0; i < gradientColorCount - 1; i++) {
if (t >= gradientStops[i] && t <= gradientStops[i+1]) {
float segmentT = (t - gradientStops[i]) / (gradientStops[i+1] - gradientStops[i]);
return mix(gradientColors[i], gradientColors[i+1], segmentT);
}
}
return gradientColors[gradientColorCount-1];
}
bool isPointInsidePolygon(vec2 p) {
if (pointCount < 3) return false;
int crossings = 0;
for (int i = 0, j = pointCount - 1; i < pointCount; j = i++) {
vec2 pi = points[i];
vec2 pj = points[j];
if (distance(pi, pj) < 0.001) continue;
if (((pi.y > p.y) != (pj.y > p.y)) &&
(p.x < (pj.x - pi.x) * (p.y - pi.y) / (pj.y - pi.y + 0.001) + pi.x)) {
crossings++;
}
}
return (crossings & 1) == 1;
}
float distanceToEdge(vec2 p) {
float minDist = 1000.0;
for (int i = 0, j = pointCount - 1; i < pointCount; j = i++) {
vec2 edge0 = points[j];
vec2 edge1 = points[i];
if (distance(edge0, edge1) < 0.0001) continue;
vec2 v1 = p - edge0;
vec2 v2 = edge1 - edge0;
float l2 = dot(v2, v2);
if (l2 < 0.0001) {
float dist = length(v1);
minDist = min(minDist, dist);
continue;
}
float t = clamp(dot(v1, v2) / l2, 0.0, 1.0);
vec2 projection = edge0 + t * v2;
float dist = length(p - projection);
minDist = min(minDist, dist);
}
return minDist;
}
void main() {
vec2 pixel = fragTexCoord * resolution;
// Compute pixel size for anti-aliasing
vec2 pixelGrad = vec2(dFdx(pixel.x), dFdy(pixel.y));
float pixelSize = length(pixelGrad);
float aaWidth = max(0.5, pixelSize * 1.5);
bool inside = isPointInsidePolygon(pixel);
if (inside) {
finalColor = useGradient == 1 ? getGradientColor(pixel) : fillColor;
return;
}
float sd = -distanceToEdge(pixel);
float alpha = smoothstep(-aaWidth, aaWidth, sd);
if (alpha > 0.0){
vec4 color = useGradient == 1 ? getGradientColor(pixel) : fillColor;
finalColor = vec4(color.rgb, color.a * alpha);
} else {
finalColor = vec4(0.0);
}
}
"""
# Default vertex shader
VERTEX_SHADER = """
#version 300 es
in vec3 vertexPosition;
in vec2 vertexTexCoord;
out vec2 fragTexCoord;
uniform mat4 mvp;
void main() {
fragTexCoord = vertexTexCoord;
gl_Position = mvp * vec4(vertexPosition, 1.0);
}
"""
UNIFORM_INT = rl.ShaderUniformDataType.SHADER_UNIFORM_INT
UNIFORM_FLOAT = rl.ShaderUniformDataType.SHADER_UNIFORM_FLOAT
UNIFORM_VEC2 = rl.ShaderUniformDataType.SHADER_UNIFORM_VEC2
UNIFORM_VEC4 = rl.ShaderUniformDataType.SHADER_UNIFORM_VEC4
class ShaderState:
_instance: Any = None
@classmethod
def get_instance(cls):
if cls._instance is None:
cls._instance = cls()
return cls._instance
def __init__(self):
if ShaderState._instance is not None:
raise Exception("This class is a singleton. Use get_instance() instead.")
self.initialized = False
self.shader = None
self.white_texture = None
# Shader uniform locations
self.locations = {
'pointCount': None,
'fillColor': None,
'resolution': None,
'points': None,
'useGradient': None,
'gradientStart': None,
'gradientEnd': None,
'gradientColors': None,
'gradientStops': None,
'gradientColorCount': None,
'mvp': None,
}
# Pre-allocated FFI objects
self.point_count_ptr = rl.ffi.new("int[]", [0])
self.resolution_ptr = rl.ffi.new("float[]", [0.0, 0.0])
self.fill_color_ptr = rl.ffi.new("float[]", [0.0, 0.0, 0.0, 0.0])
self.use_gradient_ptr = rl.ffi.new("int[]", [0])
self.gradient_start_ptr = rl.ffi.new("float[]", [0.0, 0.0])
self.gradient_end_ptr = rl.ffi.new("float[]", [0.0, 0.0])
self.color_count_ptr = rl.ffi.new("int[]", [0])
self.gradient_colors_ptr = rl.ffi.new("float[]", MAX_GRADIENT_COLORS * 4)
self.gradient_stops_ptr = rl.ffi.new("float[]", MAX_GRADIENT_COLORS)
def initialize(self):
if self.initialized:
return
self.shader = rl.load_shader_from_memory(VERTEX_SHADER, FRAGMENT_SHADER)
# Create and cache white texture
white_img = rl.gen_image_color(2, 2, rl.WHITE)
self.white_texture = rl.load_texture_from_image(white_img)
rl.set_texture_filter(self.white_texture, rl.TEXTURE_FILTER_BILINEAR)
rl.unload_image(white_img)
# Cache all uniform locations
for uniform in self.locations.keys():
self.locations[uniform] = rl.get_shader_location(self.shader, uniform)
# Setup default MVP matrix
mvp_ptr = rl.ffi.new("float[16]", [1.0, 0.0, 0.0, 0.0, 0.0, 1.0, 0.0, 0.0, 0.0, 0.0, 1.0, 0.0, 0.0, 0.0, 0.0, 1.0])
rl.set_shader_value_matrix(self.shader, self.locations['mvp'], rl.Matrix(*mvp_ptr))
self.initialized = True
def cleanup(self):
if not self.initialized:
return
if self.white_texture:
rl.unload_texture(self.white_texture)
self.white_texture = None
if self.shader:
rl.unload_shader(self.shader)
self.shader = None
self.initialized = False
def _configure_shader_color(state, color, gradient, clipped_rect, original_rect):
use_gradient = 1 if gradient else 0
state.use_gradient_ptr[0] = use_gradient
rl.set_shader_value(state.shader, state.locations['useGradient'], state.use_gradient_ptr, UNIFORM_INT)
if use_gradient:
start = np.array(gradient['start']) * np.array([original_rect.width, original_rect.height]) + np.array([original_rect.x, original_rect.y])
end = np.array(gradient['end']) * np.array([original_rect.width, original_rect.height]) + np.array([original_rect.x, original_rect.y])
start = start - np.array([clipped_rect.x, clipped_rect.y])
end = end - np.array([clipped_rect.x, clipped_rect.y])
state.gradient_start_ptr[0:2] = start.astype(np.float32)
state.gradient_end_ptr[0:2] = end.astype(np.float32)
rl.set_shader_value(state.shader, state.locations['gradientStart'], state.gradient_start_ptr, UNIFORM_VEC2)
rl.set_shader_value(state.shader, state.locations['gradientEnd'], state.gradient_end_ptr, UNIFORM_VEC2)
colors = gradient['colors']
color_count = min(len(colors), MAX_GRADIENT_COLORS)
state.color_count_ptr[0] = color_count
for i, c in enumerate(colors[:color_count]):
base_idx = i * 4
state.gradient_colors_ptr[base_idx:base_idx+4] = [c.r / 255.0, c.g / 255.0, c.b / 255.0, c.a / 255.0]
rl.set_shader_value_v(state.shader, state.locations['gradientColors'], state.gradient_colors_ptr, UNIFORM_VEC4, color_count)
stops = gradient.get('stops', [i / max(1, color_count - 1) for i in range(color_count)])
stops = np.clip(stops[:color_count], 0.0, 1.0)
state.gradient_stops_ptr[0:color_count] = stops
rl.set_shader_value_v(state.shader, state.locations['gradientStops'], state.gradient_stops_ptr, UNIFORM_FLOAT, color_count)
rl.set_shader_value(state.shader, state.locations['gradientColorCount'], state.color_count_ptr, UNIFORM_INT)
else:
color = color or rl.WHITE
state.fill_color_ptr[0:4] = [color.r / 255.0, color.g / 255.0, color.b / 255.0, color.a / 255.0]
rl.set_shader_value(state.shader, state.locations['fillColor'], state.fill_color_ptr, UNIFORM_VEC4)
def draw_polygon(origin_rect: rl.Rectangle, points: np.ndarray, color=None, gradient=None):
"""
Draw a complex polygon using shader-based even-odd fill rule
Args:
rect: Rectangle defining the drawing area
points: numpy array of (x,y) points defining the polygon
color: Solid fill color (rl.Color)
gradient: Dict with gradient parameters:
{
'start': (x1, y1), # Start point (normalized 0-1)
'end': (x2, y2), # End point (normalized 0-1)
'colors': [rl.Color], # List of colors at stops
'stops': [float] # List of positions (0-1)
}
"""
if len(points) < 3:
return
state = ShaderState.get_instance()
if not state.initialized:
state.initialize()
# Find bounding box
min_xy = np.min(points, axis=0)
max_xy = np.max(points, axis=0)
clip_x = max(origin_rect.x, min_xy[0])
clip_y = max(origin_rect.y, min_xy[1])
clip_right = min(origin_rect.x + origin_rect.width, max_xy[0])
clip_bottom = min(origin_rect.y + origin_rect.height, max_xy[1])
# Check if polygon is completely off-screen
if clip_x >= clip_right or clip_y >= clip_bottom:
return
clipped_rect = rl.Rectangle(clip_x, clip_y, clip_right - clip_x, clip_bottom - clip_y)
# Transform points relative to the CLIPPED area
transformed_points = points - np.array([clip_x, clip_y])
# Set shader values
state.point_count_ptr[0] = len(transformed_points)
rl.set_shader_value(state.shader, state.locations['pointCount'], state.point_count_ptr, UNIFORM_INT)
state.resolution_ptr[0:2] = [clipped_rect.width, clipped_rect.height]
rl.set_shader_value(state.shader, state.locations['resolution'], state.resolution_ptr, UNIFORM_VEC2)
flat_points = np.ascontiguousarray(transformed_points.flatten().astype(np.float32))
points_ptr = rl.ffi.cast("float *", flat_points.ctypes.data)
rl.set_shader_value_v(state.shader, state.locations['points'], points_ptr, UNIFORM_VEC2, len(transformed_points))
_configure_shader_color(state, color, gradient, clipped_rect, origin_rect)
# Render
rl.begin_shader_mode(state.shader)
rl.draw_texture_pro(
state.white_texture,
rl.Rectangle(0, 0, 2, 2),
clipped_rect,
rl.Vector2(0, 0),
0.0,
rl.WHITE,
)
rl.end_shader_mode()
def cleanup_shader_resources():
state = ShaderState.get_instance()
state.cleanup()
+13
View File
@@ -0,0 +1,13 @@
import pyray as rl
_cache: dict[int, rl.Vector2] = {}
def measure_text_cached(font: rl.Font, text: str, font_size: int, spacing: int = 0) -> rl.Vector2:
key = hash((font.texture.id, text, font_size, spacing))
if key in _cache:
return _cache[key]
result = rl.measure_text_ex(font, text, font_size, spacing)
_cache[key] = result
return result
+54
View File
@@ -0,0 +1,54 @@
import pyray as rl
ON_COLOR = rl.Color(51, 171, 76, 255)
OFF_COLOR = rl.Color(0x39, 0x39, 0x39, 255)
KNOB_COLOR = rl.WHITE
WIDTH, HEIGHT = 160, 80
BG_HEIGHT = 60
ANIMATION_SPEED = 8.0
class Toggle:
def __init__(self, initial_state=False):
self._state = initial_state
self._rect = rl.Rectangle(0, 0, WIDTH, HEIGHT)
self._progress = 1.0 if initial_state else 0.0
self._target = self._progress
def handle_input(self):
if rl.is_mouse_button_pressed(rl.MOUSE_LEFT_BUTTON):
if rl.check_collision_point_rec(rl.get_mouse_position(), self._rect):
self._state = not self._state
self._target = 1.0 if self._state else 0.0
return 1
return 0
def get_state(self):
return self._state
def set_state(self, state: bool):
self._state = state
def update(self):
if abs(self._progress - self._target) > 0.01:
delta = rl.get_frame_time() * ANIMATION_SPEED
self._progress += delta if self._progress < self._target else -delta
self._progress = max(0.0, min(1.0, self._progress))
def render(self, rect: rl.Rectangle):
self._rect.x, self._rect.y = rect.x, rect.y
self. update()
# Draw background
bg_rect = rl.Rectangle(self._rect.x + 5, self._rect.y + 10, WIDTH - 10, BG_HEIGHT)
bg_color = self._blend_color(OFF_COLOR, ON_COLOR, self._progress)
rl.draw_rectangle_rounded(bg_rect, 1.0, 10, bg_color)
# Draw knob
knob_x = self._rect.x + HEIGHT / 2 + (WIDTH - HEIGHT) * self._progress
knob_y = self._rect.y + HEIGHT / 2
rl.draw_circle(int(knob_x), int(knob_y), HEIGHT / 2, KNOB_COLOR)
return self.handle_input()
def _blend_color(self, c1, c2, t):
return rl.Color(int(c1.r + (c2.r - c1.r) * t), int(c1.g + (c2.g - c1.g) * t), int(c1.b + (c2.b - c1.b) * t), 255)
+716
View File
@@ -0,0 +1,716 @@
import asyncio
import concurrent.futures
import copy
import threading
import time
import uuid
from collections.abc import Callable
from dataclasses import dataclass
from enum import IntEnum
from typing import TypeVar
from dbus_next.aio import MessageBus
from dbus_next import BusType, Variant, Message
from dbus_next.errors import DBusError
from dbus_next.constants import MessageType
try:
from openpilot.common.params import Params
except ImportError:
# Params/Cythonized modules are not available in zipapp
Params = None
from openpilot.common.swaglog import cloudlog
T = TypeVar("T")
# NetworkManager constants
NM = "org.freedesktop.NetworkManager"
NM_PATH = '/org/freedesktop/NetworkManager'
NM_IFACE = 'org.freedesktop.NetworkManager'
NM_SETTINGS_PATH = '/org/freedesktop/NetworkManager/Settings'
NM_SETTINGS_IFACE = 'org.freedesktop.NetworkManager.Settings'
NM_CONNECTION_IFACE = 'org.freedesktop.NetworkManager.Settings.Connection'
NM_WIRELESS_IFACE = 'org.freedesktop.NetworkManager.Device.Wireless'
NM_PROPERTIES_IFACE = 'org.freedesktop.DBus.Properties'
NM_DEVICE_IFACE = "org.freedesktop.NetworkManager.Device"
NM_DEVICE_STATE_REASON_SUPPLICANT_DISCONNECT = 8
TETHERING_IP_ADDRESS = "192.168.43.1"
DEFAULT_TETHERING_PASSWORD = "12345678"
# NetworkManager device states
class NMDeviceState(IntEnum):
DISCONNECTED = 30
PREPARE = 40
NEED_AUTH = 60
IP_CONFIG = 70
ACTIVATED = 100
class SecurityType(IntEnum):
OPEN = 0
WPA = 1
WPA2 = 2
WPA3 = 3
UNSUPPORTED = 4
@dataclass
class NetworkInfo:
ssid: str
strength: int
is_connected: bool
security_type: SecurityType
path: str
bssid: str
is_saved: bool = False
# saved_path: str
@dataclass
class WifiManagerCallbacks:
need_auth: Callable[[str], None] | None = None
activated: Callable[[], None] | None = None
forgotten: Callable[[str], None] | None = None
networks_updated: Callable[[list[NetworkInfo]], None] | None = None
connection_failed: Callable[[str, str], None] | None = None # Added for error feedback
class WifiManager:
def __init__(self, callbacks):
self.callbacks: WifiManagerCallbacks = callbacks
self.networks: list[NetworkInfo] = []
self.bus: MessageBus = None
self.device_path: str = ""
self.device_proxy = None
self.saved_connections: dict[str, str] = {}
self.active_ap_path: str = ""
self.scan_task: asyncio.Task | None = None
# Set tethering ssid as "weedle" + first 4 characters of a dongle id
self._tethering_ssid = "weedle"
if Params is not None:
dongle_id = Params().get("DongleId", encoding="utf-8")
if dongle_id:
self._tethering_ssid += "-" + dongle_id[:4]
self.running: bool = True
self._current_connection_ssid: str | None = None
async def connect(self) -> None:
"""Connect to the DBus system bus."""
try:
self.bus = await MessageBus(bus_type=BusType.SYSTEM).connect()
if not await self._find_wifi_device():
raise ValueError("No Wi-Fi device found")
await self._setup_signals(self.device_path)
self.active_ap_path = await self.get_active_access_point()
await self.add_tethering_connection(self._tethering_ssid, DEFAULT_TETHERING_PASSWORD)
self.saved_connections = await self._get_saved_connections()
self.scan_task = asyncio.create_task(self._periodic_scan())
except DBusError as e:
cloudlog.error(f"Failed to connect to DBus: {e}")
raise
except Exception as e:
cloudlog.error(f"Unexpected error during connect: {e}")
raise
async def shutdown(self) -> None:
self.running = False
if self.scan_task:
self.scan_task.cancel()
try:
await self.scan_task
except asyncio.CancelledError:
pass
if self.bus:
self.bus.disconnect()
async def _request_scan(self) -> None:
try:
interface = self.device_proxy.get_interface(NM_WIRELESS_IFACE)
await interface.call_request_scan({})
except DBusError as e:
cloudlog.warning(f"Scan request failed: {str(e)}")
async def get_active_access_point(self):
try:
props_iface = self.device_proxy.get_interface(NM_PROPERTIES_IFACE)
ap_path = await props_iface.call_get(NM_WIRELESS_IFACE, 'ActiveAccessPoint')
return ap_path.value
except DBusError as e:
cloudlog.error(f"Error fetching active access point: {str(e)}")
return ''
async def forget_connection(self, ssid: str) -> bool:
path = self.saved_connections.get(ssid)
if not path:
return False
try:
nm_iface = await self._get_interface(NM, path, NM_CONNECTION_IFACE)
await nm_iface.call_delete()
if self._current_connection_ssid == ssid:
self._current_connection_ssid = None
if ssid in self.saved_connections:
del self.saved_connections[ssid]
for network in self.networks:
if network.ssid == ssid:
network.is_saved = False
network.is_connected = False
break
# Notify UI of forgotten connection
if self.callbacks.networks_updated:
self.callbacks.networks_updated(copy.deepcopy(self.networks))
return True
except DBusError as e:
cloudlog.error(f"Failed to delete connection for SSID: {ssid}. Error: {e}")
return False
async def activate_connection(self, ssid: str) -> bool:
connection_path = self.saved_connections.get(ssid)
if not connection_path:
return False
try:
nm_iface = await self._get_interface(NM, NM_PATH, NM_IFACE)
await nm_iface.call_activate_connection(connection_path, self.device_path, "/")
return True
except DBusError as e:
cloudlog.error(f"Failed to activate connection {ssid}: {str(e)}")
return False
async def connect_to_network(self, ssid: str, password: str = None, bssid: str = None, is_hidden: bool = False) -> None:
"""Connect to a selected Wi-Fi network."""
try:
self._current_connection_ssid = ssid
if ssid in self.saved_connections:
# Forget old connection if new password provided
if password:
await self.forget_connection(ssid)
await asyncio.sleep(0.2) # NetworkManager delay
else:
# Just activate existing connection
await self.activate_connection(ssid)
return
connection = {
'connection': {
'type': Variant('s', '802-11-wireless'),
'uuid': Variant('s', str(uuid.uuid4())),
'id': Variant('s', ssid),
'autoconnect-retries': Variant('i', 0),
},
'802-11-wireless': {
'ssid': Variant('ay', ssid.encode('utf-8')),
'hidden': Variant('b', is_hidden),
'mode': Variant('s', 'infrastructure'),
},
'ipv4': {'method': Variant('s', 'auto')},
'ipv6': {'method': Variant('s', 'ignore')},
}
if bssid:
connection['802-11-wireless']['bssid'] = Variant('ay', bssid.encode('utf-8'))
if password:
connection['802-11-wireless-security'] = {
'key-mgmt': Variant('s', 'wpa-psk'),
'auth-alg': Variant('s', 'open'),
'psk': Variant('s', password),
}
nm_iface = await self._get_interface(NM, NM_PATH, NM_IFACE)
await nm_iface.call_add_and_activate_connection(connection, self.device_path, "/")
except Exception as e:
self._current_connection_ssid = None
cloudlog.error(f"Error connecting to network: {e}")
# Notify UI of failure
if self.callbacks.connection_failed:
self.callbacks.connection_failed(ssid, str(e))
def is_saved(self, ssid: str) -> bool:
return ssid in self.saved_connections
async def _find_wifi_device(self) -> bool:
nm_iface = await self._get_interface(NM, NM_PATH, NM_IFACE)
devices = await nm_iface.get_devices()
for device_path in devices:
device = await self.bus.introspect(NM, device_path)
device_proxy = self.bus.get_proxy_object(NM, device_path, device)
device_interface = device_proxy.get_interface(NM_DEVICE_IFACE)
device_type = await device_interface.get_device_type() # type: ignore[attr-defined]
if device_type == 2: # Wi-Fi device
self.device_path = device_path
self.device_proxy = device_proxy
return True
return False
async def add_tethering_connection(self, ssid: str, password: str = "12345678") -> bool:
"""Create a WiFi tethering connection."""
if len(password) < 8:
print("Tethering password must be at least 8 characters")
return False
try:
# First, check if a hotspot connection already exists
settings_iface = await self._get_interface(NM, NM_SETTINGS_PATH, NM_SETTINGS_IFACE)
connection_paths = await settings_iface.call_list_connections()
# Look for an existing hotspot connection
for path in connection_paths:
try:
settings = await self._get_connection_settings(path)
conn_type = settings.get('connection', {}).get('type', Variant('s', '')).value
wifi_mode = settings.get('802-11-wireless', {}).get('mode', Variant('s', '')).value
if conn_type == '802-11-wireless' and wifi_mode == 'ap':
# Extract the SSID to check
connection_ssid = self._extract_ssid(settings)
if connection_ssid == ssid:
return True
except DBusError:
continue
connection = {
'connection': {
'id': Variant('s', 'Hotspot'),
'uuid': Variant('s', str(uuid.uuid4())),
'type': Variant('s', '802-11-wireless'),
'interface-name': Variant('s', 'wlan0'),
'autoconnect': Variant('b', False),
},
'802-11-wireless': {
'band': Variant('s', 'bg'),
'mode': Variant('s', 'ap'),
'ssid': Variant('ay', ssid.encode('utf-8')),
},
'802-11-wireless-security': {
'group': Variant('as', ['ccmp']),
'key-mgmt': Variant('s', 'wpa-psk'),
'pairwise': Variant('as', ['ccmp']),
'proto': Variant('as', ['rsn']),
'psk': Variant('s', password),
},
'ipv4': {
'method': Variant('s', 'shared'),
'address-data': Variant('aa{sv}', [{'address': Variant('s', TETHERING_IP_ADDRESS), 'prefix': Variant('u', 24)}]),
'gateway': Variant('s', TETHERING_IP_ADDRESS),
'never-default': Variant('b', True),
},
'ipv6': {
'method': Variant('s', 'ignore'),
},
}
settings_iface = await self._get_interface(NM, NM_SETTINGS_PATH, NM_SETTINGS_IFACE)
new_connection = await settings_iface.call_add_connection(connection)
print(f"Added tethering connection with path: {new_connection}")
return True
except DBusError as e:
print(f"Failed to add tethering connection: {e}")
return False
except Exception as e:
print(f"Unexpected error adding tethering connection: {e}")
return False
async def get_tethering_password(self) -> str:
"""Get the current tethering password."""
try:
hotspot_path = self.saved_connections.get(self._tethering_ssid)
if hotspot_path:
conn_iface = await self._get_interface(NM, hotspot_path, NM_CONNECTION_IFACE)
secrets = await conn_iface.call_get_secrets('802-11-wireless-security')
if secrets and '802-11-wireless-security' in secrets:
psk = secrets.get('802-11-wireless-security', {}).get('psk', Variant('s', '')).value
return str(psk) if psk is not None else ""
return ""
except DBusError as e:
print(f"Failed to get tethering password: {e}")
return ""
except Exception as e:
print(f"Unexpected error getting tethering password: {e}")
return ""
async def set_tethering_password(self, password: str) -> bool:
"""Set the tethering password."""
if len(password) < 8:
cloudlog.error("Tethering password must be at least 8 characters")
return False
try:
hotspot_path = self.saved_connections.get(self._tethering_ssid)
if not hotspot_path:
print("No hotspot connection found")
return False
# Update the connection settings with new password
settings = await self._get_connection_settings(hotspot_path)
if '802-11-wireless-security' not in settings:
settings['802-11-wireless-security'] = {}
settings['802-11-wireless-security']['psk'] = Variant('s', password)
# Apply changes
conn_iface = await self._get_interface(NM, hotspot_path, NM_CONNECTION_IFACE)
await conn_iface.call_update(settings)
# Check if connection is active and restart if needed
is_active = False
nm_iface = await self._get_interface(NM, NM_PATH, NM_IFACE)
active_connections = await nm_iface.get_active_connections()
for conn_path in active_connections:
props_iface = await self._get_interface(NM, conn_path, NM_PROPERTIES_IFACE)
conn_id_path = await props_iface.call_get('org.freedesktop.NetworkManager.Connection.Active', 'Connection')
if conn_id_path.value == hotspot_path:
is_active = True
await nm_iface.call_deactivate_connection(conn_path)
break
if is_active:
await nm_iface.call_activate_connection(hotspot_path, self.device_path, "/")
print("Tethering password updated successfully")
return True
except DBusError as e:
print(f"Failed to set tethering password: {e}")
return False
except Exception as e:
print(f"Unexpected error setting tethering password: {e}")
return False
async def is_tethering_active(self) -> bool:
"""Check if tethering is active for the specified SSID."""
try:
hotspot_path = self.saved_connections.get(self._tethering_ssid)
if not hotspot_path:
return False
nm_iface = await self._get_interface(NM, NM_PATH, NM_IFACE)
active_connections = await nm_iface.get_active_connections()
for conn_path in active_connections:
props_iface = await self._get_interface(NM, conn_path, NM_PROPERTIES_IFACE)
conn_id_path = await props_iface.call_get('org.freedesktop.NetworkManager.Connection.Active', 'Connection')
if conn_id_path.value == hotspot_path:
return True
return False
except Exception:
return False
async def _periodic_scan(self):
while self.running:
try:
await self._request_scan()
await asyncio.sleep(30)
except asyncio.CancelledError:
break
except DBusError as e:
cloudlog.error(f"Scan failed: {e}")
await asyncio.sleep(5)
async def _setup_signals(self, device_path: str) -> None:
rules = [
f"type='signal',interface='{NM_PROPERTIES_IFACE}',member='PropertiesChanged',path='{device_path}'",
f"type='signal',interface='{NM_DEVICE_IFACE}',member='StateChanged',path='{device_path}'",
f"type='signal',interface='{NM_SETTINGS_IFACE}',member='NewConnection',path='{NM_SETTINGS_PATH}'",
f"type='signal',interface='{NM_SETTINGS_IFACE}',member='ConnectionRemoved',path='{NM_SETTINGS_PATH}'",
]
for rule in rules:
await self._add_match_rule(rule)
# Set up signal handlers
self.device_proxy.get_interface(NM_PROPERTIES_IFACE).on_properties_changed(self._on_properties_changed)
self.device_proxy.get_interface(NM_DEVICE_IFACE).on_state_changed(self._on_state_changed)
settings_iface = await self._get_interface(NM, NM_SETTINGS_PATH, NM_SETTINGS_IFACE)
settings_iface.on_new_connection(self._on_new_connection)
settings_iface.on_connection_removed(self._on_connection_removed)
def _on_properties_changed(self, interface: str, changed: dict, invalidated: list):
# print("property changed", interface, changed, invalidated)
if 'LastScan' in changed:
asyncio.create_task(self._refresh_networks())
elif interface == NM_WIRELESS_IFACE and "ActiveAccessPoint" in changed:
new_ap_path = changed["ActiveAccessPoint"].value
if self.active_ap_path != new_ap_path:
self.active_ap_path = new_ap_path
asyncio.create_task(self._refresh_networks())
def _on_state_changed(self, new_state: int, old_state: int, reason: int):
print("State changed", new_state, old_state, reason)
if new_state == NMDeviceState.ACTIVATED:
if self.callbacks.activated:
self.callbacks.activated()
asyncio.create_task(self._refresh_networks())
self._current_connection_ssid = None
elif new_state in (NMDeviceState.DISCONNECTED, NMDeviceState.NEED_AUTH):
for network in self.networks:
network.is_connected = False
if new_state == NMDeviceState.NEED_AUTH and reason == NM_DEVICE_STATE_REASON_SUPPLICANT_DISCONNECT and self.callbacks.need_auth:
if self._current_connection_ssid:
self.callbacks.need_auth(self._current_connection_ssid)
else:
# Try to find the network from active_ap_path
for network in self.networks:
if network.path == self.active_ap_path:
self.callbacks.need_auth(network.ssid)
break
else:
# Couldn't identify the network that needs auth
cloudlog.error("Network needs authentication but couldn't identify which one")
def _on_new_connection(self, path: str) -> None:
"""Callback for NewConnection signal."""
asyncio.create_task(self._add_saved_connection(path))
def _on_connection_removed(self, path: str) -> None:
"""Callback for ConnectionRemoved signal."""
for ssid, p in list(self.saved_connections.items()):
if path == p:
del self.saved_connections[ssid]
if self.callbacks.forgotten:
self.callbacks.forgotten(ssid)
# Update network list to reflect the removed saved connection
asyncio.create_task(self._refresh_networks())
break
async def _add_saved_connection(self, path: str) -> None:
"""Add a new saved connection to the dictionary."""
try:
settings = await self._get_connection_settings(path)
if ssid := self._extract_ssid(settings):
self.saved_connections[ssid] = path
await self._refresh_networks()
except DBusError as e:
cloudlog.error(f"Failed to add connection {path}: {e}")
def _extract_ssid(self, settings: dict) -> str | None:
"""Extract SSID from connection settings."""
ssid_variant = settings.get('802-11-wireless', {}).get('ssid', Variant('ay', b'')).value
return ''.join(chr(b) for b in ssid_variant) if ssid_variant else None
async def _add_match_rule(self, rule):
"""Add a match rule on the bus."""
reply = await self.bus.call(
Message(
message_type=MessageType.METHOD_CALL,
destination='org.freedesktop.DBus',
interface="org.freedesktop.DBus",
path='/org/freedesktop/DBus',
member='AddMatch',
signature='s',
body=[rule],
)
)
assert reply.message_type == MessageType.METHOD_RETURN
return reply
async def _refresh_networks(self):
"""Get a list of available networks via NetworkManager."""
wifi_iface = self.device_proxy.get_interface(NM_WIRELESS_IFACE)
access_points = await wifi_iface.get_access_points()
self.active_ap_path = await self.get_active_access_point()
network_dict = {}
for ap_path in access_points:
try:
props_iface = await self._get_interface(NM, ap_path, NM_PROPERTIES_IFACE)
properties = await props_iface.call_get_all('org.freedesktop.NetworkManager.AccessPoint')
ssid_variant = properties['Ssid'].value
ssid = ''.join(chr(byte) for byte in ssid_variant)
if not ssid:
continue
bssid = properties.get('HwAddress', Variant('s', '')).value
strength = properties['Strength'].value
flags = properties['Flags'].value
wpa_flags = properties['WpaFlags'].value
rsn_flags = properties['RsnFlags'].value
existing_network = network_dict.get(ssid)
if not existing_network or ((not existing_network.bssid and bssid) or (existing_network.strength < strength)):
network_dict[ssid] = NetworkInfo(
ssid=ssid,
strength=strength,
security_type=self._get_security_type(flags, wpa_flags, rsn_flags),
path=ap_path,
bssid=bssid,
is_connected=self.active_ap_path == ap_path and self._current_connection_ssid != ssid,
is_saved=ssid in self.saved_connections
)
except DBusError as e:
cloudlog.error(f"Error fetching networks: {e}")
except Exception as e:
cloudlog.error({e})
self.networks = sorted(
network_dict.values(),
key=lambda network: (
not network.is_connected,
-network.strength, # Higher signal strength first
network.ssid.lower(),
),
)
if self.callbacks.networks_updated:
self.callbacks.networks_updated(copy.deepcopy(self.networks))
async def _get_connection_settings(self, path):
"""Fetch connection settings for a specific connection path."""
try:
settings = await self._get_interface(NM, path, NM_CONNECTION_IFACE)
return await settings.call_get_settings()
except DBusError as e:
cloudlog.error(f"Failed to get settings for {path}: {str(e)}")
return {}
async def _process_chunk(self, paths_chunk):
"""Process a chunk of connection paths."""
tasks = [self._get_connection_settings(path) for path in paths_chunk]
return await asyncio.gather(*tasks, return_exceptions=True)
async def _get_saved_connections(self) -> dict[str, str]:
try:
settings_iface = await self._get_interface(NM, NM_SETTINGS_PATH, NM_SETTINGS_IFACE)
connection_paths = await settings_iface.call_list_connections()
saved_ssids: dict[str, str] = {}
batch_size = 20
for i in range(0, len(connection_paths), batch_size):
chunk = connection_paths[i : i + batch_size]
results = await self._process_chunk(chunk)
for path, config in zip(chunk, results, strict=True):
if isinstance(config, dict) and '802-11-wireless' in config:
if ssid := self._extract_ssid(config):
saved_ssids[ssid] = path
return saved_ssids
except DBusError as e:
cloudlog.error(f"Error fetching saved connections: {str(e)}")
return {}
async def _get_interface(self, bus_name: str, path: str, name: str):
introspection = await self.bus.introspect(bus_name, path)
proxy = self.bus.get_proxy_object(bus_name, path, introspection)
return proxy.get_interface(name)
def _get_security_type(self, flags: int, wpa_flags: int, rsn_flags: int) -> SecurityType:
"""Determine the security type based on flags."""
if flags == 0 and not (wpa_flags or rsn_flags):
return SecurityType.OPEN
if rsn_flags & 0x200: # SAE (WPA3 Personal)
# TODO: support WPA3
return SecurityType.UNSUPPORTED
if rsn_flags: # RSN indicates WPA2 or higher
return SecurityType.WPA2
if wpa_flags: # WPA flags indicate WPA
return SecurityType.WPA
return SecurityType.UNSUPPORTED
class WifiManagerWrapper:
def __init__(self):
self._manager: WifiManager | None = None
self._callbacks: WifiManagerCallbacks = WifiManagerCallbacks()
self._thread = threading.Thread(target=self._run, daemon=True)
self._loop: asyncio.EventLoop | None = None
self._running = False
def set_callbacks(self, callbacks: WifiManagerCallbacks):
self._callbacks = callbacks
def start(self) -> None:
if not self._running:
self._thread.start()
while self._thread is not None and not self._running:
time.sleep(0.1)
def _run(self):
self._loop = asyncio.new_event_loop()
asyncio.set_event_loop(self._loop)
try:
self._manager = WifiManager(self._callbacks)
self._running = True
self._loop.run_forever()
except Exception as e:
cloudlog.error(f"Error in WifiManagerWrapper thread: {e}")
finally:
if self._loop.is_running():
self._loop.stop()
self._running = False
def shutdown(self) -> None:
if self._running:
if self._manager is not None and self._loop:
shutdown_future = asyncio.run_coroutine_threadsafe(self._manager.shutdown(), self._loop)
shutdown_future.result(timeout=3.0)
if self._loop and self._loop.is_running():
self._loop.call_soon_threadsafe(self._loop.stop)
if self._thread and self._thread.is_alive():
self._thread.join(timeout=2.0)
self._running = False
def is_saved(self, ssid: str) -> bool:
"""Check if a network is saved."""
return self._run_coroutine_sync(lambda manager: manager.is_saved(ssid), default=False)
def connect(self):
"""Connect to DBus and start Wi-Fi scanning."""
if not self._manager:
return
self._run_coroutine(self._manager.connect())
def forget_connection(self, ssid: str):
"""Forget a saved Wi-Fi connection."""
if not self._manager:
return
self._run_coroutine(self._manager.forget_connection(ssid))
def activate_connection(self, ssid: str):
"""Activate an existing Wi-Fi connection."""
if not self._manager:
return
self._run_coroutine(self._manager.activate_connection(ssid))
def connect_to_network(self, ssid: str, password: str = None, bssid: str = None, is_hidden: bool = False):
"""Connect to a Wi-Fi network."""
if not self._manager:
return
self._run_coroutine(self._manager.connect_to_network(ssid, password, bssid, is_hidden))
def _run_coroutine(self, coro):
"""Run a coroutine in the async thread."""
if not self._running or not self._loop:
cloudlog.error("WifiManager thread is not running")
return
asyncio.run_coroutine_threadsafe(coro, self._loop)
def _run_coroutine_sync(self, func: Callable[[WifiManager], T], default: T) -> T:
"""Run a function synchronously in the async thread."""
if not self._running or not self._loop or not self._manager:
return default
future = concurrent.futures.Future[T]()
def wrapper(manager: WifiManager) -> None:
try:
future.set_result(func(manager))
except Exception as e:
future.set_exception(e)
try:
self._loop.call_soon_threadsafe(wrapper, self._manager)
return future.result(timeout=1.0)
except Exception as e:
cloudlog.error(f"WifiManagerWrapper property access failed: {e}")
return default
+58
View File
@@ -0,0 +1,58 @@
import threading
import time
import os
from typing import Generic, Protocol, TypeVar
from openpilot.common.swaglog import cloudlog
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()
def _run(self):
if os.getenv("CI") is not None:
return
gui_app.init_window(self._title)
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():
cloudlog.warning(f"Failed to join {self._title} thread")
def __del__(self):
self.close()
def __exit__(self, exc_type, exc_val, exc_tb):
self.close()
+87
View File
@@ -0,0 +1,87 @@
import pyray as rl
from openpilot.system.ui.lib.text_measure import measure_text_cached
def _break_long_word(font: rl.Font, word: str, font_size: int, max_width: int) -> list[str]:
if not word:
return []
parts = []
remaining = word
while remaining:
if measure_text_cached(font, remaining, font_size).x <= max_width:
parts.append(remaining)
break
# Binary search for the longest substring that fits
left, right = 1, len(remaining)
best_fit = 1
while left <= right:
mid = (left + right) // 2
substring = remaining[:mid]
width = measure_text_cached(font, substring, font_size).x
if width <= max_width:
best_fit = mid
left = mid + 1
else:
right = mid - 1
# Add the part that fits
parts.append(remaining[:best_fit])
remaining = remaining[best_fit:]
return parts
def wrap_text(font: rl.Font, text: str, font_size: int, max_width: int) -> list[str]:
if not text or max_width <= 0:
return []
words = text.split()
if not words:
return []
lines: list[str] = []
current_line: list[str] = []
current_width = 0
space_width = int(measure_text_cached(font, " ", font_size).x)
for word in words:
word_width = int(measure_text_cached(font, word, font_size).x)
# Check if word alone exceeds max width (need to break the word)
if word_width > max_width:
# Finish current line if it has content
if current_line:
lines.append(" ".join(current_line))
current_line = []
current_width = 0
# Break the long word into parts
lines.extend(_break_long_word(font, word, font_size, max_width))
continue
# Calculate width if we add this word
needed_width = current_width
if current_line: # Need space before word
needed_width += space_width
needed_width += word_width
# Check if word fits on current line
if needed_width <= max_width:
current_line.append(word)
current_width = needed_width
else:
# Start new line with this word
if current_line:
lines.append(" ".join(current_line))
current_line = [word]
current_width = word_width
# Add remaining words
if current_line:
lines.append(" ".join(current_line))
return lines
+6 -2
View File
@@ -5,6 +5,7 @@ import sys
import threading
from enum import IntEnum
from openpilot.system.hardware import PC
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_label, gui_text_box
@@ -31,7 +32,10 @@ class Reset:
self.mode = mode
self.reset_state = ResetState.NONE
def do_reset(self):
def _do_erase(self):
if PC:
return
# Best effort to wipe NVME
os.system(f"sudo umount {NVME}")
os.system(f"yes | sudo mkfs.ext4 {NVME}")
@@ -48,7 +52,7 @@ class Reset:
def start_reset(self):
self.reset_state = ResetState.RESETTING
threading.Timer(0.1, self.do_reset).start()
threading.Timer(0.1, self._do_erase).start()
def render(self, rect: rl.Rectangle):
label_rect = rl.Rectangle(rect.x + 140, rect.y, rect.width - 280, 100)
+349
View File
@@ -0,0 +1,349 @@
#!/usr/bin/env python3
import os
import re
import threading
import time
import urllib.request
from enum import IntEnum
import pyray as rl
from cereal import log
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_label, gui_text_box
from openpilot.system.ui.widgets.network import WifiManagerUI, WifiManagerWrapper
from openpilot.system.ui.widgets.keyboard import Keyboard
NetworkType = log.DeviceState.NetworkType
MARGIN = 50
TITLE_FONT_SIZE = 116
TITLE_FONT_WEIGHT = FontWeight.MEDIUM
NEXT_BUTTON_WIDTH = 310
BODY_FONT_SIZE = 96
BUTTON_HEIGHT = 160
BUTTON_SPACING = 50
OPENPILOT_URL = "https://openpilot.comma.ai"
USER_AGENT = f"AGNOSSetup-{HARDWARE.get_os_version()}"
class SetupState(IntEnum):
LOW_VOLTAGE = 0
GETTING_STARTED = 1
NETWORK_SETUP = 2
SOFTWARE_SELECTION = 3
CUSTOM_URL = 4
DOWNLOADING = 5
DOWNLOAD_FAILED = 6
class Setup:
def __init__(self):
self.state = SetupState.GETTING_STARTED
self.network_check_thread = None
self.network_connected = threading.Event()
self.wifi_connected = threading.Event()
self.stop_network_check_thread = threading.Event()
self.failed_url = ""
self.failed_reason = ""
self.download_url = ""
self.download_progress = 0
self.download_thread = None
self.wifi_manager = WifiManagerWrapper()
self.wifi_ui = WifiManagerUI(self.wifi_manager)
self.keyboard = Keyboard()
self.selected_radio = None
self.warning = gui_app.texture("icons/warning.png", 150, 150)
self.checkmark = gui_app.texture("icons/circled_check.png", 100, 100)
try:
with open("/sys/class/hwmon/hwmon1/in1_input") as f:
voltage = float(f.read().strip()) / 1000.0
if voltage < 7:
self.state = SetupState.LOW_VOLTAGE
except (FileNotFoundError, ValueError):
self.state = SetupState.LOW_VOLTAGE
def render(self, rect: rl.Rectangle):
if self.state == SetupState.LOW_VOLTAGE:
self.render_low_voltage(rect)
elif self.state == SetupState.GETTING_STARTED:
self.render_getting_started(rect)
elif self.state == SetupState.NETWORK_SETUP:
self.render_network_setup(rect)
elif self.state == SetupState.SOFTWARE_SELECTION:
self.render_software_selection(rect)
elif self.state == SetupState.CUSTOM_URL:
self.render_custom_url()
elif self.state == SetupState.DOWNLOADING:
self.render_downloading(rect)
elif self.state == SetupState.DOWNLOAD_FAILED:
self.render_download_failed(rect)
def render_low_voltage(self, rect: rl.Rectangle):
rl.draw_texture(self.warning, int(rect.x + 150), int(rect.y + 110), rl.WHITE)
title_rect = rl.Rectangle(rect.x + 150, rect.y + 110 + 150 + 100, rect.width - 500 - 150, TITLE_FONT_SIZE)
gui_label(title_rect, "WARNING: Low Voltage", TITLE_FONT_SIZE, rl.Color(255, 89, 79, 255), FontWeight.MEDIUM)
body_rect = rl.Rectangle(rect.x + 150, rect.y + 110 + 150 + 100 + TITLE_FONT_SIZE + 25, rect.width - 500 - 150, BODY_FONT_SIZE * 3)
gui_text_box(body_rect, "Power your device in a car with a harness or proceed at your own risk.", BODY_FONT_SIZE)
button_width = (rect.width - MARGIN * 3) / 2
button_y = rect.height - MARGIN - BUTTON_HEIGHT
if gui_button(rl.Rectangle(rect.x + MARGIN, button_y, button_width, BUTTON_HEIGHT), "Power off"):
HARDWARE.shutdown()
if gui_button(rl.Rectangle(rect.x + MARGIN * 2 + button_width, button_y, button_width, BUTTON_HEIGHT), "Continue"):
self.state = SetupState.GETTING_STARTED
def render_getting_started(self, rect: rl.Rectangle):
title_rect = rl.Rectangle(rect.x + 165, rect.y + 280, rect.width - 265, TITLE_FONT_SIZE)
gui_label(title_rect, "Getting Started", TITLE_FONT_SIZE, font_weight=FontWeight.MEDIUM)
desc_rect = rl.Rectangle(rect.x + 165, rect.y + 280 + TITLE_FONT_SIZE + 90, rect.width - 500, BODY_FONT_SIZE * 3)
gui_text_box(desc_rect, "Before we get on the road, let's finish installation and cover some details.", BODY_FONT_SIZE)
btn_rect = rl.Rectangle(rect.width - NEXT_BUTTON_WIDTH, 0, NEXT_BUTTON_WIDTH, rect.height)
ret = gui_button(btn_rect, "", button_style=ButtonStyle.PRIMARY, border_radius=0)
triangle = gui_app.texture("images/button_continue_triangle.png", 54, int(btn_rect.height))
rl.draw_texture_v(triangle, rl.Vector2(btn_rect.x + btn_rect.width / 2 - triangle.width / 2, btn_rect.height / 2 - triangle.height / 2), rl.WHITE)
if ret:
self.state = SetupState.NETWORK_SETUP
self.wifi_manager.request_scan()
self.start_network_check()
def check_network_connectivity(self):
while not self.stop_network_check_thread.is_set():
if self.state == SetupState.NETWORK_SETUP:
try:
urllib.request.urlopen(OPENPILOT_URL, timeout=2)
self.network_connected.set()
if HARDWARE.get_network_type() == NetworkType.wifi:
self.wifi_connected.set()
else:
self.wifi_connected.clear()
except Exception:
self.network_connected.clear()
time.sleep(1)
def start_network_check(self):
if self.network_check_thread is None or not self.network_check_thread.is_alive():
self.network_check_thread = threading.Thread(target=self.check_network_connectivity, daemon=True)
self.network_check_thread.start()
def close(self):
if self.network_check_thread is not None:
self.stop_network_check_thread.set()
self.network_check_thread.join()
def render_network_setup(self, rect: rl.Rectangle):
if self.wifi_ui.require_full_screen:
self.wifi_ui.render(rect)
return
title_rect = rl.Rectangle(rect.x + MARGIN, rect.y + MARGIN, rect.width - MARGIN * 2, TITLE_FONT_SIZE)
gui_label(title_rect, "Connect to Wi-Fi", TITLE_FONT_SIZE, font_weight=FontWeight.MEDIUM)
wifi_rect = rl.Rectangle(rect.x + MARGIN, rect.y + TITLE_FONT_SIZE + MARGIN + 25, rect.width - MARGIN * 2,
rect.height - TITLE_FONT_SIZE - 25 - BUTTON_HEIGHT - MARGIN * 3)
rl.draw_rectangle_rounded(wifi_rect, 0.05, 10, rl.Color(51, 51, 51, 255))
wifi_content_rect = rl.Rectangle(wifi_rect.x + MARGIN, wifi_rect.y, wifi_rect.width - MARGIN * 2, wifi_rect.height)
self.wifi_ui.render(wifi_content_rect)
button_width = (rect.width - BUTTON_SPACING - MARGIN * 2) / 2
button_y = rect.height - BUTTON_HEIGHT - MARGIN
if gui_button(rl.Rectangle(rect.x + MARGIN, button_y, button_width, BUTTON_HEIGHT), "Back"):
self.state = SetupState.GETTING_STARTED
# Check network connectivity status
continue_enabled = self.network_connected.is_set()
continue_text = ("Continue" if self.wifi_connected.is_set() else "Continue without Wi-Fi") if continue_enabled else "Waiting for internet"
if gui_button(
rl.Rectangle(rect.x + MARGIN + button_width + BUTTON_SPACING, button_y, button_width, BUTTON_HEIGHT),
continue_text,
button_style=ButtonStyle.PRIMARY if continue_enabled else ButtonStyle.NORMAL,
is_enabled=continue_enabled,
):
self.state = SetupState.SOFTWARE_SELECTION
self.stop_network_check_thread.set()
def render_software_selection(self, rect: rl.Rectangle):
title_rect = rl.Rectangle(rect.x + MARGIN, rect.y + MARGIN, rect.width - MARGIN * 2, TITLE_FONT_SIZE)
gui_label(title_rect, "Choose Software to Install", TITLE_FONT_SIZE, font_weight=FontWeight.MEDIUM)
radio_height = 230
radio_spacing = 30
openpilot_rect = rl.Rectangle(rect.x + MARGIN, rect.y + TITLE_FONT_SIZE + MARGIN * 2, rect.width - MARGIN * 2, radio_height)
openpilot_selected = self.selected_radio == "openpilot"
rl.draw_rectangle_rounded(openpilot_rect, 0.1, 10, rl.Color(70, 91, 234, 255) if openpilot_selected else rl.Color(79, 79, 79, 255))
gui_label(rl.Rectangle(openpilot_rect.x + 100, openpilot_rect.y, openpilot_rect.width - 200, radio_height), "openpilot", BODY_FONT_SIZE)
if openpilot_selected:
checkmark_pos = rl.Vector2(openpilot_rect.x + openpilot_rect.width - 100 - self.checkmark.width,
openpilot_rect.y + radio_height / 2 - self.checkmark.height / 2)
rl.draw_texture_v(self.checkmark, checkmark_pos, rl.WHITE)
custom_rect = rl.Rectangle(rect.x + MARGIN, rect.y + TITLE_FONT_SIZE + MARGIN * 2 + radio_height + radio_spacing, rect.width - MARGIN * 2, radio_height)
custom_selected = self.selected_radio == "custom"
rl.draw_rectangle_rounded(custom_rect, 0.1, 10, rl.Color(70, 91, 234, 255) if custom_selected else rl.Color(79, 79, 79, 255))
gui_label(rl.Rectangle(custom_rect.x + 100, custom_rect.y, custom_rect.width - 200, radio_height), "Custom Software", BODY_FONT_SIZE)
if custom_selected:
checkmark_pos = rl.Vector2(custom_rect.x + custom_rect.width - 100 - self.checkmark.width, custom_rect.y + radio_height / 2 - self.checkmark.height / 2)
rl.draw_texture_v(self.checkmark, checkmark_pos, rl.WHITE)
mouse_pos = rl.get_mouse_position()
if rl.is_mouse_button_released(rl.MouseButton.MOUSE_BUTTON_LEFT):
if rl.check_collision_point_rec(mouse_pos, openpilot_rect):
self.selected_radio = "openpilot"
elif rl.check_collision_point_rec(mouse_pos, custom_rect):
self.selected_radio = "custom"
button_width = (rect.width - BUTTON_SPACING - MARGIN * 2) / 2
button_y = rect.height - BUTTON_HEIGHT - MARGIN
if gui_button(rl.Rectangle(rect.x + MARGIN, button_y, button_width, BUTTON_HEIGHT), "Back"):
self.state = SetupState.NETWORK_SETUP
continue_enabled = self.selected_radio is not None
if gui_button(
rl.Rectangle(rect.x + MARGIN + button_width + BUTTON_SPACING, button_y, button_width, BUTTON_HEIGHT),
"Continue",
button_style=ButtonStyle.PRIMARY,
is_enabled=continue_enabled,
):
if continue_enabled:
if self.selected_radio == "openpilot":
self.download(OPENPILOT_URL)
else:
self.state = SetupState.CUSTOM_URL
def render_downloading(self, rect: rl.Rectangle):
title_rect = rl.Rectangle(rect.x, rect.y + rect.height / 2 - TITLE_FONT_SIZE / 2, rect.width, TITLE_FONT_SIZE)
gui_label(title_rect, "Downloading...", TITLE_FONT_SIZE, font_weight=FontWeight.MEDIUM, alignment=rl.GuiTextAlignment.TEXT_ALIGN_CENTER)
def render_download_failed(self, rect: rl.Rectangle):
title_rect = rl.Rectangle(rect.x + 117, rect.y + 185, rect.width - 117, TITLE_FONT_SIZE)
gui_label(title_rect, "Download Failed", TITLE_FONT_SIZE, font_weight=FontWeight.MEDIUM)
url_rect = rl.Rectangle(rect.x + 117, rect.y + 185 + TITLE_FONT_SIZE + 67, rect.width - 117 - 100, 64)
gui_label(url_rect, self.failed_url, 64, font_weight=FontWeight.NORMAL)
error_rect = rl.Rectangle(rect.x + 117, rect.y + 185 + TITLE_FONT_SIZE + 67 + 64 + 48,
rect.width - 117 - 100, rect.height - 185 + TITLE_FONT_SIZE + 67 + 64 + 48 - BUTTON_HEIGHT - MARGIN * 2)
gui_text_box(error_rect, self.failed_reason, BODY_FONT_SIZE)
button_width = (rect.width - BUTTON_SPACING - MARGIN * 2) / 2
button_y = rect.height - BUTTON_HEIGHT - MARGIN
if gui_button(rl.Rectangle(rect.x + MARGIN, button_y, button_width, BUTTON_HEIGHT), "Reboot device"):
HARDWARE.reboot()
if gui_button(rl.Rectangle(rect.x + MARGIN + button_width + BUTTON_SPACING, button_y, button_width, BUTTON_HEIGHT), "Start over",
button_style=ButtonStyle.PRIMARY):
self.state = SetupState.GETTING_STARTED
def render_custom_url(self):
result = self.keyboard.render("Enter URL", "for Custom Software")
# Enter pressed
if result == 1:
url = self.keyboard.text
self.keyboard.clear()
if url:
self.download(url)
# Cancel pressed
elif result == 0:
self.state = SetupState.SOFTWARE_SELECTION
def download(self, url: str):
# autocomplete incomplete URLs
if re.match("^([^/.]+)/([^/]+)$", url):
url = f"https://installer.comma.ai/{url}"
self.download_url = url
self.state = SetupState.DOWNLOADING
self.download_thread = threading.Thread(target=self._download_thread, daemon=True)
self.download_thread.start()
def _download_thread(self):
try:
import tempfile
_, tmpfile = tempfile.mkstemp(prefix="installer_")
headers = {"User-Agent": USER_AGENT, "X-openpilot-serial": HARDWARE.get_serial()}
req = urllib.request.Request(self.download_url, headers=headers)
with open(tmpfile, 'wb') as f, urllib.request.urlopen(req, timeout=30) as response:
total_size = int(response.headers.get('content-length', 0))
downloaded = 0
block_size = 8192
while True:
buffer = response.read(block_size)
if not buffer:
break
downloaded += len(buffer)
f.write(buffer)
if total_size:
self.download_progress = int(downloaded * 100 / total_size)
is_elf = False
with open(tmpfile, 'rb') as f:
header = f.read(4)
is_elf = header == b'\x7fELF'
if not is_elf:
self.download_failed(self.download_url, "No custom software found at this URL.")
return
os.rename(tmpfile, "/tmp/installer")
os.chmod("/tmp/installer", 0o755)
with open("/tmp/installer_url", "w") as f:
f.write(self.download_url)
gui_app.request_close()
except Exception:
error_msg = "Ensure the entered URL is valid, and the device's internet connection is good."
self.download_failed(self.download_url, error_msg)
def download_failed(self, url: str, reason: str):
self.failed_url = url
self.failed_reason = reason
self.state = SetupState.DOWNLOAD_FAILED
def main():
try:
gui_app.init_window("Setup")
setup = Setup()
for _ in gui_app.render():
setup.render(rl.Rectangle(0, 0, gui_app.width, gui_app.height))
setup.close()
except Exception as e:
print(f"Setup error: {e}")
finally:
gui_app.close()
if __name__ == "__main__":
main()
+86 -54
View File
@@ -1,75 +1,107 @@
#!/usr/bin/env python3
import pyray as rl
import os
import select
import sys
import threading
import time
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
PROGRESS_BAR_WIDTH = 1000
PROGRESS_BAR_HEIGHT = 20
ROTATION_TIME_SECONDS = 1.0 # Time for one full circle
MARGIN = 200
DEGREES_PER_SECOND = 360.0 # one full rotation per second
MARGIN_H = 100
TEXTURE_SIZE = 360
FONT_SIZE = 80
FONT_SIZE = 96
LINE_HEIGHT = 104
DARKGRAY = (55, 55, 55, 255)
def clamp(value, min_value, max_value):
return max(min(value, max_value), min_value)
def check_input_non_blocking():
if sys.stdin in select.select([sys.stdin], [], [], 0)[0]:
return sys.stdin.readline().strip()
return ""
class SpinnerRenderer:
def __init__(self):
self._comma_texture = gui_app.texture("images/spinner_comma.png", TEXTURE_SIZE, TEXTURE_SIZE)
self._spinner_texture = gui_app.texture("images/spinner_track.png", TEXTURE_SIZE, TEXTURE_SIZE, alpha_premultiply=True)
self._rotation = 0.0
self._progress: int | None = None
self._wrapped_lines: list[str] = []
self._lock = threading.Lock()
def set_text(self, text: str) -> None:
with self._lock:
if text.isdigit():
self._progress = clamp(int(text), 0, 100)
self._wrapped_lines = []
else:
self._progress = None
self._wrapped_lines = wrap_text(text, FONT_SIZE, gui_app.width - MARGIN_H)
def render(self):
with self._lock:
progress = self._progress
wrapped_lines = self._wrapped_lines
if wrapped_lines:
# Calculate total height required for spinner and text
spacing = 50
total_height = TEXTURE_SIZE + spacing + len(wrapped_lines) * LINE_HEIGHT
center_y = (gui_app.height - total_height) / 2.0 + TEXTURE_SIZE / 2.0
else:
# Center spinner vertically
spacing = 150
center_y = gui_app.height / 2.0
y_pos = center_y + TEXTURE_SIZE / 2.0 + spacing
center = rl.Vector2(gui_app.width / 2.0, center_y)
spinner_origin = rl.Vector2(TEXTURE_SIZE / 2.0, TEXTURE_SIZE / 2.0)
comma_position = rl.Vector2(center.x - TEXTURE_SIZE / 2.0, center.y - TEXTURE_SIZE / 2.0)
delta_time = rl.get_frame_time()
self._rotation = (self._rotation + DEGREES_PER_SECOND * delta_time) % 360.0
# Draw rotating spinner and static comma logo
rl.draw_texture_pro(self._spinner_texture, rl.Rectangle(0, 0, TEXTURE_SIZE, TEXTURE_SIZE),
rl.Rectangle(center.x, center.y, TEXTURE_SIZE, TEXTURE_SIZE),
spinner_origin, self._rotation, rl.WHITE)
rl.draw_texture_v(self._comma_texture, comma_position, rl.WHITE)
# Display the progress bar or text based on user input
if progress is not None:
bar = rl.Rectangle(center.x - PROGRESS_BAR_WIDTH / 2.0, y_pos, PROGRESS_BAR_WIDTH, PROGRESS_BAR_HEIGHT)
rl.draw_rectangle_rounded(bar, 1, 10, DARKGRAY)
bar.width *= progress / 100.0
rl.draw_rectangle_rounded(bar, 1, 10, rl.WHITE)
elif wrapped_lines:
for i, line in enumerate(wrapped_lines):
text_size = rl.measure_text_ex(gui_app.font(), line, FONT_SIZE, 0.0)
rl.draw_text_ex(gui_app.font(), line, rl.Vector2(center.x - text_size.x / 2, y_pos + i * LINE_HEIGHT),
FONT_SIZE, 0.0, rl.WHITE)
class Spinner(BaseWindow[SpinnerRenderer]):
def __init__(self):
super().__init__("Spinner")
def _create_renderer(self):
return SpinnerRenderer()
def update(self, spinner_text: str):
if self._renderer is not None:
self._renderer.set_text(spinner_text)
def update_progress(self, cur: float, total: float):
self.update(str(round(100 * cur / total)))
def main():
gui_app.init_window("Spinner")
# Load textures
comma_texture = gui_app.load_texture_from_image(os.path.join(BASEDIR, "selfdrive/assets/img_spinner_comma.png"), TEXTURE_SIZE, TEXTURE_SIZE)
spinner_texture = gui_app.load_texture_from_image(os.path.join(BASEDIR, "selfdrive/assets/img_spinner_track.png"), TEXTURE_SIZE, TEXTURE_SIZE)
# Initial values
rotation = 0.0
user_input = ""
center = rl.Vector2(gui_app.width / 2.0, gui_app.height / 2.0)
spinner_origin = rl.Vector2(TEXTURE_SIZE / 2.0, TEXTURE_SIZE / 2.0)
comma_position = rl.Vector2(center.x - TEXTURE_SIZE / 2.0, center.y - TEXTURE_SIZE / 2.0)
for _ in gui_app.render():
fps = rl.get_fps()
if fps > 0:
degrees_per_frame = 360.0 / (ROTATION_TIME_SECONDS * fps)
rotation = (rotation + degrees_per_frame) % 360.0
# Draw rotating spinner and static comma logo
rl.draw_texture_pro(spinner_texture, rl.Rectangle(0, 0, TEXTURE_SIZE, TEXTURE_SIZE),
rl.Rectangle(center.x, center.y, TEXTURE_SIZE, TEXTURE_SIZE),
spinner_origin, rotation, rl.WHITE)
rl.draw_texture_v(comma_texture, comma_position, rl.WHITE)
# Read user input
if input_str := check_input_non_blocking():
user_input = input_str
# Display progress bar or text based on user input
if user_input:
y_pos = rl.get_screen_height() - MARGIN - PROGRESS_BAR_HEIGHT
if user_input.isdigit():
progress = clamp(int(user_input), 0, 100)
bar = rl.Rectangle(center.x - PROGRESS_BAR_WIDTH / 2.0, y_pos, PROGRESS_BAR_WIDTH, PROGRESS_BAR_HEIGHT)
rl.draw_rectangle_rounded(bar, 0.5, 10, rl.GRAY)
bar.width *= progress / 100.0
rl.draw_rectangle_rounded(bar, 0.5, 10, rl.WHITE)
else:
text_size = rl.measure_text_ex(gui_app.font(), user_input, FONT_SIZE, 1.0)
rl.draw_text_ex(gui_app.font(), user_input,
rl.Vector2(center.x - text_size.x / 2, y_pos), FONT_SIZE, 1.0, rl.WHITE)
with Spinner() as s:
s.update("Spinner text")
time.sleep(5)
if __name__ == "__main__":
+63 -36
View File
@@ -1,16 +1,17 @@
#!/usr/bin/env python3
import sys
import re
import time
import pyray as rl
from openpilot.system.hardware import HARDWARE
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 = 50
FONT_SIZE = 60
LINE_HEIGHT = 64
SPACING = 40
FONT_SIZE = 72
LINE_HEIGHT = 80
BUTTON_SIZE = rl.Vector2(310, 160)
DEMO_TEXT = """This is a sample text that will be wrapped and scrolled if necessary.
@@ -18,47 +19,73 @@ DEMO_TEXT = """This is a sample text that will be wrapped and scrolled if necess
def wrap_text(text, font_size, max_width):
lines = []
current_line = ""
font = gui_app.font()
for word in text.split():
test_line = current_line + word + " "
if rl.measure_text_ex(font, test_line, font_size, 0).x <= max_width:
current_line = test_line
else:
for paragraph in text.split("\n"):
if not paragraph.strip():
# Don't add empty lines first, ensuring wrap_text("") returns []
if lines:
lines.append("")
continue
indent = re.match(r"^\s*", paragraph).group()
current_line = indent
words = re.split(r"(\s+)", paragraph[len(indent):])
while len(words):
word = words.pop(0)
test_line = current_line + word + (words.pop(0) if words else "")
if rl.measure_text_ex(font, test_line, font_size, 0).x <= max_width:
current_line = test_line
else:
lines.append(current_line)
current_line = word + " "
current_line = current_line.rstrip()
if current_line:
lines.append(current_line)
current_line = word + " "
if current_line:
lines.append(current_line)
return lines
def main():
gui_app.init_window("Text")
class TextWindowRenderer:
def __init__(self, text: str):
self._textarea_rect = rl.Rectangle(MARGIN, MARGIN, gui_app.width - MARGIN * 2, gui_app.height - MARGIN * 2)
self._wrapped_lines = wrap_text(text, FONT_SIZE, self._textarea_rect.width - 20)
self._content_rect = rl.Rectangle(0, 0, self._textarea_rect.width - 20, len(self._wrapped_lines) * LINE_HEIGHT)
self._scroll_panel = GuiScrollPanel(show_vertical_scroll_bar=True)
self._scroll_panel._offset.y = -max(self._content_rect.height - self._textarea_rect.height, 0)
text_content = sys.argv[1] if len(sys.argv) > 1 else DEMO_TEXT
textarea_rect = rl.Rectangle(MARGIN, MARGIN, gui_app.width - MARGIN * 2, gui_app.height - MARGIN * 2)
wrapped_lines = wrap_text(text_content, FONT_SIZE, textarea_rect.width - 20)
content_rect = rl.Rectangle(0, 0, textarea_rect.width - 20, len(wrapped_lines) * LINE_HEIGHT)
scroll_panel = GuiScrollPanel(show_vertical_scroll_bar=True)
for _ in gui_app.render():
scroll = scroll_panel.handle_scroll(textarea_rect, content_rect)
rl.begin_scissor_mode(int(textarea_rect.x), int(textarea_rect.y), int(textarea_rect.width), int(textarea_rect.height))
for i, line in enumerate(wrapped_lines):
position = rl.Vector2(textarea_rect.x + scroll.x, textarea_rect.y + scroll.y + i * LINE_HEIGHT)
if position.y + LINE_HEIGHT < textarea_rect.y or position.y > textarea_rect.y + textarea_rect.height:
def render(self):
scroll = self._scroll_panel.handle_scroll(self._textarea_rect, self._content_rect)
rl.begin_scissor_mode(int(self._textarea_rect.x), int(self._textarea_rect.y), int(self._textarea_rect.width), int(self._textarea_rect.height))
for i, line in enumerate(self._wrapped_lines):
position = rl.Vector2(self._textarea_rect.x + scroll.x, self._textarea_rect.y + scroll.y + i * LINE_HEIGHT)
if position.y + LINE_HEIGHT < self._textarea_rect.y or position.y > self._textarea_rect.y + self._textarea_rect.height:
continue
rl.draw_text_ex(gui_app.font(), line.strip(), position, FONT_SIZE, 0, rl.WHITE)
rl.draw_text_ex(gui_app.font(), line, position, FONT_SIZE, 0, rl.WHITE)
rl.end_scissor_mode()
button_bounds = rl.Rectangle(gui_app.width - MARGIN - BUTTON_SIZE.x, gui_app.height - MARGIN - BUTTON_SIZE.y, BUTTON_SIZE.x, BUTTON_SIZE.y)
if gui_button(button_bounds, "Reboot", button_style=ButtonStyle.TRANSPARENT):
HARDWARE.reboot()
button_bounds = rl.Rectangle(gui_app.width - MARGIN - BUTTON_SIZE.x - SPACING, gui_app.height - MARGIN - BUTTON_SIZE.y, BUTTON_SIZE.x, BUTTON_SIZE.y)
ret = gui_button(button_bounds, "Exit" if PC else "Reboot", button_style=ButtonStyle.TRANSPARENT)
if ret:
if PC:
gui_app.request_close()
else:
HARDWARE.reboot()
return ret
class TextWindow(BaseWindow[TextWindowRenderer]):
def __init__(self, text: str):
self._text = text
super().__init__("Text")
def _create_renderer(self):
return TextWindowRenderer(self._text)
def wait_for_exit(self):
while self._thread.is_alive():
time.sleep(0.01)
if __name__ == "__main__":
main()
with TextWindow(DEMO_TEXT):
time.sleep(30)
+171
View File
@@ -0,0 +1,171 @@
#!/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
from openpilot.system.ui.lib.wifi_manager import WifiManagerWrapper
from openpilot.system.ui.widgets.network import WifiManagerUI
# 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
class Updater:
def __init__(self, updater_path, manifest_path):
self.updater = updater_path
self.manifest = manifest_path
self.current_screen = Screen.PROMPT
self.progress_value = 0
self.progress_text = "Loading..."
self.show_reboot_button = False
self.process = None
self.update_thread = None
self.wifi_manager = WifiManagerWrapper()
self.wifi_manager_ui = WifiManagerUI(self.wifi_manager)
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):
# TODO: just import it and run in a thread without a subprocess
cmd = [self.updater, "--swap", self.manifest]
self.process = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE,
text=True, bufsize=1, universal_newlines=True)
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
exit_code = self.process.wait()
if exit_code == 0:
HARDWARE.reboot()
else:
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
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):
# Draw the Wi-Fi manager UI
wifi_rect = rl.Rectangle(MARGIN + 50, MARGIN, gui_app.width - MARGIN * 2 - 100, gui_app.height - MARGIN * 2 - BUTTON_HEIGHT - 20)
self.wifi_manager_ui.render(wifi_rect)
if self.wifi_manager_ui.require_full_screen:
return
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
return # Return to avoid processing other interactions after screen change
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):
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()
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():
updater.render()
finally:
# Make sure we clean up even if there's an error
gui_app.close()
if __name__ == "__main__":
main()
+6 -11
View File
@@ -1,4 +1,5 @@
import pyray as rl
from openpilot.system.ui.lib.application import gui_app
from openpilot.system.ui.lib.button import gui_button, ButtonStyle
from openpilot.system.ui.lib.label import gui_text_box
@@ -11,10 +12,9 @@ TEXT_AREA_HEIGHT_REDUCTION = 200
BACKGROUND_COLOR = rl.Color(27, 27, 27, 255)
def confirm_dialog(rect: rl.Rectangle, message: str, confirm_text: str, cancel_text: str = "Cancel") -> int:
# Calculate dialog position and size, centered within the parent rectangle
dialog_x = rect.x + (rect.width - DIALOG_WIDTH) / 2
dialog_y = rect.y + (rect.height - DIALOG_HEIGHT) / 2
def confirm_dialog(message: str, confirm_text: str, cancel_text: str = "Cancel") -> int:
dialog_x = (gui_app.width - DIALOG_WIDTH) / 2
dialog_y = (gui_app.height - DIALOG_HEIGHT) / 2
dialog_rect = rl.Rectangle(dialog_x, dialog_y, DIALOG_WIDTH, DIALOG_HEIGHT)
# Calculate button positions at the bottom of the dialog
@@ -27,19 +27,14 @@ def confirm_dialog(rect: rl.Rectangle, message: str, confirm_text: str, cancel_t
yes_button = rl.Rectangle(yes_button_x, button_y, button_width, BUTTON_HEIGHT)
# Draw the dialog background
rl.draw_rectangle(
int(dialog_rect.x),
int(dialog_rect.y),
int(dialog_rect.width),
int(dialog_rect.height),
BACKGROUND_COLOR,
)
rl.draw_rectangle_rec(dialog_rect, BACKGROUND_COLOR)
# Draw the message in the dialog, centered
text_rect = rl.Rectangle(dialog_rect.x, dialog_rect.y, dialog_rect.width, dialog_rect.height - TEXT_AREA_HEIGHT_REDUCTION)
gui_text_box(
text_rect,
message,
font_size=88,
alignment=rl.GuiTextAlignment.TEXT_ALIGN_CENTER,
alignment_vertical=rl.GuiTextAlignmentVertical.TEXT_ALIGN_MIDDLE,
)
+163 -34
View File
@@ -1,29 +1,40 @@
import time
from typing import Literal
import pyray as rl
from openpilot.system.ui.lib.button import gui_button
from openpilot.system.ui.lib.application import gui_app, FontWeight
from openpilot.system.ui.lib.button import ButtonStyle, gui_button
from openpilot.system.ui.lib.inputbox import InputBox
from openpilot.system.ui.lib.label import gui_label
KEY_FONT_SIZE = 96
DOUBLE_CLICK_THRESHOLD = 0.5 # seconds
DELETE_REPEAT_DELAY = 0.5
DELETE_REPEAT_INTERVAL = 0.07
# Constants for special keys
CONTENT_MARGIN = 50
BACKSPACE_KEY = "<-"
ENTER_KEY = "Enter"
ENTER_KEY = "->"
SPACE_KEY = " "
SHIFT_KEY = ""
SHIFT_DOWN_KEY = ""
SHIFT_INACTIVE_KEY = "SHIFT_OFF"
SHIFT_ACTIVE_KEY = "SHIFT_ON"
CAPS_LOCK_KEY = "CAPS"
NUMERIC_KEY = "123"
SYMBOL_KEY = "#+="
ABC_KEY = "ABC"
# Define keyboard layouts as a dictionary for easier access
keyboard_layouts = {
KEYBOARD_LAYOUTS = {
"lowercase": [
["q", "w", "e", "r", "t", "y", "u", "i", "o", "p"],
["a", "s", "d", "f", "g", "h", "j", "k", "l"],
[SHIFT_KEY, "z", "x", "c", "v", "b", "n", "m", BACKSPACE_KEY],
[SHIFT_INACTIVE_KEY, "z", "x", "c", "v", "b", "n", "m", BACKSPACE_KEY],
[NUMERIC_KEY, "/", "-", SPACE_KEY, ".", ENTER_KEY],
],
"uppercase": [
["Q", "W", "E", "R", "T", "Y", "U", "I", "O", "P"],
["A", "S", "D", "F", "G", "H", "J", "K", "L"],
[SHIFT_DOWN_KEY, "Z", "X", "C", "V", "B", "N", "M", BACKSPACE_KEY],
[SHIFT_ACTIVE_KEY, "Z", "X", "C", "V", "B", "N", "M", BACKSPACE_KEY],
[NUMERIC_KEY, "/", "-", SPACE_KEY, ".", ENTER_KEY],
],
"numbers": [
@@ -42,34 +53,79 @@ keyboard_layouts = {
class Keyboard:
def __init__(self, max_text_size: int = 255):
self._layout = keyboard_layouts["lowercase"]
self._input_text = ""
def __init__(self, max_text_size: int = 255, min_text_size: int = 0, password_mode: bool = False, show_password_toggle: bool = False):
self._layout_name: Literal["lowercase", "uppercase", "numbers", "specials"] = "lowercase"
self._caps_lock = False
self._last_shift_press_time = 0
self._max_text_size = max_text_size
self._min_text_size = min_text_size
self._input_box = InputBox(max_text_size)
self._password_mode = password_mode
self._show_password_toggle = show_password_toggle
# Backspace key repeat tracking
self._backspace_pressed: bool = False
self._backspace_press_time: float = 0.0
self._backspace_last_repeat:float = 0.0
self._eye_open_texture = gui_app.texture("icons/eye_open.png", 81, 54)
self._eye_closed_texture = gui_app.texture("icons/eye_closed.png", 81, 54)
self._key_icons = {
BACKSPACE_KEY: gui_app.texture("icons/backspace.png", 80, 80),
SHIFT_INACTIVE_KEY: gui_app.texture("icons/shift.png", 80, 80),
SHIFT_ACTIVE_KEY: gui_app.texture("icons/shift-fill.png", 80, 80),
CAPS_LOCK_KEY: gui_app.texture("icons/capslock-fill.png", 80, 80),
ENTER_KEY: gui_app.texture("icons/arrow-right.png", 80, 80),
}
@property
def text(self) -> str:
return self._input_text
def text(self):
return self._input_box.text
def clear(self):
self._input_text = ""
self._layout_name = "lowercase"
self._caps_lock = False
self._input_box.clear()
self._backspace_pressed = False
def render(self, rect, title, sub_title):
gui_label(rl.Rectangle(rect.x, rect.y, rect.width, 95), title, 90)
gui_label(rl.Rectangle(rect.x, rect.y + 95, rect.width, 60), sub_title, 55, rl.GRAY)
if gui_button(rl.Rectangle(rect.x + rect.width - 300, rect.y, 300, 100), "Cancel"):
return -1
def render(self, title: str, sub_title: str):
rect = rl.Rectangle(CONTENT_MARGIN, CONTENT_MARGIN, gui_app.width - 2 * CONTENT_MARGIN, gui_app.height - 2 * CONTENT_MARGIN)
gui_label(rl.Rectangle(rect.x, rect.y, rect.width, 95), title, 90, font_weight=FontWeight.BOLD)
gui_label(rl.Rectangle(rect.x, rect.y + 95, rect.width, 60), sub_title, 55, font_weight=FontWeight.NORMAL)
if gui_button(rl.Rectangle(rect.x + rect.width - 386, rect.y, 386, 125), "Cancel"):
self.clear()
return 0
# Text box for input
rl.gui_text_box(rl.Rectangle(rect.x, rect.y + 160, rect.width, 100), self._input_text, self._max_text_size, True)
# Draw input box and password toggle
input_margin = 25
input_box_rect = rl.Rectangle(rect.x + input_margin, rect.y + 160, rect.width - input_margin, 100)
self._render_input_area(input_box_rect)
# Process backspace key repeat if it's held down
if not rl.is_mouse_button_down(rl.MouseButton.MOUSE_BUTTON_LEFT):
self._backspace_pressed = False
if self._backspace_pressed:
current_time = time.monotonic()
time_since_press = current_time - self._backspace_press_time
# After initial delay, start repeating with shorter intervals
if time_since_press > DELETE_REPEAT_DELAY:
time_since_last_repeat = current_time - self._backspace_last_repeat
if time_since_last_repeat > DELETE_REPEAT_INTERVAL:
self._input_box.delete_char_before_cursor()
self._backspace_last_repeat = current_time
layout = KEYBOARD_LAYOUTS[self._layout_name]
h_space, v_space = 15, 15
row_y_start = rect.y + 300 # Starting Y position for the first row
key_height = (rect.height - 300 - 3 * v_space) / 4
key_max_width = (rect.width - (len(self._layout[2]) - 1) * h_space) / len(self._layout[2])
key_max_width = (rect.width - (len(layout[2]) - 1) * h_space) / len(layout[2])
# Iterate over the rows of keys in the current layout
for row, keys in enumerate(self._layout):
for row, keys in enumerate(layout):
key_width = min((rect.width - (180 if row == 1 else 0) - h_space * (len(keys) - 1)) / len(keys), key_max_width)
start_x = rect.x + (90 if row == 1 else 0)
@@ -81,24 +137,97 @@ class Keyboard:
key_rect = rl.Rectangle(start_x, row_y_start + row * (key_height + v_space), new_width, key_height)
start_x += new_width
if gui_button(key_rect, key):
is_enabled = key != ENTER_KEY or len(self._input_box.text) >= self._min_text_size
result = -1
# Check for backspace key press-and-hold
mouse_pos = rl.get_mouse_position()
mouse_over_key = rl.check_collision_point_rec(mouse_pos, key_rect)
if key == BACKSPACE_KEY and mouse_over_key:
if rl.is_mouse_button_pressed(rl.MouseButton.MOUSE_BUTTON_LEFT):
self._backspace_pressed = True
self._backspace_press_time = time.monotonic()
self._backspace_last_repeat = time.monotonic()
if key in self._key_icons:
if key == SHIFT_ACTIVE_KEY and self._caps_lock:
key = CAPS_LOCK_KEY
texture = self._key_icons[key]
result = gui_button(key_rect, "", icon=texture, button_style=ButtonStyle.PRIMARY if key == ENTER_KEY else ButtonStyle.NORMAL, is_enabled=is_enabled)
else:
result = gui_button(key_rect, key, KEY_FONT_SIZE, is_enabled=is_enabled)
if result:
if key == ENTER_KEY:
return 1
else:
self.handle_key_press(key)
return 0
return -1
def _render_input_area(self, input_rect: rl.Rectangle):
if self._show_password_toggle:
self._input_box.set_password_mode(self._password_mode)
self._input_box.render(rl.Rectangle(input_rect.x, input_rect.y, input_rect.width - 100, input_rect.height))
# render eye icon
eye_texture = self._eye_closed_texture if self._password_mode else self._eye_open_texture
eye_rect = rl.Rectangle(input_rect.x + input_rect.width - 90, input_rect.y, 80, input_rect.height)
eye_x = eye_rect.x + (eye_rect.width - eye_texture.width) / 2
eye_y = eye_rect.y + (eye_rect.height - eye_texture.height) / 2
rl.draw_texture_v(eye_texture, rl.Vector2(eye_x, eye_y), rl.WHITE)
# Handle click on eye icon
if rl.is_mouse_button_pressed(rl.MouseButton.MOUSE_BUTTON_LEFT) and rl.check_collision_point_rec(
rl.get_mouse_position(), eye_rect
):
self._password_mode = not self._password_mode
else:
self._input_box.render(input_rect)
rl.draw_line_ex(
rl.Vector2(input_rect.x, input_rect.y + input_rect.height - 2),
rl.Vector2(input_rect.x + input_rect.width, input_rect.y + input_rect.height - 2),
3.0, # 3 pixel thickness
rl.Color(189, 189, 189, 255),
)
def handle_key_press(self, key):
if key in (SHIFT_DOWN_KEY, ABC_KEY):
self._layout = keyboard_layouts["lowercase"]
elif key == SHIFT_KEY:
self._layout = keyboard_layouts["uppercase"]
if key in (CAPS_LOCK_KEY, ABC_KEY):
self._caps_lock = False
self._layout_name = "lowercase"
elif key == SHIFT_INACTIVE_KEY:
self._last_shift_press_time = time.monotonic()
self._layout_name = "uppercase"
elif key == SHIFT_ACTIVE_KEY:
if time.monotonic() - self._last_shift_press_time < DOUBLE_CLICK_THRESHOLD:
self._caps_lock = True
else:
self._layout_name = "lowercase"
elif key == NUMERIC_KEY:
self._layout = keyboard_layouts["numbers"]
self._layout_name = "numbers"
elif key == SYMBOL_KEY:
self._layout = keyboard_layouts["specials"]
elif key == BACKSPACE_KEY and len(self._input_text) > 0:
self._input_text = self._input_text[:-1]
elif key != BACKSPACE_KEY and len(self._input_text) < self._max_text_size:
self._input_text += key
self._layout_name = "specials"
elif key == BACKSPACE_KEY:
self._input_box.delete_char_before_cursor()
else:
self._input_box.add_char_at_cursor(key)
if not self._caps_lock and self._layout_name == "uppercase":
self._layout_name = "lowercase"
if __name__ == "__main__":
gui_app.init_window("Keyboard")
keyboard = Keyboard(min_text_size=8, show_password_toggle=True)
for _ in gui_app.render():
result = keyboard.render("Keyboard", "Type here")
if result == 1:
print(f"You typed: {keyboard.text}")
gui_app.request_close()
elif result == 0:
print("Canceled")
gui_app.request_close()
gui_app.close()
+244
View File
@@ -0,0 +1,244 @@
from dataclasses import dataclass
from threading import Lock
from typing import Literal
import pyray as rl
from openpilot.system.ui.lib.application import gui_app
from openpilot.system.ui.lib.button import ButtonStyle, gui_button
from openpilot.system.ui.lib.label import gui_label
from openpilot.system.ui.lib.scroll_panel import GuiScrollPanel
from openpilot.system.ui.lib.wifi_manager import NetworkInfo, WifiManagerCallbacks, WifiManagerWrapper, SecurityType
from openpilot.system.ui.widgets.keyboard import Keyboard
from openpilot.system.ui.widgets.confirm_dialog import confirm_dialog
NM_DEVICE_STATE_NEED_AUTH = 60
MIN_PASSWORD_LENGTH = 8
MAX_PASSWORD_LENGTH = 64
ITEM_HEIGHT = 160
ICON_SIZE = 50
STRENGTH_ICONS = [
"icons/wifi_strength_low.png",
"icons/wifi_strength_medium.png",
"icons/wifi_strength_high.png",
"icons/wifi_strength_full.png",
]
@dataclass
class StateIdle:
action: Literal["idle"] = "idle"
@dataclass
class StateConnecting:
network: NetworkInfo
action: Literal["connecting"] = "connecting"
@dataclass
class StateNeedsAuth:
network: NetworkInfo
action: Literal["needs_auth"] = "needs_auth"
@dataclass
class StateShowForgetConfirm:
network: NetworkInfo
action: Literal["show_forget_confirm"] = "show_forget_confirm"
@dataclass
class StateForgetting:
network: NetworkInfo
action: Literal["forgetting"] = "forgetting"
UIState = StateIdle | StateConnecting | StateNeedsAuth | StateShowForgetConfirm | StateForgetting
class WifiManagerUI:
def __init__(self, wifi_manager: WifiManagerWrapper):
self.state: UIState = StateIdle()
self.btn_width: int = 200
self.scroll_panel = GuiScrollPanel()
self.keyboard = Keyboard(max_text_size=MAX_PASSWORD_LENGTH, min_text_size=MIN_PASSWORD_LENGTH, show_password_toggle=True)
self._networks: list[NetworkInfo] = []
self._lock = Lock()
self.wifi_manager = wifi_manager
self.wifi_manager.set_callbacks(
WifiManagerCallbacks(
need_auth = self._on_need_auth,
activated = self._on_activated,
forgotten = self._on_forgotten,
networks_updated = self._on_network_updated,
connection_failed = self._on_connection_failed
)
)
self.wifi_manager.start()
self.wifi_manager.connect()
def render(self, rect: rl.Rectangle):
with self._lock:
if not self._networks:
gui_label(rect, "Scanning Wi-Fi networks...", 72, alignment=rl.GuiTextAlignment.TEXT_ALIGN_CENTER)
return
match self.state:
case StateNeedsAuth(network):
result = self.keyboard.render("Enter password", f"for {network.ssid}")
if result == 1:
password = self.keyboard.text
self.keyboard.clear()
if len(password) >= MIN_PASSWORD_LENGTH:
self.connect_to_network(network, password)
elif result == 0:
self.state = StateIdle()
case StateShowForgetConfirm(network):
result = confirm_dialog(f'Forget Wi-Fi Network "{network.ssid}"?', "Forget")
if result == 1:
self.forget_network(network)
elif result == 0:
self.state = StateIdle()
case _:
self._draw_network_list(rect)
@property
def require_full_screen(self) -> bool:
"""Check if the WiFi UI requires exclusive full-screen rendering."""
with self._lock:
return isinstance(self.state, (StateNeedsAuth, StateShowForgetConfirm))
def _draw_network_list(self, rect: rl.Rectangle):
content_rect = rl.Rectangle(rect.x, rect.y, rect.width, len(self._networks) * ITEM_HEIGHT)
offset = self.scroll_panel.handle_scroll(rect, content_rect)
clicked = self.scroll_panel.is_click_valid()
rl.begin_scissor_mode(int(rect.x), int(rect.y), int(rect.width), int(rect.height))
for i, network in enumerate(self._networks):
y_offset = rect.y + i * ITEM_HEIGHT + offset.y
item_rect = rl.Rectangle(rect.x, y_offset, rect.width, ITEM_HEIGHT)
if not rl.check_collision_recs(item_rect, rect):
continue
self._draw_network_item(item_rect, network, clicked)
if i < len(self._networks) - 1:
line_y = int(item_rect.y + item_rect.height - 1)
rl.draw_line(int(item_rect.x), int(line_y), int(item_rect.x + item_rect.width), line_y, rl.LIGHTGRAY)
rl.end_scissor_mode()
def _draw_network_item(self, rect, network: NetworkInfo, clicked: bool):
spacing = 50
ssid_rect = rl.Rectangle(rect.x, rect.y, rect.width - self.btn_width * 2, ITEM_HEIGHT)
signal_icon_rect = rl.Rectangle(rect.x + rect.width - ICON_SIZE, rect.y + (ITEM_HEIGHT - ICON_SIZE) / 2, ICON_SIZE, ICON_SIZE)
security_icon_rect = rl.Rectangle(signal_icon_rect.x - spacing - ICON_SIZE, rect.y + (ITEM_HEIGHT - ICON_SIZE) / 2, ICON_SIZE, ICON_SIZE)
gui_label(ssid_rect, network.ssid, 55)
status_text = ""
match self.state:
case StateConnecting(network=connecting):
if connecting.ssid == network.ssid:
status_text = "CONNECTING..."
case StateForgetting(network=forgetting):
if forgetting.ssid == network.ssid:
status_text = "FORGETTING..."
if status_text:
status_text_rect = rl.Rectangle(security_icon_rect.x - 410, rect.y, 410, ITEM_HEIGHT)
gui_label(status_text_rect, status_text, font_size=48, alignment=rl.GuiTextAlignment.TEXT_ALIGN_CENTER)
else:
# If the network is saved, show the "Forget" button
if network.is_saved:
forget_btn_rect = rl.Rectangle(security_icon_rect.x - self.btn_width - spacing,
rect.y + (ITEM_HEIGHT - 80) / 2,
self.btn_width,
80,
)
if isinstance(self.state, StateIdle) and gui_button(forget_btn_rect, "Forget", button_style=ButtonStyle.ACTION) and clicked:
self.state = StateShowForgetConfirm(network)
self._draw_status_icon(security_icon_rect, network)
self._draw_signal_strength_icon(signal_icon_rect, network)
if isinstance(self.state, StateIdle) and rl.check_collision_point_rec(rl.get_mouse_position(), ssid_rect) and clicked:
if not network.is_saved and network.security_type != SecurityType.OPEN:
self.state = StateNeedsAuth(network)
elif not network.is_connected:
self.connect_to_network(network)
def _draw_status_icon(self, rect, network: NetworkInfo):
"""Draw the status icon based on network's connection state"""
icon_file = None
if network.is_connected:
icon_file = "icons/checkmark.png"
elif network.security_type == SecurityType.UNSUPPORTED:
icon_file = "icons/circled_slash.png"
elif network.security_type != SecurityType.OPEN:
icon_file = "icons/lock_closed.png"
if not icon_file:
return
texture = gui_app.texture(icon_file, ICON_SIZE, ICON_SIZE)
icon_rect = rl.Vector2(rect.x, rect.y + (ICON_SIZE - texture.height) / 2)
rl.draw_texture_v(texture, icon_rect, rl.WHITE)
def _draw_signal_strength_icon(self, rect: rl.Rectangle, network: NetworkInfo):
"""Draw the Wi-Fi signal strength icon based on network's signal strength"""
strength_level = max(0, min(3, round(network.strength / 33.0)))
rl.draw_texture_v(gui_app.texture(STRENGTH_ICONS[strength_level], ICON_SIZE, ICON_SIZE), rl.Vector2(rect.x, rect.y), rl.WHITE)
def connect_to_network(self, network: NetworkInfo, password=''):
self.state = StateConnecting(network)
if network.is_saved and not password:
self.wifi_manager.activate_connection(network.ssid)
else:
self.wifi_manager.connect_to_network(network.ssid, password)
def forget_network(self, network: NetworkInfo):
self.state = StateForgetting(network)
network.is_saved = False
self.wifi_manager.forget_connection(network.ssid)
def _on_network_updated(self, networks: list[NetworkInfo]):
with self._lock:
self._networks = networks
def _on_need_auth(self, ssid):
with self._lock:
network = next((n for n in self._networks if n.ssid == ssid), None)
if network:
self.state = StateNeedsAuth(network)
def _on_activated(self):
with self._lock:
if isinstance(self.state, StateConnecting):
self.state = StateIdle()
def _on_forgotten(self, ssid):
with self._lock:
if isinstance(self.state, StateForgetting):
self.state = StateIdle()
def _on_connection_failed(self, ssid: str, error: str):
with self._lock:
if isinstance(self.state, StateConnecting):
self.state = StateIdle()
def main():
gui_app.init_window("Wi-Fi Manager")
wifi_manager = WifiManagerWrapper()
wifi_ui = WifiManagerUI(wifi_manager)
for _ in gui_app.render():
wifi_ui.render(rl.Rectangle(50, 50, gui_app.width - 100, gui_app.height - 100))
wifi_manager.shutdown()
gui_app.close()
if __name__ == "__main__":
main()
+81
View File
@@ -0,0 +1,81 @@
import pyray as rl
from openpilot.system.ui.lib.button import gui_button, ButtonStyle, TextAlignment
from openpilot.system.ui.lib.label import gui_label
from openpilot.system.ui.lib.scroll_panel import GuiScrollPanel
class MultiOptionDialog:
def __init__(self, title, options, current=""):
self._title = title
self._options = options
self._current = current if current in options else ""
self._selection = self._current
self._option_height = 80
self._padding = 20
self.scroll_panel = GuiScrollPanel()
@property
def selection(self):
return self._selection
def render(self, rect):
title_rect = rl.Rectangle(rect.x + self._padding, rect.y + self._padding, rect.width - 2 * self._padding, 70)
gui_label(title_rect, self._title, 70)
options_y_start = rect.y + 120
options_height = len(self._options) * (self._option_height + 10)
options_rect = rl.Rectangle(rect.x + self._padding, options_y_start, rect.width - 2 * self._padding, options_height)
view_rect = rl.Rectangle(
rect.x + self._padding, options_y_start, rect.width - 2 * self._padding, rect.height - 200 - 2 * self._padding
)
offset = self.scroll_panel.handle_scroll(view_rect, options_rect)
is_click_valid = self.scroll_panel.is_click_valid()
rl.begin_scissor_mode(int(view_rect.x), int(view_rect.y), int(view_rect.width), int(view_rect.height))
for i, option in enumerate(self._options):
y_pos = view_rect.y + i * (self._option_height + 10) + offset.y
item_rect = rl.Rectangle(view_rect.x, y_pos, view_rect.width, self._option_height)
if not rl.check_collision_recs(item_rect, view_rect):
continue
is_selected = option == self._selection
button_style = ButtonStyle.PRIMARY if is_selected else ButtonStyle.NORMAL
if gui_button(item_rect, option, button_style=button_style, text_alignment=TextAlignment.LEFT) and is_click_valid:
self._selection = option
rl.end_scissor_mode()
button_y = rect.y + rect.height - 80 - self._padding
button_width = (rect.width - 3 * self._padding) / 2
cancel_rect = rl.Rectangle(rect.x + self._padding, button_y, button_width, 80)
if gui_button(cancel_rect, "Cancel"):
return 0 # Canceled
select_rect = rl.Rectangle(rect.x + 2 * self._padding + button_width, button_y, button_width, 80)
has_new_selection = self._selection != "" and self._selection != self._current
if gui_button(select_rect, "Select", is_enabled=has_new_selection, button_style=ButtonStyle.PRIMARY):
return 1 # Selected
return -1 # Still active
if __name__ == "__main__":
from openpilot.system.ui.lib.application import gui_app
gui_app.init_window("Multi Option Dialog Example")
options = [f"Option {i}" for i in range(1, 11)]
dialog = MultiOptionDialog("Choose an option", options, options[0])
for _ in gui_app.render():
result = dialog.render(rl.Rectangle(100, 100, 1024, 800))
if result >= 0:
print(f"Selected: {dialog.selection}" if result > 0 else "Canceled")
break