diff --git a/openpilot/selfdrive/ui/layouts/arabesque/__init__.py b/openpilot/selfdrive/ui/layouts/arabesque/__init__.py new file mode 100644 index 0000000000..79a226b819 --- /dev/null +++ b/openpilot/selfdrive/ui/layouts/arabesque/__init__.py @@ -0,0 +1 @@ +from openpilot.selfdrive.ui.layouts.arabesque.desktop import ArabesqueLayout as ArabesqueLayout diff --git a/openpilot/selfdrive/ui/layouts/arabesque/desktop.py b/openpilot/selfdrive/ui/layouts/arabesque/desktop.py new file mode 100644 index 0000000000..144bd89ae1 --- /dev/null +++ b/openpilot/selfdrive/ui/layouts/arabesque/desktop.py @@ -0,0 +1,153 @@ +from collections.abc import Callable + +import pyray as rl + +from openpilot.selfdrive.ui.layouts.arabesque.theme import DEFAULT_THEME +from openpilot.selfdrive.ui.layouts.arabesque.tools.terminal import Terminal +from openpilot.system.ui.lib.application import gui_app, MouseEvent, MousePos, TextAlignment +from openpilot.system.ui.lib.multilang import tr +from openpilot.system.ui.widgets import Widget +from openpilot.system.ui.widgets.label import gui_label + + +_DOCK_SHADOW = rl.Color(0, 0, 0, 52) +_DOCK_FACE = rl.Color(72, 54, 98, 255) +_DOCK_RIM = rl.Color(113, 88, 147, 255) + + +class ArabesqueLayout(Widget): + def __init__(self, on_exit: Callable[[], None]): + super().__init__() + self._on_exit = on_exit + self._menu_open = False + self._pressed_rect: rl.Rectangle | None = None + self._active_tool: Widget | None = None + self._active_name = "" + + def _update_layout_rects(self): + rect = self._rect + # GuiApplication scales both drawing and input from these logical coordinates. + self._scale = s = min(rect.width / 2160, rect.height / 1080) + self._bar = rl.Rectangle(rect.x, rect.y, rect.width, 104 * s) + self._menu_button = rl.Rectangle(rect.x + 24 * s, rect.y + 8 * s, 88 * s, 88 * s) + self._exit_button = rl.Rectangle(rect.x + 24 * s, rect.y + 116 * s, 456 * s, 116 * s) + self._back_button = rl.Rectangle(rect.x + rect.width - 296 * s, rect.y + 8 * s, 272 * s, 88 * s) + self._dock = dock = rl.Rectangle(rect.x + (rect.width - 360 * s) / 2, rect.y + rect.height - 272 * s, 360 * s, 240 * s) + self._dock_shelf = rl.Rectangle(dock.x, dock.y + 178 * s, dock.width, 10 * s) + self._dock_top = (rl.Vector2(dock.x + 36 * s, dock.y + 118 * s), rl.Vector2(dock.x, self._dock_shelf.y), + rl.Vector2(dock.x + dock.width, self._dock_shelf.y), rl.Vector2(dock.x + dock.width - 36 * s, dock.y + 118 * s)) + self._dock_shadows = (rl.Rectangle(dock.x - 10 * s, dock.y + 174 * s, dock.width + 20 * s, 32 * s), + rl.Rectangle(dock.x - 4 * s, dock.y + 178 * s, dock.width + 8 * s, 22 * s)) + self._dock_icon = rl.Vector2(dock.x + dock.width / 2, dock.y + 72 * s) + self._tool_rect = rl.Rectangle(rect.x, self._bar.y + self._bar.height, rect.width, rect.height - self._bar.height) + + def _open_tool(self, name: str, tool: Widget): + self._close_tool() + self._active_tool, self._active_name = tool, name + tool.set_rect(self._tool_rect) + tool.show_event() + + def _close_tool(self): + tool, self._active_tool = self._active_tool, None + self._active_name = "" + self._menu_open = False + self._pressed_rect = None + gui_app.mouse_events.clear() + # Transient tools have one explicit lifecycle owner, rather than persistent _child registration. + if tool is not None: + tool.hide_event() + + def hide_event(self): + super().hide_event() + self._close_tool() + + def _handle_mouse_press(self, pos: MousePos): + self._pressed_rect = None + if rl.check_collision_point_rec(pos, self._menu_button): + self._pressed_rect = self._menu_button + elif self._menu_open: + if rl.check_collision_point_rec(pos, self._exit_button): + self._pressed_rect = self._exit_button + elif self._active_tool is not None and rl.check_collision_point_rec(pos, self._back_button): + self._pressed_rect = self._back_button + elif self._active_tool is None and rl.check_collision_point_rec(pos, self._dock): + self._pressed_rect = self._dock + + def _handle_mouse_release(self, pos: MousePos): + pressed = self._pressed_rect + self._pressed_rect = None + if pressed is self._menu_button and rl.check_collision_point_rec(pos, self._menu_button): + self._menu_open = not self._menu_open + elif self._menu_open: + self._menu_open = False + if pressed is self._exit_button and rl.check_collision_point_rec(pos, self._exit_button): + self._on_exit() + elif pressed is self._back_button and rl.check_collision_point_rec(pos, self._back_button): + self._close_tool() + elif pressed is self._dock and rl.check_collision_point_rec(pos, self._dock): + self._open_tool(tr("Terminal"), Terminal()) + + def _handle_mouse_event(self, mouse_event: MouseEvent): + if mouse_event.left_released and not rl.check_collision_point_rec(mouse_event.pos, self._rect): + self._pressed_rect = None + self._menu_open = False + + def _render(self, rect: rl.Rectangle): + s, theme = self._scale, DEFAULT_THEME + rl.draw_rectangle_rec(rect, theme.background) + if self._active_tool is not None: + # Widget handles input after drawing: gate the whole frame before the tool sees chrome gestures. + tool_enabled = self.enabled and not self._menu_open and self._pressed_rect is None and not any( + event.slot == 0 and not rl.check_collision_point_rec(event.pos, self._tool_rect) for event in gui_app.mouse_events) + self._active_tool.set_enabled(tool_enabled) + rl.begin_scissor_mode(int(self._tool_rect.x), int(self._tool_rect.y), int(self._tool_rect.width), int(self._tool_rect.height)) + self._active_tool.render(self._tool_rect) + rl.end_scissor_mode() + + rl.draw_rectangle_rec(self._bar, theme.surface) + rl.draw_line_ex(rl.Vector2(rect.x, self._bar.y + self._bar.height), + rl.Vector2(rect.x + rect.width, self._bar.y + self._bar.height), 2 * s, theme.control) + menu_pressed = self._menu_open or self._pressed_rect is self._menu_button and self.is_pressed + rl.draw_rectangle_rounded(self._menu_button, 0.28, 12, theme.pressed if menu_pressed else theme.control) + + center = rl.Vector2(self._menu_button.x + 42 * s, self._menu_button.y + 44 * s) + rl.draw_ring(center, 15 * s, 20 * s, 0, 360, 24, theme.background) + rl.draw_line_ex(rl.Vector2(center.x + 18 * s, center.y - 20 * s), rl.Vector2(center.x + 18 * s, center.y + 20 * s), + 5 * s, theme.background) + gui_label(rl.Rectangle(rect.x + 136 * s, rect.y, self._back_button.x - rect.x - 160 * s, self._bar.height), + self._active_name or "Arabesque", round(40 * s), theme.text, theme.font_weight) + + if self._active_tool is not None: + back_pressed = self._pressed_rect is self._back_button and self.is_pressed + rl.draw_rectangle_rounded(self._back_button, 0.28, 12, theme.pressed if back_pressed else theme.control) + gui_label(self._back_button, "< " + tr("Desktop"), round(32 * s), theme.background, theme.font_weight, + alignment=TextAlignment.CENTER) + else: + # Two fixed translucent shapes soften the shelf's shadow without a blur pass or texture. + for shadow in self._dock_shadows: + rl.draw_rectangle_rounded(shadow, 0.9, 12, _DOCK_SHADOW) + rl.draw_rectangle_rec(self._dock_shelf, theme.surface) + top = self._dock_top + rl.draw_triangle(top[0], top[1], top[2], _DOCK_FACE) + rl.draw_triangle(top[0], top[2], top[3], _DOCK_FACE) + rl.draw_line_ex(top[1], top[2], 2 * s, _DOCK_RIM) + icon = self._dock_icon + rl.draw_ellipse(int(icon.x), int(icon.y + 68 * s), 52 * s, 9 * s, _DOCK_SHADOW) + dock_pressed = self._pressed_rect is self._dock and self.is_pressed + tint = theme.pressed if dock_pressed else theme.control + rl.draw_circle_v(rl.Vector2(icon.x, icon.y + 5 * s), 64 * s, theme.surface) + rl.draw_circle_v(icon, 64 * s, tint) + rl.draw_circle_v(icon, 54 * s, theme.background) + rl.draw_line_ex(rl.Vector2(icon.x - 32 * s, icon.y - 24 * s), rl.Vector2(icon.x - 8 * s, icon.y), 6 * s, tint) + rl.draw_line_ex(rl.Vector2(icon.x - 8 * s, icon.y), rl.Vector2(icon.x - 32 * s, icon.y + 24 * s), 6 * s, tint) + rl.draw_line_ex(rl.Vector2(icon.x + 8 * s, icon.y + 24 * s), rl.Vector2(icon.x + 36 * s, icon.y + 24 * s), 6 * s, tint) + gui_label(rl.Rectangle(icon.x - 96 * s, icon.y + 68 * s, 192 * s, 36 * s), + tr("Terminal"), round(28 * s), theme.text, theme.font_weight, alignment=TextAlignment.CENTER) + + if self._menu_open: + exit_pressed = self._pressed_rect is self._exit_button and self.is_pressed + color = theme.pressed if exit_pressed else theme.surface + rl.draw_rectangle_rounded(self._exit_button, 0.22, 12, color) + rl.draw_rectangle_rounded_lines_ex(self._exit_button, 0.22, 12, 2 * s, theme.control) + gui_label(rl.Rectangle(self._exit_button.x + 28 * s, self._exit_button.y, self._exit_button.width - 56 * s, self._exit_button.height), + tr("Exit Arabesque"), round(40 * s), theme.background if exit_pressed else theme.text, theme.font_weight) diff --git a/openpilot/selfdrive/ui/layouts/arabesque/theme.py b/openpilot/selfdrive/ui/layouts/arabesque/theme.py new file mode 100644 index 0000000000..804d6b2cdd --- /dev/null +++ b/openpilot/selfdrive/ui/layouts/arabesque/theme.py @@ -0,0 +1,27 @@ +from dataclasses import dataclass + +import pyray as rl + +from openpilot.system.ui.lib.application import FontWeight + + +@dataclass(frozen=True) +class Theme: + background: rl.Color + surface: rl.Color + text: rl.Color + secondary_text: rl.Color + control: rl.Color + pressed: rl.Color + font_weight: FontWeight = FontWeight.NORMAL + + +# Shared colors are read-only; use dataclasses.replace() for a tool's overrides. +DEFAULT_THEME = Theme( + background=rl.Color(12, 10, 18, 255), + surface=rl.Color(58, 42, 82, 255), + text=rl.Color(236, 242, 250, 255), + secondary_text=rl.Color(210, 196, 229, 255), + control=rl.Color(205, 181, 247, 255), + pressed=rl.Color(175, 140, 235, 255), +) diff --git a/openpilot/selfdrive/ui/layouts/arabesque/tools/__init__.py b/openpilot/selfdrive/ui/layouts/arabesque/tools/__init__.py new file mode 100644 index 0000000000..e69de29bb2 diff --git a/openpilot/selfdrive/ui/layouts/arabesque/tools/terminal.py b/openpilot/selfdrive/ui/layouts/arabesque/tools/terminal.py new file mode 100644 index 0000000000..351ce0349f --- /dev/null +++ b/openpilot/selfdrive/ui/layouts/arabesque/tools/terminal.py @@ -0,0 +1,512 @@ +"""One local shell; the installed tmux owns terminal interpretation.""" + +from collections import deque +from importlib.resources import as_file +from itertools import groupby +import os +import re +import subprocess +import tempfile +import time +import unicodedata + +import pyray as rl + +from openpilot.selfdrive.ui.layouts.arabesque.theme import DEFAULT_THEME, Theme +from openpilot.system.ui.lib.application import FONT_DIR, FONT_SCALE, FontWeight, gui_app, MousePos, TextAlignment +from openpilot.system.ui.lib.multilang import tr +from openpilot.system.ui.widgets import Widget, device +from openpilot.system.ui.widgets.keyboard import KEYBOARD_LAYOUTS, BACKSPACE_KEY, ENTER_KEY, SHIFT_INACTIVE_KEY, SHIFT_ACTIVE_KEY +from openpilot.system.ui.widgets.label import gui_label + +_SGR = re.compile(r'\x1b\[([0-9;:]*)m|([\x0e\x0f])') +_LINK = re.compile(r'\x1b\]8;[^\x07\x1b]*(?:\x07|\x1b\\)') +_ACS = str.maketrans({'q': '─', 'x': '│', 'l': '┌', 'k': '┐', 'm': '└', 'j': '┘', 'n': '┼', 't': '├', 'u': '┤', 'v': '┴', 'w': '┬', 'a': '▒'}) +_PALETTE = ((0, 0, 0), (205, 49, 49), (13, 188, 121), (229, 229, 16), (36, 114, 200), (188, 63, 188), (17, 168, 205), + (229, 229, 229), (102, 102, 102), (241, 76, 76), (35, 209, 139), (245, 245, 67), (59, 142, 234), + (214, 112, 214), (41, 184, 219), (255, 255, 255)) +_KEYS = {rl.KEY_ENTER: 'Enter', rl.KEY_KP_ENTER: 'Enter', rl.KEY_BACKSPACE: 'BSpace', rl.KEY_TAB: 'Tab', + rl.KEY_ESCAPE: 'Escape', rl.KEY_LEFT: 'Left', rl.KEY_RIGHT: 'Right', rl.KEY_UP: 'Up', rl.KEY_DOWN: 'Down', + rl.KEY_HOME: 'Home', rl.KEY_END: 'End', rl.KEY_DELETE: 'DC', rl.KEY_INSERT: 'IC', + rl.KEY_PAGE_UP: 'PPage', rl.KEY_PAGE_DOWN: 'NPage', **{getattr(rl, f'KEY_F{i}'): f'F{i}' for i in range(1, 13)}} + +_KEY_NAMES = frozenset(_KEYS.values()) | {'C-Left', 'C-Right', 'C-Up', 'C-Down', 'C-Home', 'C-End', 'BTab'} + + +def _color(index): + if index < 16: + return _PALETTE[index] + if index < 232: + levels = (0, 95, 135, 175, 215, 255) + return tuple(levels[(index - 16) // divisor % 6] for divisor in (36, 6, 1)) + return (8 + (index - 232) * 10,) * 3 + + +def _styled_rows(lines): + """tmux positioned the text; interpret only its exported cell styling.""" + rows, fg, bg, reverse, hidden, acs = [], None, None, False, False, False + for line in lines: + line = _LINK.sub('', line) + spans, column, start = [], 0, 0 + for match in (*_SGR.finditer(line), None): + end = match.start() if match else len(line) + text = line[start:end].translate(_ACS) if acs else line[start:end] + if text: + width = sum(0 if unicodedata.combining(c) else 2 if unicodedata.east_asian_width(c) in ('W', 'F') else 1 for c in text) + spans.append((column, text, fg, bg, reverse, hidden, width)) + column += width + if match is None: + break + if match[2]: + acs = match[2] == '\x0e' + start = match.end() + continue + codes = [int(value or 0) for value in match[1].replace(':2::', ';2;').replace(':', ';').split(';')] + i = 0 + while i < len(codes): + code = codes[i] + if code == 0: + fg, bg, reverse, hidden = None, None, False, False + elif code in (7, 27): + reverse = code == 7 + elif code in (8, 28): + hidden = code == 8 + elif 30 <= code <= 37 or 90 <= code <= 97: + fg = _color(code - 30 if code < 90 else code - 90 + 8) + elif 40 <= code <= 47 or 100 <= code <= 107: + bg = _color(code - 40 if code < 100 else code - 100 + 8) + elif code == 39: + fg = None + elif code == 49: + bg = None + elif code in (38, 48, 58) and i + 1 < len(codes): + count = 1 if codes[i + 1] == 5 else 3 if codes[i + 1] == 2 else 0 + values = codes[i + 2:i + 2 + count] + if count and len(values) == count and all(0 <= value <= 255 for value in values): + color = _color(values[0]) if count == 1 else tuple(values) + if code == 38: + fg = color + elif code == 48: + bg = color + i += count + 1 + i += 1 + start = match.end() + rows.append(spans) + return rows + + +class _Session: + def __init__(self, columns, rows): + self.rows, self.cursor, self.error = [], (0, 0, False), '' + self._size = self._desired_size = columns, rows + self._pending = deque([('ack', 0), ('open', 1)]) + self._block, self._body = None, [] + self._buffer, self._out = bytearray(), bytearray() + self._snapshot, self._last_capture = False, 0.0 + self._deadline = time.monotonic() + 3 + self._folder = None + self._process = None + try: + self._folder = tempfile.TemporaryDirectory(prefix='arabesque-') + self._process = subprocess.Popen( + ['tmux', '-u', '-S', self._folder.name + '/socket', '-f', '/dev/null', '-C', + 'set-option', '-s', 'exit-unattached', 'on', ';', 'new-session', '-f', 'no-output', '-s', 'arabesque', '-P', '-F', '#{pane_id}', + '-x', str(columns), '-y', str(rows), '-c', os.path.expanduser('~'), '/bin/bash', '-i'], + stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, bufsize=0, + # A fresh server inherits this UI's environment, including the private host Params namespace. + env={key: value for key, value in os.environ.items() if key not in ('TMUX', 'TMUX_PANE')}) + os.set_blocking(self._process.stdin.fileno(), False) + os.set_blocking(self._process.stdout.fileno(), False) + except OSError: + self.error = tr('Local shell unavailable. Bash and tmux must be installed.') + self.close() + + def _request(self, command, kind='ack', lines=0): + encoded = command.encode() + b'\n' + if len(self._out) + len(encoded) > 32768 or len(self._pending) >= 64: + return False + if not self._pending: + self._deadline = time.monotonic() + 3 + self._out.extend(encoded) + self._pending.append((kind, lines)) + return True + + def keys(self, value): + if self._process is None or not hasattr(self, '_pane') or self.error: + return + if isinstance(value, bytes): + if not value: + return + command = f'send-keys -t {self._pane} -H ' + ' '.join(f'{byte:02x}' for byte in value) if len(value) <= 2048 else '' + else: + assert value in _KEY_NAMES + command = f'send-keys -t {self._pane} {value}' + if not command or not self._request(command): + self.error = tr('Shell input limit reached. Return to Desktop to open a new session.') + self.close() + else: + self._last_capture = 0 + self._flush() + + def resize(self, columns, rows): + self._desired_size = columns, rows + + def _line(self, line): + if self._block is None: + if line.startswith(b'%begin '): + if not self._pending or not re.fullmatch(rb'%begin \d+ \d+ \d+', line): + raise ValueError('Unexpected tmux reply') + self._block, self._body, self._reply_bytes = line[7:], [], 0 + return + kind, count = self._pending[0] + if line == b'%error ' + self._block: + raise ValueError('tmux command failed') + # Count rows before accepting the guard: terminal text can itself start with "%end". + if len(self._body) < count: + self._reply_bytes += len(line) + if self._reply_bytes > 262144: + raise ValueError('Oversized tmux screen') + self._body.append(line) + return + if line != b'%end ' + self._block: + raise ValueError('Malformed tmux reply') + body = [value.decode('utf-8', errors='replace') for value in self._body] + self._pending.popleft() + self._block = None + if kind == 'open': + if not re.fullmatch(r'%\d+', body[0]): + raise ValueError('Missing tmux pane') + self._pane = body[0] + elif kind == 'screen': + if body != getattr(self, '_previous', None): + self.rows = _styled_rows(body) + self._previous = body + elif kind == 'cursor': + x, y, visible = map(int, body[0].split()) + self.cursor = x, y, bool(visible) + self._snapshot = False + + def _flush(self): + if not self._out: + return True + try: + del self._out[:os.write(self._process.stdin.fileno(), self._out)] + except BlockingIOError: + pass + except OSError: + self.error = tr('Local shell connection ended. Return to Desktop to try again.') + self.close() + return False + return True + + def update(self): + process = self._process + if process is None: + return + now = time.monotonic() + try: + if not self._flush(): + return + # Snapshot-only output avoids copying arbitrary command output into Python queues. + remaining = 65536 + while remaining > 0: + try: + chunk = os.read(process.stdout.fileno(), min(16384, remaining)) + except BlockingIOError: + break + if not chunk: + self.error = tr('Shell exited. Return to Desktop to open a new session.') + self.close() + return + remaining -= len(chunk) + self._buffer.extend(chunk) + if len(self._buffer) > 262144: + raise ValueError('Oversized tmux reply') + while (end := self._buffer.find(b'\n')) >= 0: + line = bytes(self._buffer[:end]) + del self._buffer[:end + 1] + self._line(line) + if self._pending and now > self._deadline: + raise ValueError('tmux stopped responding') + if not self._pending and not self._out and not self._snapshot and now - self._last_capture >= 1 / 30: + if self._desired_size != self._size: + self._size = self._desired_size + self._request(f'refresh-client -C {self._size[0]},{self._size[1]}') + self._request(f'capture-pane -p -e -N -t {self._pane} -S 0 -E {self._size[1] - 1}', 'screen', self._size[1]) + self._request(f"display-message -p -t {self._pane} '#{{cursor_x}} #{{cursor_y}} #{{cursor_flag}}'", 'cursor', 1) + self._snapshot, self._last_capture, self._deadline = True, now, now + 3 + self._flush() + except (OSError, ValueError, IndexError): + self.error = tr('Local shell connection ended. Return to Desktop to try again.') + self.close() + + def close(self): + process, self._process = self._process, None + if process is not None: + try: + # This socket belongs only to this tool; never address the launcher server. + os.write(process.stdin.fileno(), b'kill-server\n') + except OSError: + pass + process.stdin.close() + try: + process.wait(timeout=0.02) + except subprocess.TimeoutExpired: + process.kill() + try: + process.wait(timeout=0.02) + except subprocess.TimeoutExpired: + # shortcut: Popen reaps a stalled child on its next launch; add async reaping only if observed on device. + pass + process.stdout.close() + if self._folder is not None: + self._folder.cleanup() + self._out.clear() + self._buffer.clear() + self._pending.clear() + + +class Terminal(Widget): + def __init__(self, theme: Theme = DEFAULT_THEME): + super().__init__() + self._theme = theme + self._session, self._font = None, None + self._fallback_font = None + self._layout_name = 'lowercase' + self._keyboard, self._ctrl = False, False + self._pressed = None + self._buttons, self._draw_rows = [], [] + self._previous_rows = None + + def show_event(self): + super().show_event() + rl.set_exit_key(0) + self._drain_keys() + if self._font is not None: + return + self._font = self._fallback_font = gui_app.font(FontWeight.UNIFONT) + try: + with as_file(FONT_DIR) as directory: + font = rl.load_font_ex(str(directory / 'JetBrainsMono-Medium.ttf'), 64, None, 0) + except OSError: + font = rl.get_font_default() + if font.texture.id != rl.get_font_default().texture.id: + self._font = font + rl.gen_texture_mipmaps(font.texture) + rl.set_texture_filter(font.texture, rl.TextureFilter.TEXTURE_FILTER_TRILINEAR) + self._update_layout_rects() + self._session = _Session(*self._size) + gui_app.add_render_prepare(self._prepare, self.hide_event) + + def hide_event(self): + super().hide_event() + if self._session is not None: + gui_app.remove_render_prepare(self._prepare) + self._session.close() + self._session = None + rl.set_exit_key(rl.KEY_ESCAPE) + self._drain_keys() + if self._font is not None and self._font is not self._fallback_font: + # A Desktop/menu release may close us after glyphs were queued in this frame. + rl.rl_draw_render_batch_active() + rl.unload_font(self._font) + self._font = self._fallback_font = None + self._draw_rows, self._previous_rows, self._pressed = [], None, None + + def _prepare(self): + session = self._session + if session is None: + return + if session.rows is not self._previous_rows: + self._draw_rows = [] + for row in session.rows: + spans = [] + for column, text, foreground, background, reverse, hidden, _width in row: + fg = rl.Color(*foreground, 255) if foreground else self._theme.text + bg = rl.Color(*background, 255) if background else self._theme.background + if reverse: + fg, bg = bg, fg + for offset, run, font, run_width in self._font_runs(text): + spans.append((column + offset, run, fg, bg if background or reverse else None, hidden, run_width, font)) + self._draw_rows.append(spans) + self._previous_rows = session.rows + + def _font_text(self, text): + if text.isascii(): + return text + result = [] + borders = {'─': '-', '│': '|', **dict.fromkeys('┌┐└┘├┤┴┬┼', '+')} + for char in unicodedata.normalize('NFC', text): + if unicodedata.combining(char): + continue + if ord(char) >= 128 and self._fallback_font.glyphs[rl.get_glyph_index(self._fallback_font, ord(char))].value != ord(char): + # Unsupported wide characters must still occupy two terminal columns. + char = borders.get(char, '?') + (' ' if unicodedata.east_asian_width(char) in ('W', 'F') else '') + result.append(char) + return ''.join(result) + + def _font_runs(self, text): + text = self._font_text(text) + if text.isascii(): + yield 0, text, self._font, len(text) + return + column = 0 + for ascii_text, characters in groupby(text, str.isascii): + run = ''.join(characters) + width = len(run) if ascii_text else sum(2 if unicodedata.east_asian_width(char) in ('W', 'F') else 1 for char in run) + yield column, run, self._font if ascii_text else self._fallback_font, width + column += width + + @staticmethod + def _drain_keys(): + while rl.get_key_pressed(): + pass + while rl.get_char_pressed(): + pass + + def _update_layout_rects(self): + rect = self._rect + self._scale = s = min(rect.width / 2160, rect.height / 768) + self._fallback_size = 31.2 * s * FONT_SCALE + self._cell_w = self._fallback_size / 2 # Keep Unifont's 8×16 grid, enlarged 20% as in preview C. + self._cell_h = self._fallback_size * 1.2 + advance = self._font.glyphs[rl.get_glyph_index(self._font, ord('M'))].advanceX if self._font else 0 + self._font_size = self._cell_w * self._font.baseSize / advance if advance else self._fallback_size + keyboard_height = 368 * s if self._keyboard else 0 + self._screen = rl.Rectangle(rect.x + 20 * s, rect.y + 16 * s, rect.width - 40 * s, rect.height - 128 * s - keyboard_height) + self._size = (max(20, min(240, int(self._screen.width / max(1, self._cell_w)))), + max(4, min(80, int(self._screen.height / max(1, self._cell_h))))) + self._buttons = [] + controls = [('Esc', 'Escape'), ('Tab', 'Tab'), ('Ctrl', 'ctrl'), ('Left', 'Left'), ('Down', 'Down'), + ('Up', 'Up'), ('Right', 'Right'), ('Hide keys' if self._keyboard else 'Keyboard', 'keyboard')] + width = (rect.width - 40 * s) / len(controls) + for i, (label, key) in enumerate(controls): + self._buttons.append((rl.Rectangle(rect.x + 20 * s + i * width, rect.y + rect.height - 96 * s, width - 8 * s, 80 * s), label, key)) + if self._keyboard: + labels = {SHIFT_INACTIVE_KEY: 'Shift', SHIFT_ACTIVE_KEY: 'Shift', BACKSPACE_KEY: 'Back', ENTER_KEY: 'Enter', ' ': 'Space'} + for row, keys in enumerate(KEYBOARD_LAYOUTS[self._layout_name]): + unit = (rect.width - 40 * s) / sum(3 if key == ' ' else 1 for key in keys) + x = rect.x + 20 * s + for key in keys: + width = unit * (3 if key == ' ' else 1) + self._buttons.append((rl.Rectangle(x, rect.y + rect.height - 464 * s + row * 88 * s, width - 8 * s, 80 * s), labels.get(key, key), key)) + x += width + if self._session: + self._session.resize(*self._size) + + def _send(self, value): + if self._session is None: + return + if isinstance(value, str): + if value in _KEY_NAMES: + if self._ctrl and 'C-' + value in _KEY_NAMES: + value = 'C-' + value + elif self._ctrl and len(value) == 1: + value = bytes([ord(value.upper()) & 31]) + else: + value = value.encode() + self._ctrl = False + if reset_timeout := getattr(device, '_reset_interactive_timeout', None): + reset_timeout() + self._session.keys(value) + + def _update_state(self): + if self._session is None: + return + # Main has checked off-road authority before rendering this tool. + self._session.update() + self._prepare() + if not self.enabled: + self._pressed, self._ctrl = None, False + self._drain_keys() + return + ctrl = rl.is_key_down(rl.KEY_LEFT_CONTROL) or rl.is_key_down(rl.KEY_RIGHT_CONTROL) + alt = rl.is_key_down(rl.KEY_LEFT_ALT) or rl.is_key_down(rl.KEY_RIGHT_ALT) + characters = deque() + char = rl.get_char_pressed() + while char: + if not ctrl and char >= 32: + characters.append(chr(char)) + char = rl.get_char_pressed() + keys = [] + key = rl.get_key_pressed() + while key: + keys.append(key) + key = rl.get_key_pressed() + for position, key in enumerate(keys): + if key in _KEYS and characters and not any(32 <= later <= 96 for later in keys[position + 1:]): + self._send(''.join(('\x1b' if alt else '') + char for char in characters)) + characters.clear() + if ctrl and 32 <= key <= 95: + self._send(bytes([key & 31])) + elif key in _KEYS: + self._send_key(key, ctrl, alt) + elif 32 <= key <= 96 and characters: + self._send(('\x1b' if alt else '') + characters.popleft()) + for key in _KEYS: + if rl.is_key_pressed_repeat(key): + self._send_key(key, ctrl, alt) + if characters: + self._send(''.join(('\x1b' if alt else '') + char for char in characters)) + + def _send_key(self, key, ctrl, alt): + name = _KEYS[key] + if key == rl.KEY_TAB and (rl.is_key_down(rl.KEY_LEFT_SHIFT) or rl.is_key_down(rl.KEY_RIGHT_SHIFT)): + name = 'BTab' + elif ctrl and 'C-' + name in _KEY_NAMES: + name = 'C-' + name + if alt: + self._send(b'\x1b') + self._send(name) + + def _handle_mouse_press(self, pos: MousePos): + self._pressed = next((key for rect, _, key in self._buttons if rl.check_collision_point_rec(pos, rect)), None) + + def _handle_mouse_release(self, pos: MousePos): + key, self._pressed = self._pressed, None + if key is None or not any(value == key and rl.check_collision_point_rec(pos, rect) for rect, _, value in self._buttons): + return + if key == 'keyboard': + self._keyboard = not self._keyboard + elif key == 'ctrl': + self._ctrl = not self._ctrl + elif key in (SHIFT_ACTIVE_KEY, SHIFT_INACTIVE_KEY): + self._layout_name = 'lowercase' if self._layout_name == 'uppercase' else 'uppercase' + elif key in ('123', '#+=', 'ABC'): + self._layout_name = {'123': 'numbers', '#+=': 'specials', 'ABC': 'lowercase'}[key] + else: + self._send({BACKSPACE_KEY: 'BSpace', ENTER_KEY: 'Enter'}.get(key, key)) + if self._layout_name == 'uppercase': + self._layout_name = 'lowercase' + self._update_layout_rects() + + def _render(self, rect): + theme, s = self._theme, self._scale + rl.draw_rectangle_rec(rect, theme.background) + if self._session and self._font: + rl.begin_scissor_mode(int(self._screen.x), int(self._screen.y), int(self._screen.width), int(self._screen.height)) + # Terminal fonts must not be replaced by the UI language's proportional fallback. + draw_text = getattr(rl, '_orig_draw_text_ex', rl.draw_text_ex) + for row, spans in enumerate(self._draw_rows[:self._size[1]]): + y = self._screen.y + row * self._cell_h + for column, text, fg, bg, hidden, width, font in spans: + x = self._screen.x + column * self._cell_w + if bg: + rl.draw_rectangle_rec(rl.Rectangle(x, y, width * self._cell_w, self._cell_h), bg) + if not hidden: + size = self._font_size if font is self._font else self._fallback_size + draw_text(font, text, rl.Vector2(x, y), size, 0, fg) + x, y, visible = self._session.cursor + if visible and 0 <= x < self._size[0] and 0 <= y < self._size[1]: + rl.draw_rectangle_lines_ex(rl.Rectangle(self._screen.x + x * self._cell_w, self._screen.y + y * self._cell_h, + self._cell_w, self._cell_h), max(1, s), theme.control) + rl.end_scissor_mode() + error = self._session.error if self._session else '' + if not error and not self._draw_rows: + error = tr('Opening local shell…') + if error: + gui_label(self._screen, error, round(30 * s), theme.text, theme.font_weight, alignment=TextAlignment.CENTER) + for bounds, label, key in self._buttons: + pressed = self.enabled and (key == self._pressed and self.is_pressed or key == 'ctrl' and self._ctrl) + rl.draw_rectangle_rounded(bounds, 0.2, 8, theme.control if pressed else theme.surface) + gui_label(bounds, label, round(28 * s), theme.background if pressed else theme.text, theme.font_weight, alignment=TextAlignment.CENTER) diff --git a/openpilot/selfdrive/ui/layouts/main.py b/openpilot/selfdrive/ui/layouts/main.py index 91f799b9f5..6ac7f93786 100644 --- a/openpilot/selfdrive/ui/layouts/main.py +++ b/openpilot/selfdrive/ui/layouts/main.py @@ -5,6 +5,7 @@ from openpilot.system.ui.lib.application import gui_app from openpilot.system.ui.widgets import Widget from openpilot.selfdrive.ui.layouts.sidebar import Sidebar, SIDEBAR_WIDTH from openpilot.selfdrive.ui.layouts.home import HomeLayout +from openpilot.selfdrive.ui.layouts.arabesque import ArabesqueLayout from openpilot.selfdrive.ui.layouts.settings.settings import SettingsLayout, PanelType from openpilot.selfdrive.ui.onroad.augmented_road_view import AugmentedRoadView from openpilot.selfdrive.ui.ui_state import device, ui_state @@ -16,6 +17,7 @@ class MainState(IntEnum): HOME = 0 SETTINGS = 1 ONROAD = 2 + ARABESQUE = 3 class MainLayout(Widget): @@ -33,8 +35,9 @@ class MainLayout(Widget): self._home_body_layout = BodyLayout() self._layouts: dict[MainState, Widget] = { MainState.HOME: self._home_layout, - MainState.SETTINGS: SettingsLayout(), + MainState.SETTINGS: SettingsLayout(on_open_arabesque=self._open_arabesque, arabesque_available=self._arabesque_allowed), MainState.ONROAD: AugmentedRoadView(), + MainState.ARABESQUE: ArabesqueLayout(self._exit_arabesque), } self._sidebar_rect = rl.Rectangle(0, 0, 0, 0) @@ -52,6 +55,8 @@ class MainLayout(Widget): def _render(self, _): self._handle_onroad_transition() + if self._current_mode == MainState.ARABESQUE and not self._arabesque_allowed(): + self._set_mode_for_state() self._render_main_content() def _setup_callbacks(self): @@ -98,10 +103,31 @@ class MainLayout(Widget): def _set_current_layout(self, layout: MainState): if layout != self._current_mode: + if MainState.ARABESQUE in (layout, self._current_mode): + # Discard the handoff frame so queued touches cannot reach the next view. + gui_app.mouse_events.clear() self._layouts[self._current_mode].hide_event() self._current_mode = layout self._layouts[self._current_mode].show_event() + def _arabesque_allowed(self) -> bool: + return ui_state.is_offroad() + + def _open_arabesque(self): + if self._current_mode == MainState.SETTINGS and self._arabesque_allowed(): + self._sidebar.set_visible(False) + self._set_current_layout(MainState.ARABESQUE) + + def _exit_arabesque(self): + if self._arabesque_allowed(): + self._set_current_layout(MainState.SETTINGS) + else: + self._set_mode_for_state() + + def close(self): + if self._current_mode == MainState.ARABESQUE: + self._layouts[MainState.ARABESQUE].hide_event() + def open_settings(self, panel_type: PanelType): self._layouts[MainState.SETTINGS].set_current_panel(panel_type) self._set_current_layout(MainState.SETTINGS) diff --git a/openpilot/selfdrive/ui/layouts/settings/developer.py b/openpilot/selfdrive/ui/layouts/settings/developer.py index 9ad1e0bbd0..1bfae8f845 100644 --- a/openpilot/selfdrive/ui/layouts/settings/developer.py +++ b/openpilot/selfdrive/ui/layouts/settings/developer.py @@ -1,9 +1,11 @@ +from collections.abc import Callable + from openpilot.common.params import Params from openpilot.starpilot.car.hyundai.aol import ioniq6_settings_capable from openpilot.selfdrive.ui.widgets.ssh_key import ssh_key_item from openpilot.selfdrive.ui.ui_state import ui_state from openpilot.system.ui.widgets import Widget -from openpilot.system.ui.widgets.list_view import toggle_item +from openpilot.system.ui.widgets.list_view import button_item, toggle_item from openpilot.system.ui.widgets.scroller_tici import Scroller from openpilot.system.ui.widgets.confirm_dialog import ConfirmDialog from openpilot.system.ui.lib.application import gui_app @@ -31,7 +33,7 @@ DESCRIPTIONS = { class DeveloperLayout(Widget): - def __init__(self): + def __init__(self, on_open_arabesque: Callable | None = None, arabesque_available: Callable[[], bool] = lambda: False): super().__init__() self._params = Params() self._is_release = self._params.get_bool("IsReleaseBranch") @@ -93,6 +95,8 @@ class DeveloperLayout(Widget): self._on_enable_ui_debug(self._params.get_bool("ShowDebugInfo")) self._scroller = Scroller([ + button_item("Arabesque", lambda: tr("OPEN"), callback=on_open_arabesque, + enabled=lambda: on_open_arabesque is not None and arabesque_available()), self._adb_toggle, self._ssh_toggle, self._ssh_keys, diff --git a/openpilot/selfdrive/ui/layouts/settings/settings.py b/openpilot/selfdrive/ui/layouts/settings/settings.py index 48b75e5dbd..0cc8cf4d43 100644 --- a/openpilot/selfdrive/ui/layouts/settings/settings.py +++ b/openpilot/selfdrive/ui/layouts/settings/settings.py @@ -47,7 +47,7 @@ class PanelInfo: class SettingsLayout(Widget): - def __init__(self): + def __init__(self, on_open_arabesque: Callable | None = None, arabesque_available: Callable[[], bool] = lambda: False): super().__init__() self._current_panel = PanelType.DEVICE @@ -61,7 +61,7 @@ class SettingsLayout(Widget): PanelType.TOGGLES: PanelInfo(tr_noop("Toggles"), TogglesLayout()), PanelType.SOFTWARE: PanelInfo(tr_noop("Software"), SoftwareLayout()), PanelType.FIREHOSE: PanelInfo(tr_noop("Firehose"), FirehoseLayout()), - PanelType.DEVELOPER: PanelInfo(tr_noop("Developer"), DeveloperLayout()), + PanelType.DEVELOPER: PanelInfo(tr_noop("Developer"), DeveloperLayout(on_open_arabesque, arabesque_available)), } self._font_medium = gui_app.font(FontWeight.MEDIUM) diff --git a/openpilot/selfdrive/ui/tests/test_arabesque.py b/openpilot/selfdrive/ui/tests/test_arabesque.py new file mode 100644 index 0000000000..1347a0573a --- /dev/null +++ b/openpilot/selfdrive/ui/tests/test_arabesque.py @@ -0,0 +1,413 @@ +"""Exercise real layout routing and Widget input with native service owners isolated.""" + +import ast +from dataclasses import replace +from enum import IntEnum +import gc +from pathlib import Path +from types import SimpleNamespace as NS +from unittest.mock import Mock +import weakref + +import pyray as rl +import pytest + +from openpilot.selfdrive.ui.layouts.arabesque import ArabesqueLayout +from openpilot.selfdrive.ui.layouts.arabesque import desktop +from openpilot.selfdrive.ui.layouts.arabesque.theme import DEFAULT_THEME +from openpilot.selfdrive.ui.layouts.arabesque.tools import terminal +from openpilot.system.ui.lib.application import gui_app, FontWeight, MouseEvent, MousePos +from openpilot.system.ui.widgets import Widget + +ROOT = Path(__file__).resolve().parents[2] + + +def layout_classes(path, names, namespace): + # UI owners import device-native IPC; keep these tests usable on a desktop too. + tree = ast.parse(path.read_text()) + tree.body = ast.parse('from __future__ import annotations').body + [ + node for node in tree.body if isinstance(node, ast.ClassDef) and node.name in names + ] + exec(compile(ast.fix_missing_locations(tree), str(path), 'exec'), namespace) + return NS(**namespace) + + +@pytest.fixture +def shell(monkeypatch): + import openpilot.system.ui.widgets as widgets + monkeypatch.setattr(widgets, 'device', NS(awake=True)) + monkeypatch.setattr(gui_app, '_mouse_events', []) + monkeypatch.setattr(gui_app, '_show_touches', False) + monkeypatch.setattr(desktop, 'gui_label', Mock()) + monkeypatch.setattr(terminal, 'gui_label', Mock()) + monkeypatch.setattr(terminal.Terminal, 'show_event', lambda self: None) + for name in ('draw_rectangle_rec', 'draw_rectangle_rounded', 'draw_rectangle_rounded_lines_ex', 'draw_line_ex', 'draw_ring', + 'draw_circle_v', 'draw_triangle', 'draw_ellipse', 'begin_scissor_mode', 'end_scissor_mode'): + monkeypatch.setattr(rl, name, lambda *a: None) + return ArabesqueLayout(Mock()) + + +def event(x, y, *, pressed=False, released=False, slot=0): + return MouseEvent(MousePos(x, y), slot, pressed, released, pressed, 1.0) + + +def frame(view, *events): + gui_app._mouse_events = list(events) + view.render() + + +def tap(view, rect): + x, y = rect.x + rect.width / 2, rect.y + rect.height / 2 + frame(view, event(x, y, pressed=True)) + frame(view, event(x, y, released=True)) + + +def test_menu_dismissal_tool_title_and_desktop_return(shell): + shell.set_rect(rl.Rectangle(0, 0, 2160, 1080)) + tap(shell, shell._menu_button) + assert shell._menu_open + tap(shell, shell._dock) + assert not shell._menu_open + assert shell._active_tool is None + tap(shell, shell._dock) + assert isinstance(shell._active_tool, terminal.Terminal) + assert not gui_app.mouse_events + frame(shell) + assert desktop.gui_label.call_args_list[-1].args[1] == '< Desktop' + assert desktop.gui_label.call_args_list[-2].args[1] == 'Terminal' + assert any(call.args[1] == 'Opening local shell…' for call in terminal.gui_label.call_args_list) + tap(shell, shell._back_button) + assert shell._active_tool is None + frame(shell) + assert desktop.gui_label.call_args_list[-2].args[1] == 'Arabesque' + shell._on_exit.assert_not_called() + tap(shell, shell._menu_button) + tap(shell, shell._exit_button) + shell._on_exit.assert_called_once() + + +def test_drag_stray_release_second_finger_and_hidden_press_do_not_activate(shell): + shell.set_rect(rl.Rectangle(0, 0, 2160, 1080)) + tap(shell, shell._menu_button) + frame(shell, event(900, 400, pressed=True), event(100, 170, released=True)) + assert not shell._menu_open + frame(shell, event(60, 50, pressed=True, slot=1), event(60, 50, released=True, slot=1)) + assert not shell._menu_open + frame(shell, event(60, 50, released=True)) + assert not shell._menu_open + tap(shell, shell._menu_button) + frame(shell, event(100, 170, pressed=True)) + shell.hide_event() + frame(shell, event(100, 170, released=True)) + shell._on_exit.assert_not_called() + + +@pytest.mark.parametrize('bounds', [(0, 0, 2160, 1080), (0, 0, 1920, 1080), (45, 30, 1280, 720), (0, 0, 2560, 1080)]) +def test_layout_anchors_and_touch_targets(shell, bounds): + shell.set_rect(rl.Rectangle(*bounds)) + rect, dock = shell.rect, shell._dock + assert dock.x + dock.width / 2 == pytest.approx(rect.x + rect.width / 2) + assert dock.y + dock.height < rect.y + rect.height + assert dock.width >= 44 and dock.height >= 44 + assert shell._dock_icon.x == pytest.approx(dock.x + dock.width / 2) + frame(shell) + caption = desktop.gui_label.call_args.args[0] + assert caption.x + caption.width / 2 == pytest.approx(shell._dock_icon.x) + assert caption.y >= shell._dock_icon.y + 64 * shell._scale + assert caption.y + caption.height < shell._dock_shelf.y + assert caption.width < dock.width + back_left, front_left, front_right, back_right = shell._dock_top + assert back_left.x > front_left.x and back_right.x < front_right.x + assert back_left.y < shell._dock_icon.y + 64 * shell._scale < front_left.y + assert front_left.y == pytest.approx(shell._dock_shelf.y) + assert front_right.y == pytest.approx(shell._dock_shelf.y) + assert shell._dock_shelf.y + shell._dock_shelf.height < dock.y + dock.height + assert shell._bar.width == rect.width + assert shell._exit_button.y > shell._bar.y + shell._bar.height + assert min(shell._menu_button.width, shell._menu_button.height) >= 44 + assert min(shell._back_button.width, shell._back_button.height) >= 44 + assert shell._tool_rect.y == shell._bar.y + shell._bar.height + assert shell._tool_rect.x == rect.x + assert shell._tool_rect.width == rect.width + assert shell._tool_rect.y + shell._tool_rect.height == pytest.approx(rect.y + rect.height) + tap(shell, shell._menu_button) + assert shell._menu_open + tap(shell, shell._dock) + tap(shell, shell._dock) + frame(shell) + assert shell._active_tool.rect == shell._tool_rect + label = desktop.gui_label.call_args + assert label.args[0] is shell._back_button + assert label.args[1] == '< Desktop' + assert label.kwargs['alignment'] == desktop.TextAlignment.CENTER + + +def test_tool_theme_override_does_not_change_desktop_or_defaults(shell, monkeypatch): + draw = Mock() + monkeypatch.setattr(rl, 'draw_rectangle_rec', draw) + custom = replace(DEFAULT_THEME, background=rl.Color(25, 31, 37, 255), text=rl.Color(240, 230, 220, 255), font_weight=FontWeight.BOLD) + assert terminal.Terminal()._theme is DEFAULT_THEME + shell.set_rect(rl.Rectangle(0, 0, 2160, 1080)) + shell._open_tool('Terminal', terminal.Terminal(theme=custom)) + frame(shell) + assert draw.call_args_list[1].args[1] is custom.background + assert any(call.args[3] is custom.text and call.args[4] is FontWeight.BOLD for call in terminal.gui_label.call_args_list) + assert desktop.gui_label.call_args_list[0].args[3] is DEFAULT_THEME.text + assert DEFAULT_THEME.background.r == 12 + + +@pytest.fixture +def probe(): + class Probe(Widget): + def __init__(self): + super().__init__() + self.show_event = Mock() + self.hide_event = Mock() + self._handle_mouse_press = Mock() + self._handle_mouse_release = Mock() + self._update_state = Mock() + self.render_calls = Mock() + + def _render(self, rect, /): + self.render_calls(rect) + return Probe() + + +def test_dock_is_hidden_for_tools_and_returns_on_desktop(shell, probe, monkeypatch): + triangles = Mock() + monkeypatch.setattr(rl, 'draw_triangle', triangles) + shell.set_rect(rl.Rectangle(0, 0, 2160, 1080)) + frame(shell) + assert any(call.args[0] is shell._dock_top[0] for call in triangles.call_args_list) + + shell._open_tool('Probe', probe) + triangles.reset_mock() + frame(shell) + assert not any(call.args[0] is shell._dock_top[0] for call in triangles.call_args_list) + assert probe.rect == shell._tool_rect + + # The bottom edge now belongs to the tool, including the former dock area. + x, y = shell.rect.x + shell.rect.width / 2, shell.rect.y + shell.rect.height - 1 + frame(shell, event(x, y, pressed=True)) + frame(shell, event(x, y, released=True)) + probe._handle_mouse_press.assert_called_once() + probe._handle_mouse_release.assert_called_once() + + tap(shell, shell._back_button) + triangles.reset_mock() + frame(shell) + assert shell._active_tool is None + assert any(call.args[0] is shell._dock_top[0] for call in triangles.call_args_list) + + +def test_tool_input_is_gated_for_menu_chrome_and_disabled_desktop(shell, probe): + shell.set_rect(rl.Rectangle(0, 0, 2160, 1080)) + shell._open_tool('Probe', probe) + tap(shell, shell._tool_rect) + probe._handle_mouse_press.assert_called_once() + probe._handle_mouse_release.assert_called_once() + probe._handle_mouse_press.reset_mock() + probe._handle_mouse_release.reset_mock() + + # A whole queued gesture after the menu tap must not reach content in the same frame. + frame(shell, event(60, 50, pressed=True), event(60, 50, released=True), + event(1000, 500, pressed=True), event(1000, 500, released=True)) + assert not probe.enabled + probe._handle_mouse_press.assert_not_called() + probe._handle_mouse_release.assert_not_called() + tap(shell, shell._menu_button) + assert shell._menu_open + tap(shell, shell._tool_rect) + assert not shell._menu_open + probe._handle_mouse_press.assert_not_called() + probe._handle_mouse_release.assert_not_called() + shell.set_enabled(False) + tap(shell, shell._tool_rect) + assert not probe.enabled + probe._handle_mouse_press.assert_not_called() + shell.set_enabled(True) + tap(shell, shell._tool_rect) + probe._handle_mouse_press.assert_called_once() + probe._handle_mouse_release.assert_called_once() + + +def test_tool_lifecycle_runs_once_and_closed_tool_is_not_rendered(shell, probe, monkeypatch): + shell.set_rect(rl.Rectangle(0, 0, 2160, 1080)) + create = Mock(return_value=probe) + monkeypatch.setattr(desktop, 'Terminal', create) + frame(shell) + create.assert_not_called() + tap(shell, shell._dock) + probe.show_event.assert_called_once() + tap(shell, shell._dock) + create.assert_called_once() + probe._handle_mouse_press.assert_called_once() + probe._handle_mouse_release.assert_called_once() + probe._handle_mouse_press.reset_mock() + probe._handle_mouse_release.reset_mock() + x, y = shell._back_button.x + 40, shell._back_button.y + 40 + frame(shell, event(x, y, pressed=True)) + frame(shell, event(x, y, released=True), event(1000, 500, pressed=True), event(1000, 500, released=True)) + assert not gui_app.mouse_events + probe._handle_mouse_press.assert_not_called() + probe._handle_mouse_release.assert_not_called() + probe.hide_event.assert_called_once() + probe.render_calls.reset_mock() + probe._update_state.reset_mock() + shell.hide_event() + for _ in range(3): + frame(shell) + probe.hide_event.assert_called_once() + probe.render_calls.assert_not_called() + probe._update_state.assert_not_called() + + +def test_closed_tool_instances_are_released(shell): + shell.set_rect(rl.Rectangle(0, 0, 2160, 1080)) + instances = [] + for _ in range(100): + tap(shell, shell._dock) + instances.append(weakref.ref(shell._active_tool)) + tap(shell, shell._back_button) + gc.collect() + assert all(instance() is None for instance in instances) + + +@pytest.fixture +def main(shell, monkeypatch): + ui = NS(started=False, is_body=False, add_on_body_changed_callbacks=Mock()) + ui.is_offroad = lambda: not ui.started + sidebar = Mock(is_visible=False) + sidebar.set_visible.side_effect = lambda visible: setattr(sidebar, 'is_visible', visible) + settings = Mock() + namespace = {'rl': rl, 'IntEnum': IntEnum, 'Widget': Widget, 'gui_app': gui_app, 'ui_state': ui, + 'ArabesqueLayout': ArabesqueLayout, 'Sidebar': lambda: sidebar, + 'SIDEBAR_WIDTH': 300, 'HomeLayout': Mock, 'BodyLayout': Mock, 'SettingsLayout': Mock(return_value=settings), + 'AugmentedRoadView': Mock, 'messaging': NS(PubMaster=Mock()), + 'device': NS(add_interactive_timeout_callback=Mock()), 'OnboardingWindow': lambda: NS(completed=True), + 'PanelType': NS(TOGGLES=2, FIREHOSE=4)} + native = layout_classes(ROOT / 'ui/layouts/main.py', {'MainState', 'MainLayout'}, namespace) + monkeypatch.setattr(gui_app, 'push_widget', lambda widget: None) + layout = native.MainLayout() + layout.set_rect(rl.Rectangle(0, 0, 2160, 1080)) + layout._set_current_layout(native.MainState.SETTINGS) + return NS(layout=layout, state=native.MainState, ui=ui, settings=settings, native=native) + + +def test_entry_wiring_and_offroad_gate(main): + m = main + callbacks = m.native.SettingsLayout.call_args.kwargs + assert callbacks['on_open_arabesque'] == m.layout._open_arabesque + m.ui.started = True + callbacks['on_open_arabesque']() + assert m.layout._current_mode == m.state.SETTINGS + m.ui.started = False + callbacks['on_open_arabesque']() + assert m.layout._current_mode == m.state.ARABESQUE + frame(m.layout) + m.settings.render.assert_not_called() + + +def test_handoff_discards_queued_touches_and_restores_settings(main, probe): + m = main + gui_app._mouse_events = [event(60, 50, pressed=True), event(60, 50, released=True)] + m.layout._open_arabesque() + assert not gui_app.mouse_events + view = m.layout._layouts[m.state.ARABESQUE] + frame(m.layout) + view._open_tool('Probe', probe) + tap(m.layout, view._menu_button) + frame(m.layout, event(100, 170, pressed=True)) + frame(m.layout, event(100, 170, released=True), event(1000, 500, pressed=True), event(1000, 500, released=True)) + assert m.layout._current_mode == m.state.SETTINGS + assert not gui_app.mouse_events + m.settings.render.assert_not_called() + assert not view._menu_open + assert view._active_tool is None + probe.hide_event.assert_called_once() + frame(m.layout) + m.settings.render.assert_called_once() + + +def test_onroad_transition_draws_normal_alert_owner_immediately(main, probe): + m = main + m.layout._open_arabesque() + view = m.layout._layouts[m.state.ARABESQUE] + frame(m.layout) + view._open_tool('Probe', probe) + tap(m.layout, view._menu_button) + m.ui.started = True + frame(m.layout, event(100, 170, pressed=True)) + assert m.layout._current_mode == m.state.ONROAD + assert not view._menu_open + assert view._active_tool is None + probe.hide_event.assert_called_once() + assert not gui_app.mouse_events + m.layout._layouts[m.state.ONROAD].render.assert_called_once() + m.ui.started = False + frame(m.layout) + assert m.layout._current_mode == m.state.HOME + + +def test_shutdown_closes_active_tool_once(main, probe): + main.layout._open_arabesque() + view = main.layout._layouts[main.state.ARABESQUE] + frame(main.layout) + view._open_tool('Probe', probe) + main.layout.close() + main.layout.close() + assert view._active_tool is None + probe.hide_event.assert_called_once() + + +def test_repeated_entry_exit_reuses_view_and_never_runs_it_closed(main): + m = main + view = m.layout._layouts[m.state.ARABESQUE] + for _ in range(100): + m.layout._open_arabesque() + frame(m.layout) + tap(m.layout, view._menu_button) + tap(m.layout, view._exit_button) + assert m.layout._current_mode == m.state.SETTINGS + assert m.layout._layouts[m.state.ARABESQUE] is view + view._render = Mock() + frame(m.layout) + view._render.assert_not_called() + + +def test_star_large_routes_shell_and_uses_confirmed_offroad(main, probe): + namespace = vars(main.native).copy() + namespace['Destination'] = NS(NETWORK='network') + namespace['ShellMode'] = NS(HOME='home', SETTINGS='settings', ONROAD='onroad') + star = layout_classes(ROOT.parent / 'starpilot/ui/runtime_app.py', {'StarMainLayout'}, namespace) + main.layout.__class__ = star.StarMainLayout + layout = main.layout + layout.star, layout.page = Mock(), Mock() + layout.star.selected = 'developer' + layout._large_destination = None + layout._network_bridge = Mock() + layout.star.adapter.confirmed_offroad.return_value = False + layout._open_arabesque() + assert layout._current_mode == main.state.SETTINGS + layout.star.adapter.confirmed_offroad.return_value = True + layout._open_arabesque() + frame(layout) + view = layout._layouts[main.state.ARABESQUE] + view._open_tool('Probe', probe) + layout.page.render.assert_not_called() + layout.star.adapter.confirmed_offroad.return_value = False + frame(layout) + assert layout._current_mode == main.state.HOME + layout.page.render.assert_called_once() + probe.hide_event.assert_called_once() + assert view._active_tool is None + layout.star.adapter.confirmed_offroad.return_value = True + layout._set_current_layout(main.state.SETTINGS) + layout._open_arabesque() + view._open_tool('Probe', probe) + layout._native_onroad = layout._layouts[main.state.ONROAD] + layout.close() + assert probe.hide_event.call_count == 2 + assert view._active_tool is None + layout.star.close.assert_called_once() + layout._native_onroad.close.assert_called_once() diff --git a/openpilot/selfdrive/ui/tests/test_arabesque_terminal.py b/openpilot/selfdrive/ui/tests/test_arabesque_terminal.py new file mode 100644 index 0000000000..fa20b9278f --- /dev/null +++ b/openpilot/selfdrive/ui/tests/test_arabesque_terminal.py @@ -0,0 +1,392 @@ +"""Terminal styling, real Widget input, and a private native shell session.""" +from collections import deque +import ctypes +import os +from pathlib import Path +import shutil +import sys +import time +from types import SimpleNamespace as NS +from unittest.mock import Mock, call + +import pyray as rl +import pytest + +from openpilot.selfdrive.ui.layouts.arabesque.tools import terminal +from openpilot.system.ui.lib.application import gui_app, MouseEvent, MousePos + + +def test_styles_preserve_positions_colors_reverse_and_hidden_text(): + rows = terminal._styled_rows(['a\x1b[31;44mred\x1b[7mR\x1b[27;8msecret\x1b[0m界e\u0301', + '\x1b[38;5;196;48;2;10;20;30mrgb\x1b[39;49mdefault']) + assert rows[0][1][0:4] == (1, 'red', terminal._PALETTE[1], terminal._PALETTE[4]) + assert rows[0][2][4] is True + assert rows[0][3][5] is True + assert rows[0][4][2:6] == (None, None, False, False) + assert rows[0][4][6] == 3 + assert rows[1][0][2:4] == ((255, 0, 0), (10, 20, 30)) + assert rows[1][1][2:4] == (None, None) + + +def test_screen_text_cannot_impersonate_a_control_reply(): + session = terminal._Session.__new__(terminal._Session) + session._block = None + session._pending = deque([('screen', 2), ('cursor', 1)]) + for line in (b'%begin 123 4 1', b'%end 123 4 1', 'café'.encode(), b'%end 123 4 1', + b'%begin 123 5 1', b'3 1 1', b'%end 123 5 1'): + session._line(line) + assert session.rows[0][0][1] == '%end 123 4 1' + assert session.rows[1][0][1] == 'café' + assert session.cursor == (3, 1, True) + assert not session._pending + + +@pytest.fixture +def tool(monkeypatch): + import openpilot.system.ui.widgets as widgets + monkeypatch.setattr(widgets, 'device', NS(awake=True)) + monkeypatch.setattr(gui_app, '_mouse_events', []) + monkeypatch.setattr(gui_app, '_render_prepare', {}) + monkeypatch.setattr(gui_app, '_show_touches', False) + monkeypatch.setattr(terminal, 'gui_label', Mock()) + monkeypatch.setattr(terminal, 'device', NS(_reset_interactive_timeout=Mock())) + monkeypatch.setattr(gui_app, 'font', Mock(return_value=NS(texture=NS(id=2), baseSize=16, glyphs=[NS(advanceX=8, value=63)]))) + monkeypatch.setattr(rl, 'get_font_default', lambda: NS(texture=NS(id=1))) + monkeypatch.setattr(rl, 'load_font_ex', Mock(return_value=NS(texture=NS(id=3), baseSize=64, glyphs=[NS(advanceX=29, value=63)]))) + monkeypatch.setattr(rl, 'gen_texture_mipmaps', Mock()) + monkeypatch.setattr(rl, '_orig_draw_text_ex', Mock(), raising=False) + monkeypatch.setattr(rl, 'get_glyph_index', lambda *a: 0) + for name in ('unload_font', 'set_texture_filter', 'draw_rectangle_rec', 'draw_rectangle_rounded', + 'draw_rectangle_lines_ex', 'rl_draw_render_batch_active', 'begin_scissor_mode', 'end_scissor_mode', 'draw_text_ex'): + monkeypatch.setattr(rl, name, Mock()) + monkeypatch.setattr(rl, 'set_exit_key', Mock()) + monkeypatch.setattr(rl, 'get_key_pressed', lambda: 0) + monkeypatch.setattr(rl, 'get_char_pressed', lambda: 0) + monkeypatch.setattr(rl, 'is_key_down', lambda key: False) + monkeypatch.setattr(rl, 'is_key_pressed_repeat', lambda key: False) + session = NS(rows=[], cursor=(0, 0, False), error='', update=Mock(), close=Mock(), resize=Mock(), keys=Mock()) + monkeypatch.setattr(terminal, '_Session', Mock(return_value=session)) + widget = terminal.Terminal() + widget.set_rect(rl.Rectangle(0, 104, 2160, 768)) + widget.show_event() + yield widget, session + widget.hide_event() + + +def touch(widget, key): + bounds = next(rect for rect, _, value in widget._buttons if value == key) + pos = MousePos(bounds.x + bounds.width / 2, bounds.y + bounds.height / 2) + for down in (True, False): + gui_app._mouse_events = [MouseEvent(pos, 0, down, not down, down, 1)] + widget.render() + gui_app._mouse_events = [] + + +def test_touch_keyboard_sends_bytes_and_control_keys_and_resizes(tool): + widget, session = tool + initial_rows = widget._size[1] + touch(widget, 'keyboard') + assert widget._keyboard and widget._size[1] < initial_rows + for key in ('l', 's', terminal.ENTER_KEY, 'ctrl', 'c', 'Up'): + touch(widget, key) + assert [call.args[0] for call in session.keys.call_args_list] == [b'l', b's', 'Enter', b'\x03', 'Up'] + assert not widget._ctrl + assert min(rect.height for rect, _, _ in widget._buttons) >= 44 + session.resize.assert_called_with(*widget._size) + + +def test_disabled_tool_drops_keyboard_input_without_sending_it(tool, monkeypatch): + widget, session = tool + widget.set_enabled(False) + keys = iter([rl.KEY_ENTER, 0]) + chars = iter([ord('x'), 0]) + monkeypatch.setattr(rl, 'get_key_pressed', lambda: next(keys, 0)) + monkeypatch.setattr(rl, 'get_char_pressed', lambda: next(chars, 0)) + widget._update_state() + session.keys.assert_not_called() + + +def test_physical_control_and_escape_are_terminal_input(tool, monkeypatch): + widget, session = tool + keys = iter([rl.KEY_C, rl.KEY_ESCAPE, 0]) + chars = iter([ord('c'), 0]) + monkeypatch.setattr(rl, 'get_key_pressed', lambda: next(keys, 0)) + monkeypatch.setattr(rl, 'get_char_pressed', lambda: next(chars, 0)) + monkeypatch.setattr(rl, 'is_key_down', lambda key: key == rl.KEY_LEFT_CONTROL) + widget._update_state() + assert [call.args[0] for call in session.keys.call_args_list] == [b'\x03', 'Escape'] + rl.set_exit_key.assert_called_with(0) + + +def test_prepare_caches_rows_and_hide_releases_only_owned_resources(tool): + widget, session = tool + owned = widget._font + lifecycle = Mock() + lifecycle.attach_mock(rl.rl_draw_render_batch_active, 'flush') + lifecycle.attach_mock(rl.unload_font, 'unload') + session.rows = terminal._styled_rows(['\x1b[7minverse']) + widget._prepare() + spans = widget._draw_rows + assert spans[0][0][2] is widget._theme.background + assert spans[0][0][3] is widget._theme.text + widget._prepare() + assert widget._draw_rows is spans + assert widget._prepare in gui_app._render_prepare + widget.hide_event() + widget.hide_event() + assert not gui_app._render_prepare + session.close.assert_called_once() + assert lifecycle.mock_calls == [call.flush(), call.unload(owned)] + rl.set_exit_key.assert_called_with(rl.KEY_ESCAPE) + + +def test_font_is_loaded_once_per_entry_and_not_during_render(tool): + widget, session = tool + args = rl.load_font_ex.call_args.args + assert Path(args[0]).name == 'JetBrainsMono-Medium.ttf' + assert args[1:] == (64, None, 0) + rl.set_texture_filter.assert_called_once_with(widget._font.texture, rl.TextureFilter.TEXTURE_FILTER_TRILINEAR) + session.rows = terminal._styled_rows(['local shell']) + widget.show_event() + for _ in range(3): + widget.render() + rl.load_font_ex.assert_called_once() + rl.gen_texture_mipmaps.assert_called_once() + assert len(gui_app._render_prepare) == 1 + + +def test_grid_matches_selected_c_preview_and_sizes_from_actual_glyphs(tool, monkeypatch): + widget, session = tool + # Calibrate the large-UI preview independently of the surrounding suite's UI mode. + monkeypatch.setattr(terminal, 'FONT_SCALE', 1.242) + widget.set_rect(rl.Rectangle(0, 104, 2160, 976)) + assert widget._size == (109, 18) + for font, size in ((widget._font, widget._font_size), (widget._fallback_font, widget._fallback_size)): + assert font.glyphs[0].advanceX * size / font.baseSize == pytest.approx(widget._cell_w) + session.resize.assert_called_with(109, 18) + + +def test_unicode_fallback_keeps_colors_columns_cursor_and_explicit_font(tool, monkeypatch): + widget, session = tool + shared, owned = widget._fallback_font, widget._font + shared.glyphs += [NS(advanceX=8, value=ord('ü')), NS(advanceX=16, value=ord('日'))] + monkeypatch.setattr(rl, 'get_glyph_index', lambda font, value: next((i for i, g in enumerate(font.glyphs) if g.value == value), 0)) + session.rows = terminal._styled_rows(['A\x1b[31;44mü日B\x1b[7mC']) + session.cursor = (6, 0, True) + widget._prepare() + spans = widget._draw_rows[0] + assert [(s[0], s[1], s[5], s[6]) for s in spans] == [ + (0, 'A', 1, owned), (1, 'ü日', 3, shared), (4, 'B', 1, owned), (5, 'C', 1, owned)] + assert spans[1][2:4] == spans[2][2:4] + assert (spans[3][2].r, spans[3][2].g, spans[3][2].b) == terminal._PALETTE[4] + assert (spans[3][3].r, spans[3][3].g, spans[3][3].b) == terminal._PALETTE[1] + widget._render(widget.rect) + draws = rl._orig_draw_text_ex.call_args_list + assert [draw.args[0] for draw in draws] == [owned, shared, owned, owned] + for draw, column in zip(draws, (0, 1, 4, 5), strict=True): + assert draw.args[2].x == pytest.approx(widget._screen.x + column * widget._cell_w) + assert draws[1].args[3] == widget._fallback_size + assert rl.draw_rectangle_lines_ex.call_args.args[0].x == pytest.approx(widget._screen.x + 6 * widget._cell_w) + rl.draw_text_ex.assert_not_called() + + +@pytest.mark.parametrize('failure', ['default', 'oserror']) +def test_font_load_failure_uses_shared_font_without_unloading_it(tool, failure): + widget, session = tool + widget.hide_event() + rl.unload_font.reset_mock() + rl.rl_draw_render_batch_active.reset_mock() + if failure == 'default': + rl.load_font_ex.return_value = rl.get_font_default() + else: + rl.load_font_ex.side_effect = OSError('font unavailable') + widget.show_event() + assert widget._font is widget._fallback_font + session.rows = terminal._styled_rows(['still usable']) + widget._prepare() + widget._render(widget.rect) + widget.hide_event() + rl.unload_font.assert_not_called() + rl.rl_draw_render_batch_active.assert_not_called() + + +@pytest.fixture +def real_session(tmp_path, monkeypatch): + if sys.platform != 'linux' or shutil.which('tmux') is None: + raise pytest.skip.Exception('requires the Linux OS tmux binary') + # Own/reap test servers even under a container PID 1; production uses the OS init. + assert ctypes.CDLL(None, use_errno=True).prctl(36, 1, 0, 0, 0) == 0 + monkeypatch.setenv('HOME', str(tmp_path)) + monkeypatch.setenv('HISTFILE', '/dev/null') + before = len(list(Path('/proc/self/fd').iterdir())) + children = Path(f'/proc/{os.getpid()}/task/{os.getpid()}/children') + before_children = set(children.read_text().split()) + session = terminal._Session(80, 24) + yield session + session.close() + deadline = time.monotonic() + .5 + while time.monotonic() < deadline: + try: + if not os.waitpid(-1, os.WNOHANG)[0]: + time.sleep(.005) + except ChildProcessError: + break + assert len(list(Path('/proc/self/fd').iterdir())) == before + assert not set(children.read_text().split()) - before_children + + +def wait_screen(session, predicate): + deadline = time.monotonic() + 3 + while time.monotonic() < deadline: + session.update() + assert not session.error, session.error + text = [''.join(span[1] for span in row).rstrip() for row in session.rows] + if predicate(text): + return text + time.sleep(.003) + raise AssertionError('shell did not produce expected output') + + +def test_real_shell_state_unicode_resize_interrupt_and_exit(real_session): + session = real_session + wait_screen(session, lambda rows: bool(rows)) + session.keys("cd /tmp; printf 'READY_CAFE:café\\n'\r".encode()) + wait_screen(session, lambda rows: 'READY_CAFE:café' in rows) + session.resize(100, 30) + wait_screen(session, lambda rows: len(rows) == 30) + session.keys(b"pwd; stty size\r") + wait_screen(session, lambda rows: '/tmp' in rows and '30 100' in rows) + session.keys(b'sleep 30\r') + for _ in range(15): + session.update() + time.sleep(.003) + session.keys(b'\x03') + session.keys(b"printf 'INTERRUPTED\\n'\r") + wait_screen(session, lambda rows: 'INTERRUPTED' in rows) + session.keys(b'exit\r') + deadline = time.monotonic() + 2 + while session._process is not None and time.monotonic() < deadline: + session.update() + time.sleep(.003) + assert session._process is None and 'exited' in session.error + + +def test_real_output_flood_and_atomic_input_limit(real_session): + session = real_session + wait_screen(session, lambda rows: bool(rows)) + session.keys(b'yes flood\r') + for _ in range(50): + session.update() + assert len(session._buffer) <= 262144 and len(session._pending) <= 64 + time.sleep(.003) + session.keys(b'\x03') + session.keys(b"printf 'FLOOD_STOPPED\\n'\r") + wait_screen(session, lambda rows: 'FLOOD_STOPPED' in rows) + session.keys(b'x' * 2049) + assert session._process is None and 'input limit' in session.error + session.update() + + +def test_exported_links_and_legacy_borders_do_not_occupy_text_cells(): + rows = terminal._styled_rows(['\x1b]8;;https://example.org\x1b\\link\x1b]8;;\x1b\\ \x0elqqk\x0f']) + assert ''.join(span[1] for span in rows[0]) == 'link ┌──┐' + assert sum(span[6] for span in rows[0]) == 9 + + +def test_prepare_cannot_flush_commands_before_main_checks_offroad(tool): + widget, session = tool + widget._prepare() + session.update.assert_not_called() + widget._update_state() + session.update.assert_called_once() + + +def test_shared_font_fallback_preserves_columns(tool): + widget, session = tool + session.rows = terminal._styled_rows(['─界e\u0301']) + widget._prepare() + assert widget._draw_rows[0][0][1] == '-? ?' + assert widget._draw_rows[0][0][5] == 4 + + +def test_underline_color_does_not_change_foreground(): + spans = terminal._styled_rows(['\x1b[58:2::1:2:31mtext'])[0] + assert spans[0][2:4] == (None, None) + + +@pytest.mark.parametrize('key,modifier,expected', [(rl.KEY_LEFT, rl.KEY_LEFT_CONTROL, 'C-Left'), + (rl.KEY_TAB, rl.KEY_LEFT_SHIFT, 'BTab')]) +def test_physical_navigation_modifiers(tool, monkeypatch, key, modifier, expected): + widget, session = tool + keys = iter([key]) + monkeypatch.setattr(rl, 'get_key_pressed', lambda: next(keys, 0)) + monkeypatch.setattr(rl, 'is_key_down', lambda value: value == modifier) + widget._update_state() + session.keys.assert_called_once_with(expected) + + +def test_text_and_enter_in_one_frame_keep_their_order(tool, monkeypatch): + widget, session = tool + keys = iter([rl.KEY_L, rl.KEY_S, rl.KEY_ENTER]) + chars = iter([ord('l'), ord('s')]) + monkeypatch.setattr(rl, 'get_key_pressed', lambda: next(keys, 0)) + monkeypatch.setattr(rl, 'get_char_pressed', lambda: next(chars, 0)) + widget._update_state() + assert [call.args[0] for call in session.keys.call_args_list] == [b'l', b's', 'Enter'] + + +def test_cleanup_waits_are_bounded_even_after_forced_termination(monkeypatch): + session = terminal._Session.__new__(terminal._Session) + process = Mock() + process.wait.side_effect = terminal.subprocess.TimeoutExpired('tmux', .02) + session._process, session._folder = process, Mock() + session._out, session._buffer, session._pending = bytearray(), bytearray(), deque() + monkeypatch.setattr(os, 'write', Mock()) + session.close() + process.kill.assert_called_once() + assert [call.kwargs for call in process.wait.call_args_list] == [{'timeout': .02}, {'timeout': .02}] + + +def test_repeated_character_before_enter_is_not_sent_after_enter(tool, monkeypatch): + widget, session = tool + keys = iter([rl.KEY_ENTER]) + chars = iter([ord('s')]) + monkeypatch.setattr(rl, 'get_key_pressed', lambda: next(keys, 0)) + monkeypatch.setattr(rl, 'get_char_pressed', lambda: next(chars, 0)) + widget._update_state() + assert [call.args[0] for call in session.keys.call_args_list] == [b's', 'Enter'] + + +@pytest.mark.parametrize('failure', ['process', 'directory']) +def test_startup_failure_is_visible_and_can_be_closed(monkeypatch, failure): + target = terminal.subprocess if failure == 'process' else terminal.tempfile + monkeypatch.setattr(target, 'Popen' if failure == 'process' else 'TemporaryDirectory', Mock(side_effect=OSError('unavailable'))) + session = terminal._Session(80, 24) + assert session.error and session._process is None + if session._folder: + assert not Path(session._folder.name).exists() + session.update() + session.close() + + +def test_real_client_failure_cleans_up_private_server(real_session): + session = real_session + wait_screen(session, lambda rows: bool(rows)) + session._process.terminate() + deadline = time.monotonic() + 2 + while session._process is not None and time.monotonic() < deadline: + session.update() + time.sleep(.003) + assert session._process is None and session.error + + +def test_real_startup_can_be_cancelled_immediately(real_session): + real_session.close() + assert real_session._process is None + + +def test_keyboard_activity_resets_the_existing_idle_timer(tool): + widget, session = tool + widget._send('x') + terminal.device._reset_interactive_timeout.assert_called_once() + session.keys.assert_called_once_with(b'x') diff --git a/openpilot/selfdrive/ui/ui.py b/openpilot/selfdrive/ui/ui.py index 0132f04c35..91ddcfb197 100755 --- a/openpilot/selfdrive/ui/ui.py +++ b/openpilot/selfdrive/ui/ui.py @@ -94,6 +94,8 @@ def main(): controllers.close() finally: layout.close() + elif BIG_UI: + layout.close() if __name__ == "__main__": diff --git a/openpilot/starpilot/ui/runtime_app.py b/openpilot/starpilot/ui/runtime_app.py index d18aa92e25..c8f9e15ba8 100644 --- a/openpilot/starpilot/ui/runtime_app.py +++ b/openpilot/starpilot/ui/runtime_app.py @@ -1791,6 +1791,9 @@ class StarMainLayout(MainLayout): # the native layout's own gate immediately before invoking an effect owner. return bool(ui_state.is_offroad() and self.star.adapter.confirmed_offroad()) + def _arabesque_allowed(self) -> bool: + return self._confirmed_offroad() + def _system_power_action(self, index: int) -> None: from openpilot.selfdrive.ui.layouts.settings.settings import PanelType def allowed() -> bool: @@ -1916,7 +1919,7 @@ class StarMainLayout(MainLayout): return True def _render_main_content(self) -> None: - if self._current_mode == MainState.HOME and ui_state.is_body: + if self._current_mode == MainState.ARABESQUE or self._current_mode == MainState.HOME and ui_state.is_body: super()._render_main_content() return rect = self._rect @@ -1948,7 +1951,7 @@ class StarMainLayout(MainLayout): if changed and hasattr(self, "page"): self.page.hide_event() super()._set_current_layout(layout) - if changed and hasattr(self, "page"): + if changed and hasattr(self, "page") and layout != MainState.ARABESQUE: self.page.show_event() def open_settings(self, panel_type: Any) -> None: @@ -1973,6 +1976,7 @@ class StarMainLayout(MainLayout): super()._on_body_changed() def close(self) -> None: + super().close() self._leave_large_panel() self._network_bridge.leave() self.star.close() diff --git a/openpilot/starpilot/ui/runtime_snapshot.py b/openpilot/starpilot/ui/runtime_snapshot.py index 10f74064a8..399893c02d 100644 --- a/openpilot/starpilot/ui/runtime_snapshot.py +++ b/openpilot/starpilot/ui/runtime_snapshot.py @@ -24,7 +24,7 @@ from openpilot.starpilot.conditional_mode.projection import paired_clocks_ns from openpilot.starpilot.conditional_mode.runtime_settings import ConditionalSettingsOwner from openpilot.starpilot.conditional_mode.status import settings_fingerprint from openpilot.starpilot.longitudinal.profile_runtime import read_traffic_settings -from openpilot.starpilot.parked_evidence import ParkedEvidence, RESUME_SKEW_NS, fresh_offroad +from openpilot.starpilot.parked_evidence import ParkedEvidence, RESUME_SKEW_NS, fresh_offroad, host_parked from openpilot.starpilot.ui.conditional_status import ConditionalDisplayProjector, ConditionalPerceptionProjector, configured_choice from openpilot.starpilot.ui.traffic_status import TrafficDisplayProjector from openpilot.starpilot.controllers.mode_actions import SwitchbackStatusOwner @@ -324,6 +324,8 @@ class RuntimeSnapshotAdapter: self._replay_epoch: int | None = None def _parked(self, ui: Any, pair: tuple[int, int, int] | None) -> bool: + if host_parked(): + return not ui.started if pair is None: self._device_offset_ns = None self._pair_failed = True diff --git a/openpilot/starpilot/ui/tests/test_runtime_snapshot.py b/openpilot/starpilot/ui/tests/test_runtime_snapshot.py index 59728f10d1..e6b6388cdd 100644 --- a/openpilot/starpilot/ui/tests/test_runtime_snapshot.py +++ b/openpilot/starpilot/ui/tests/test_runtime_snapshot.py @@ -989,5 +989,25 @@ class TestParkedClockDomains(unittest.TestCase): self.assertTrue(self.parked()) +class TestHostOffroad(unittest.TestCase): + def test_only_private_parked_host_ui_has_authority_without_publishers(self): + env = {'SP_HOST_PARKED': '1', 'SP_HOST_RUNTIME': '1', 'OPENPILOT_PREFIX': 'starpilot-dev-test', 'SP_HOST_PREFIX': 'starpilot-dev-test'} + for overrides, expected in (({}, True), ({'SP_HOST_PARKED': '0'}, False), ({'SP_HOST_RUNTIME': '0'}, False), + ({'SP_HOST_PREFIX': 'other'}, False), ({'OPENPILOT_PREFIX': ''}, False)): + with self.subTest(overrides=overrides), patch.dict('os.environ', {**env, **overrides}, clear=True): + ui = ui_fake() + ui.started = False + ui.sm = SubMasterFake() + adapter = RuntimeSnapshotAdapter(ui) + self.assertEqual(adapter.confirmed_offroad(), expected) + self.assertEqual(adapter.build(ShellMode.HOME).device.offroad, expected) + ui.started = True + self.assertFalse(adapter.confirmed_offroad()) + ui.started = False + ui.replay_clock = object() + self.assertFalse(adapter.confirmed_offroad()) + self.assertFalse(adapter.build(ShellMode.HOME).device.offroad) + + if __name__ == "__main__": unittest.main() diff --git a/openpilot/starpilot/ui/tests/test_startup.py b/openpilot/starpilot/ui/tests/test_startup.py index cc4e387e80..163807618d 100644 --- a/openpilot/starpilot/ui/tests/test_startup.py +++ b/openpilot/starpilot/ui/tests/test_startup.py @@ -51,11 +51,12 @@ class TestStartup(unittest.TestCase): events.append("window") with patch.object(ui, "select_ui", side_effect=selection), patch.object(ui.gui_app, "init_window", side_effect=window), \ - patch.object(ui.gui_app, "render", return_value=iter(())), patch.object(ui, "MainLayout", return_value=object()), \ + patch.object(ui.gui_app, "render", return_value=iter(())), patch.object(ui, "MainLayout") as layout, \ patch.object(ui, "MiciMainLayout", return_value=object()), patch.object(ui.messaging, "PubMaster"), \ patch.object(ui, "config_realtime_process"), patch.object(ui, "BIG_UI", True): ui.main() self.assertEqual(events, ["select:large", "window"]) + layout.return_value.close.assert_called_once_with() with patch.object(ui, "select_ui", side_effect=RuntimeError("bad reviewed font")), \ patch.object(ui.gui_app, "init_window") as initialize, patch.object(ui, "config_realtime_process"): @@ -72,7 +73,7 @@ class TestStartup(unittest.TestCase): patch.object(ui, "select_ui", return_value=startup.UiSelection(False, "fixture")), \ patch.object(ui.gui_app, "init_window"), \ patch.object(ui.gui_app, "render", return_value=iter(((False, 0., 0.), (True, .01, .005), (True, .01, .005)))), \ - patch.object(ui, "MainLayout", return_value=object()), patch.object(ui, "MiciMainLayout", return_value=object()), \ + patch.object(ui, "MainLayout"), patch.object(ui, "MiciMainLayout", return_value=object()), \ patch.object(ui.messaging, "PubMaster"), patch.object(ui, "config_realtime_process") as realtime, \ patch.object(ui, "drop_realtime") as normal, patch.object(ui.os, "setpriority") as nice, \ patch.object(ui.os, "sched_getscheduler", return_value=0, create=True), \