Cycle shared personality widget on tap

This commit is contained in:
firestarsdog
2026-10-30 22:26:44 -04:00
parent 70b705782c
commit b9744f1cab
6 changed files with 268 additions and 31 deletions
@@ -13,8 +13,10 @@ from openpilot.starpilot.controllers.tests.test_wheel_actions import Publisher,
from openpilot.starpilot.controllers.wheel_actions import WheelPublisher, WheelConsumer
from openpilot.starpilot.favorites.actions import BOOKMARK, SET_SPEED, TRAFFIC, FORCE_COAST
from openpilot.starpilot.favorites.owner import FavoritesOwner, default_slots
from openpilot.starpilot.ui.onroad_state import OnroadState, SpeedLimitObservation
from openpilot.starpilot.ui.shell import ShellMode
from openpilot.starpilot.ui.onroad_state import OnroadInput, OnroadState, SpeedLimitObservation, personality_bounds
from openpilot.starpilot.ui.presentation import Profile
from openpilot.starpilot.ui.settings_state import Destination
from openpilot.starpilot.ui.shell import ShellMode, ShellRequest
@pytest.fixture
@@ -269,7 +271,8 @@ def test_personality_cache_rejects_unusable_selfdrive_state(personality_session,
@pytest.mark.parametrize('profile', ['c4', 'c3'])
def test_rapid_personality_cycles_accept_own_notice_and_reject_safety_alert(personality_session, monkeypatch, profile):
@pytest.mark.parametrize('input_kind', ['wheel', 'tap'])
def test_rapid_personality_cycles_accept_own_notice_and_reject_safety_alert(personality_session, monkeypatch, profile, input_kind):
import pyray as rl
from dataclasses import replace
from openpilot.selfdrive.ui.layouts.settings import common
@@ -279,7 +282,6 @@ def test_rapid_personality_cycles_accept_own_notice_and_reject_safety_alert(pers
from openpilot.starpilot.controllers.tests.test_wheel_actions import state
from openpilot.starpilot.favorites.actions import CYCLE_PERSONALITY, EXPERIMENTAL
from openpilot.starpilot.ui.onroad_state import AlertSize, OnroadAlert
from openpilot.starpilot.ui.presentation import Profile
from openpilot.system.ui.lib.application import gui_app
runtime, session, ui, _, clock = personality_session
@@ -292,6 +294,7 @@ def test_rapid_personality_cycles_accept_own_notice_and_reject_safety_alert(pers
panel = compact.TogglesLayoutMici() if profile == 'c4' else large.TogglesLayout()
panel.set_visible(False)
session.profile = Profile.COMPACT if profile == 'c4' else Profile.LARGE
session.selected = Destination.STAR
del vars(session)['_favorite_authority'] # Exercise the real runtime admission.
shown = [OnroadState(True, True, 15, 80, SpeedLimitObservation(),
alert=OnroadAlert(AlertSize.MID, 'Standard', 'Driving Personality',
@@ -299,6 +302,7 @@ def test_rapid_personality_cycles_accept_own_notice_and_reject_safety_alert(pers
vars(session)['snapshot'] = lambda *_: NS(onroad=shown[0])
session._native_favorite_actions = lambda: session.native_favorite_actions(
lambda: True, panel.request_personality, lambda: True)
touch = OnroadInput(lambda action: session._emit(ShellRequest('onroad', action)), session.profile)
actions = session._native_favorite_actions()
assert actions[CYCLE_PERSONALITY].available and actions[CYCLE_PERSONALITY].invoke()
assert ui.params.get('LongitudinalPersonality') == 2
@@ -313,13 +317,24 @@ def test_rapid_personality_cycles_accept_own_notice_and_reject_safety_alert(pers
clock[0] = NOW + sequence * 5_000_000
for name in ui.sm:
ui.sm.logMonoTime[name], ui.sm.recv_time[name] = clock[0], clock[0] / 1e9
source.observe(ui.params, ui.CP, state(True), now_ns=clock[0] - 1_000_000, drive_id=DRIVE)
commands = source.observe(ui.params, ui.CP, state(False), now_ns=clock[0], drive_id=DRIVE)
assert commands == (('DistanceButtonControl', 1),)
source.publish(commands, ui.CP, publisher, now_ns=clock[0], drive_id=DRIVE,
source_car_ns=clock[0], source_control_ns=clock[0])
queue.append(messaging.log_from_bytes(publisher.events[-1][1]))
session._poll_wheel(clock[0])
if input_kind == 'wheel':
source.observe(ui.params, ui.CP, state(True), now_ns=clock[0] - 1_000_000, drive_id=DRIVE)
commands = source.observe(ui.params, ui.CP, state(False), now_ns=clock[0], drive_id=DRIVE)
assert commands == (('DistanceButtonControl', 1),)
source.publish(commands, ui.CP, publisher, now_ns=clock[0], drive_id=DRIVE,
source_car_ns=clock[0], source_control_ns=clock[0])
queue.append(messaging.log_from_bytes(publisher.events[-1][1]))
session._poll_wheel(clock[0])
else:
bounds = personality_bounds(session.profile, shown[0])
assert bounds is not None
left, top, right, bottom = bounds
point = (left + right) / 2, (top + bottom) / 2
before = ui.params.get('LongitudinalPersonality')
touch.press(*point, shown[0])
assert ui.params.get('LongitudinalPersonality') == before
touch.release(*point, shown[0])
assert session._request_emitted
assert ui.params.get('LongitudinalPersonality') == expected
assert ui.personality == 1 # No selfdriveState acknowledgement between presses.
@@ -339,3 +354,35 @@ def test_rapid_personality_cycles_accept_own_notice_and_reject_safety_alert(pers
path.write_bytes(b'bad')
assert not actions[CYCLE_PERSONALITY].invoke()
assert path.read_bytes() == b'bad'
@pytest.mark.parametrize('profile', list(Profile))
@pytest.mark.parametrize('blocked', ['safe_mode', 'no_longitudinal', 'no_car', 'replay', 'offroad', 'invalid_saved'])
def test_personality_tap_rechecks_wheel_authority_on_release(native_session, profile, blocked):
_, session, ui, _, calls = native_session
session.profile, session.selected = profile, Destination.STAR
del vars(session)['_favorite_authority']
ui.params.put('LongitudinalPersonality', 1, block=True)
state = session.snapshot(ShellMode.ONROAD).onroad
touch = OnroadInput(lambda action: session._emit(ShellRequest('onroad', action)), profile)
bounds = personality_bounds(profile, state)
assert bounds is not None
left, top, right, bottom = bounds
point = (left + right) / 2, (top + bottom) / 2
touch.press(*point, state)
if blocked == 'safe_mode':
ui.params.put_bool('SafeMode', True, block=True)
elif blocked == 'no_longitudinal':
ui.has_longitudinal_control = False
elif blocked == 'no_car':
ui.CP = None
elif blocked == 'replay':
ui.replay_clock = object()
elif blocked == 'offroad':
session._mode = ShellMode.HOME
else:
Path(ui.params.get_param_path('LongitudinalPersonality')).write_bytes(b'bad')
touch.release(*point, state)
assert session._request_emitted and calls == []
path = Path(ui.params.get_param_path('LongitudinalPersonality'))
assert path.read_bytes() == (b'bad' if blocked == 'invalid_saved' else b'1')
+7 -16
View File
@@ -3,9 +3,7 @@
import pyray as rl
from openpilot.common.filter_simple import FirstOrderFilter
from openpilot.starpilot.ui.appearance_preferences import CameraViewChoice
from openpilot.starpilot.ui.onroad_customization import placement, widget_size
from openpilot.starpilot.ui.onroad_state import AlertSize, OnroadState, is_lane_change_notice
from openpilot.starpilot.ui.onroad_state import AlertSize, OnroadState, is_lane_change_notice, personality_bounds
from openpilot.starpilot.ui.onroad_widget_style import draw_widget_frame
from openpilot.starpilot.ui.presentation import BitmapFonts, FontRole, Profile
from openpilot.system.ui.lib.application import gui_app
@@ -41,21 +39,13 @@ class PersonalityWidget:
def render(self, state: OnroadState) -> None:
profile = self.fonts.profile if self.fonts is not None else Profile.COMPACT
position = placement(state.customization, profile, 'following_distance')
bounds = personality_bounds(profile, state)
routine = state.alert.size == AlertSize.NONE or is_lane_change_notice(state.alert)
if (not position['enabled'] or state.appearance.camera_view == CameraViewChoice.DRIVER or state.reverse_driver_camera or
profile == Profile.COMPACT and position['x'] < 476 and not routine):
if bounds is None or profile == Profile.COMPACT and bounds[0] < 476 and not routine:
return
width, height = widget_size(state.customization, profile, 'following_distance')
rect = rl.Rectangle(position['x'], position['y'], width, height)
if profile == Profile.LARGE and state.right_hand_drive:
from openpilot.starpilot.ui.onroad_customization import PROFILES
from openpilot.starpilot.ui.onroad_dm import monitor_bounds
default = PROFILES['large']['widgets']['following_distance']['default']
monitor = placement(state.customization, profile, 'driver_monitor')
monitor_default = PROFILES['large']['widgets']['driver_monitor']['default']
if all(position[axis] == default[axis] and monitor[axis] == monitor_default[axis] for axis in ('x', 'y')):
rect.x = monitor_bounds(state.customization, profile, True).x - 32 - width
left, top, right, bottom = bounds
width, height = right - left, bottom - top
rect = rl.Rectangle(left, top, width, height)
draw_widget_frame(rect, state.customization, profile, 'following_distance')
scale = width / 60
label = ({'active': 'TRF', 'paused': 'PAUSE', 'off': 'OFF'}.get(state.traffic_display.state)
@@ -74,6 +64,7 @@ class PersonalityWidget:
def draw(self, rect: rl.Rectangle, *, personality: int, has_lead: bool, longitudinal_active: bool,
traffic_mode: bool = False, gas_override: bool = False, visible: bool = True, fit: bool = True) -> None:
self._prepare()
assert self._lead_car is not None
alpha = self._alpha_filter.update(visible)
engaged_lead = has_lead and longitudinal_active
white_target = 0.0 if engaged_lead else (1.0 if has_lead else 0.35)
+41 -2
View File
@@ -4,7 +4,7 @@ The producer owns freshness and units. A missing observation is never inferred
from a zero-valued speed or a default-initialized message.
"""
from openpilot.starpilot.ui.onroad_customization import default_document, offset, placement, widget_size
from openpilot.starpilot.ui.onroad_customization import PROFILES, default_document, offset, placement, widget_size
from collections.abc import Callable
from dataclasses import dataclass, field, replace
@@ -331,6 +331,23 @@ def speed_limit_bounds(profile: Profile, state: OnroadState) -> tuple[float, flo
return x + dx - 12, y + dy - 12, x + dx + width + 12, y + dy + height + 8
def personality_bounds(profile: Profile, state: OnroadState) -> tuple[float, float, float, float] | None:
position = placement(state.customization, profile, 'following_distance')
if not position['enabled'] or state.appearance.camera_view == CameraViewChoice.DRIVER or state.reverse_driver_camera:
return None
width, height = widget_size(state.customization, profile, 'following_distance')
x, y = position['x'], position['y']
if profile == Profile.LARGE and state.right_hand_drive:
widgets = PROFILES['large']['widgets']
monitor = placement(state.customization, profile, 'driver_monitor')
if all(position[axis] == widgets['following_distance']['default'][axis] and
monitor[axis] == widgets['driver_monitor']['default'][axis] for axis in ('x', 'y')):
bounds = PROFILES['large']['bounds']
monitor_width, _ = widget_size(state.customization, profile, 'driver_monitor')
x = 2 * bounds['x'] + bounds['width'] - monitor['x'] - monitor_width - 32 - width
return x, y, x + width, y + height
class OnroadInput:
"""The onroad wheel emits intent only; the supplied state must be refreshed."""
@@ -341,6 +358,7 @@ class OnroadInput:
self._press: tuple[float, float, bool, str] | None = None
self._slc_press: SlcControl | None = None
self._sign_press: tuple[tuple, tuple] | None = None
self._personality_press: tuple[float, float, int | None, tuple] | None = None
self._prompt: tuple | None = None
def _pending_context(self, state: OnroadState) -> tuple | None:
@@ -349,6 +367,13 @@ class OnroadInput:
return obs.session_id, obs.decision_id, obs.presentation_id, obs.pending_speed_limit_mps
return None
def _personality_target(self, state: OnroadState) -> tuple | None:
if (not state.camera_available or state.quick_select_open or
state.alert.size != AlertSize.NONE and not is_personality_notice(state.alert) or
self.profile == Profile.COMPACT and state.reversing and state.appearance.camera_view != CameraViewChoice.NONE):
return None
return personality_bounds(self.profile, state)
def present(self, state: OnroadState) -> OnroadState:
if self._prompt != self._pending_context(state):
self._prompt = None
@@ -360,7 +385,7 @@ class OnroadInput:
@property
def claimed(self) -> bool:
return self._press is not None or self._slc_press is not None or self._sign_press is not None
return self._press is not None or self._slc_press is not None or self._sign_press is not None or self._personality_press is not None
def press(self, x: float, y: float, state: OnroadState) -> None:
self.cancel()
@@ -375,6 +400,10 @@ class OnroadInput:
if bounds is not None and context is not None and bounds[0] <= x <= bounds[2] and bounds[1] <= y <= bounds[3]:
self._sign_press = context, bounds
return
bounds = self._personality_target(state)
if bounds is not None and bounds[0] <= x <= bounds[2] and bounds[1] <= y <= bounds[3]:
self._personality_press = x, y, state.drive_frame, bounds
return
# Only the large-profile wheel is an experimental-mode touch target.
if self.profile != Profile.LARGE:
self.cancel()
@@ -384,6 +413,13 @@ class OnroadInput:
def move(self, x: float, y: float, state: OnroadState) -> None:
state = self.present(state)
if self._personality_press:
px, py, drive_frame, original_bounds = self._personality_press
bounds = self._personality_target(state)
if (abs(x - px) > self.slop or abs(y - py) > self.slop or drive_frame != state.drive_frame or bounds != original_bounds or
bounds is None or not (bounds[0] <= x <= bounds[2] and bounds[1] <= y <= bounds[3])):
self.cancel()
return
if self._sign_press:
context, original_bounds = self._sign_press
bounds = speed_limit_bounds(self.profile, state)
@@ -410,6 +446,8 @@ class OnroadInput:
self._prompt = None
if self._sign_press:
self._prompt = self._sign_press[0]
if self._personality_press:
self.emit(OnroadRequest('cycle_personality'))
if self._press:
self.emit(OnroadRequest("set_experimental", not self._press[2], self._press[3]))
self.cancel()
@@ -418,6 +456,7 @@ class OnroadInput:
self._press = None
self._slc_press = None
self._sign_press = None
self._personality_press = None
if close_prompt:
self._prompt = None
+5
View File
@@ -1455,6 +1455,11 @@ class StarShellSession:
self._request_emitted = True
previous_destination = self.selected
action = request.action
if request.source == 'onroad' and getattr(action, 'kind', None) == 'cycle_personality':
self._native_favorite_actions()
if self._wheel_personality is not None:
self._wheel_personality()
return
if request.source == "onroad" and getattr(action, "kind", None) == "set_experimental":
binding = self._native_favorite_actions().get(EXPERIMENTAL)
if not (binding is not None and binding.available and binding.invoke is not None and
@@ -1,5 +1,6 @@
from dataclasses import replace
from types import SimpleNamespace as NS
from typing import cast
from unittest.mock import Mock
import pyray as rl
@@ -9,7 +10,7 @@ from openpilot.selfdrive.ui.mici.onroad import long_indicator as native
from openpilot.starpilot.ui import onroad_personality as indicator
from openpilot.starpilot.ui.onroad_customization import default_document
from openpilot.starpilot.ui.onroad_state import AlertSize, OnroadAlert, OnroadState, SpeedLimitObservation
from openpilot.starpilot.ui.presentation import Profile
from openpilot.starpilot.ui.presentation import BitmapFonts, Profile
from openpilot.starpilot.ui.traffic_status import TrafficDisplay
@@ -22,7 +23,7 @@ def road(**changes):
def display(monkeypatch):
monkeypatch.setattr(indicator.gui_app, 'texture', lambda path: NS(path=path, width=200, height=200))
monkeypatch.setattr(rl, 'get_time', lambda: 1.)
fonts = NS(profile=Profile.COMPACT, measure=Mock(return_value=NS(width=20)), draw=Mock())
fonts = cast(BitmapFonts, NS(profile=Profile.COMPACT, measure=Mock(return_value=NS(width=20)), draw=Mock()))
widget = indicator.PersonalityWidget(fonts)
widget._draw_centered = Mock()
return widget
@@ -0,0 +1,154 @@
from dataclasses import replace
from types import SimpleNamespace as NS
from typing import cast
from unittest.mock import Mock
import pytest
from openpilot.starpilot.ui.appearance_preferences import CameraViewChoice
from openpilot.starpilot.ui.onroad_customization import widget_size
from openpilot.starpilot.ui.onroad_personality import PersonalityWidget
from openpilot.starpilot.ui.onroad_state import (
AlertSize, ObservationKind, OnroadAlert, OnroadInput, OnroadRequest, OnroadState, SlcUiRequest, SpeedLimitObservation,
personality_bounds, slc_controls, speed_limit_bounds,
)
from openpilot.starpilot.ui.presentation import BitmapFonts, Profile
from openpilot.starpilot.ui.runtime_app import StarShellSession
from openpilot.starpilot.ui.settings_state import Destination
from openpilot.starpilot.ui.shell import ShellInput, ShellMode
@pytest.fixture(params=list(Profile))
def gesture(request):
requests = []
profile = request.param
state = OnroadState(True, True, 15, 80, SpeedLimitObservation(), longitudinal_active=True, drive_frame=1)
return profile, OnroadInput(requests.append, profile), requests, state
def center(bounds):
left, top, right, bottom = bounds
return (left + right) / 2, (top + bottom) / 2
def test_tap_emits_once_on_release_and_never_changes_the_supplied_state(gesture):
profile, touch, requests, state = gesture
point = center(personality_bounds(profile, state))
touch.press(*point, state)
assert touch.claimed and not requests
touch.release(*point, state)
touch.release(*point, state)
assert requests == [OnroadRequest('cycle_personality')]
assert not touch.claimed and state.personality == 1
@pytest.mark.parametrize('moved', (False, True))
@pytest.mark.parametrize('rhd', (False, True))
def test_drawn_bounds_and_tap_target_agree_for_both_layouts(gesture, moved, rhd):
profile, touch, requests, state = gesture
if moved:
state.customization['layouts'][profile]['following_distance'].update(x=260, y=20)
state = replace(state, right_hand_drive=rhd)
bounds = personality_bounds(profile, state)
assert bounds is not None
if profile == Profile.LARGE and rhd and not moved:
assert bounds[0] == 1280
widget = PersonalityWidget(cast(BitmapFonts, NS(profile=profile)))
widget.draw = Mock()
widget.render(state)
rect = widget.draw.call_args.args[0]
assert (rect.x, rect.y, rect.x + rect.width, rect.y + rect.height) == bounds
touch.press(*center(bounds), state)
touch.release(*center(bounds), state)
assert requests == [OnroadRequest('cycle_personality')]
@pytest.mark.parametrize('change', ('drag', 'exit', 'disabled', 'layout', 'drive', 'camera', 'driver', 'reverse_driver', 'alert', 'menu', 'cancel'))
def test_interrupted_gesture_cannot_activate_or_reactivate(gesture, change):
profile, touch, requests, state = gesture
bounds = personality_bounds(profile, state)
assert bounds is not None
x, y = bounds[0] + 1, (bounds[1] + bounds[3]) / 2
touch.press(x, y, state)
if change == 'drag':
touch.move(x + 6, y, state)
elif change == 'exit':
touch.move(x - 2, y, state)
elif change in ('disabled', 'layout'):
state.customization['layouts'][profile]['following_distance'].update({'enabled': False} if change == 'disabled' else {'x': bounds[0] + 1})
elif change == 'drive':
state = replace(state, drive_frame=2)
elif change == 'camera':
state = replace(state, camera_available=False)
elif change == 'driver':
state = replace(state, appearance=replace(state.appearance, camera_view=CameraViewChoice.DRIVER))
elif change == 'reverse_driver':
state = replace(state, reverse_driver_camera=True)
elif change == 'alert':
state = replace(state, alert=OnroadAlert(AlertSize.FULL, 'Take control', critical=True))
elif change == 'menu':
state = replace(state, quick_select_open=True)
else:
touch.cancel()
touch.release(x, y, state)
assert not requests and not touch.claimed
def test_personality_notice_and_incoming_acknowledgement_do_not_block_taps(gesture):
profile, touch, requests, state = gesture
state = replace(state, alert=OnroadAlert(AlertSize.MID, 'Standard', 'Driving Personality', alert_type='personalityChanged/warning'))
point = center(personality_bounds(profile, state))
for _ in range(3):
touch.press(*point, state)
touch.release(*point, replace(state, personality=2))
assert requests == [OnroadRequest('cycle_personality')] * 3
def test_critical_alert_cannot_impersonate_a_personality_notice(gesture):
profile, touch, requests, state = gesture
point = center(personality_bounds(profile, state))
state = replace(state, alert=OnroadAlert(AlertSize.MID, critical=True, alert_type='personalityChanged/warning'))
touch.press(*point, state)
touch.release(*point, state)
assert not requests and not touch.claimed
def test_speed_limit_controls_keep_priority_over_an_overlapping_widget(gesture):
profile, touch, requests, state = gesture
state = replace(state, speed_limit=SpeedLimitObservation(
kind=ObservationKind.VALID, speed_limit_mps=20, pending_speed_limit_mps=22, action_enabled=True,
session_id='drive', decision_id=1, presentation_id=1))
point = center(speed_limit_bounds(profile, state))
touch.press(*point, state)
touch.release(*point, state)
state = touch.present(state)
control = slc_controls(profile, state)[0]
x, y = center(control.bounds)
width, height = widget_size(state.customization, profile, 'following_distance')
state.customization['layouts'][profile]['following_distance'].update(x=x - width / 2, y=y - height / 2)
touch.press(x, y, state)
touch.release(x, y, state)
assert requests == [control.request] and isinstance(requests[0], SlcUiRequest)
@pytest.mark.parametrize('overlay', (False, True))
def test_session_consumes_widget_tap_without_home_or_favorite_fallthrough(gesture, overlay):
profile, _, requests, state = gesture
state.customization['layouts'][profile]['following_distance'].update(x=260, y=20)
session = StarShellSession.__new__(StarShellSession)
session.profile = profile
vars(session)['favorites'] = NS(is_open=overlay, claims_corner=lambda *_: False, press=Mock(return_value=overlay), release=Mock())
vars(session)['snapshot'] = vars(session)['_input_snapshot'] = lambda *_: NS(mode=ShellMode.ONROAD, selected=Destination.STAR, onroad=state)
session._update_favorites = Mock()
session.input = ShellInput(profile, requests.append)
vars(session)['view'] = NS(onroad=NS(navigation=NS(press=Mock(return_value=False))))
point = center(personality_bounds(profile, state))
session.press(ShellMode.ONROAD, *point)
assert session.release(ShellMode.ONROAD, *point)
if overlay:
assert not requests
session.favorites.release.assert_called_once()
else:
assert len(requests) == 1 and requests[0].action == OnroadRequest('cycle_personality')
session.favorites.press.assert_not_called()
session.view.onroad.navigation.press.assert_not_called()