mirror of
https://github.com/firestar5683/StarPilot.git
synced 2026-10-28 00:53:45 +08:00
Arabesque
This commit is contained in:
firestarsdog
@@ -0,0 +1 @@
|
||||
from openpilot.selfdrive.ui.layouts.arabesque.desktop import ArabesqueLayout as ArabesqueLayout
|
||||
@@ -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)
|
||||
@@ -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),
|
||||
)
|
||||
@@ -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)
|
||||
@@ -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)
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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()
|
||||
@@ -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')
|
||||
@@ -94,6 +94,8 @@ def main():
|
||||
controllers.close()
|
||||
finally:
|
||||
layout.close()
|
||||
elif BIG_UI:
|
||||
layout.close()
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
|
||||
@@ -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()
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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()
|
||||
|
||||
@@ -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), \
|
||||
|
||||
Reference in New Issue
Block a user