Use general standby wakes and six event selections

This commit is contained in:
AngusBell97
2026-09-15 16:14:36 +01:00
committed by firestar5683
parent cdaf33529a
commit 20f38f3d8e
21 changed files with 251 additions and 829 deletions
-5
View File
@@ -701,17 +701,12 @@ inline static std::unordered_map<std::string, ParamKeyAttributes> keys = {
{"StandardJerkSpeedDecrease", {PERSISTENT, FLOAT, "100.0", "100.0", 3}},
{"StandbyMode", {PERSISTENT, BOOL, "0", "0", 1, SETTINGS_SIMPLE}},
{"StandbyButtonPressTime", {CLEAR_ON_MANAGER_START | DONT_LOG, INT, "0", "0"}},
{"StandbyWakeTouch", {PERSISTENT, BOOL, "1", "1", 2, SETTINGS_SIMPLE}},
{"StandbyWakeDriveState", {PERSISTENT, BOOL, "1", "1", 2, SETTINGS_SIMPLE}},
{"StandbyWakeButton", {PERSISTENT, BOOL, "0", "0", 2, SETTINGS_SIMPLE}},
{"StandbyWakeEngage", {PERSISTENT, BOOL, "1", "1", 2, SETTINGS_SIMPLE}},
{"StandbyWakeDisengage", {PERSISTENT, BOOL, "1", "1", 2, SETTINGS_SIMPLE}},
{"StandbyWakeInfoAlert", {PERSISTENT, BOOL, "1", "1", 2, SETTINGS_SIMPLE}},
{"StandbyWakeWarningAlert", {PERSISTENT, BOOL, "1", "1", 2, SETTINGS_SIMPLE}},
{"StandbyWakeCriticalAlert", {PERSISTENT, BOOL, "1", "1", 2, SETTINGS_SIMPLE}},
{"StandbyWakeTurnSignal", {PERSISTENT, BOOL, "0", "0", 2, SETTINGS_SIMPLE}},
{"StandbyWakeBrake", {PERSISTENT, BOOL, "0", "0", 2, SETTINGS_SIMPLE}},
{"StandbyWakeAccelerator", {PERSISTENT, BOOL, "0", "0", 2, SETTINGS_SIMPLE}},
{"StartAccel", {PERSISTENT, FLOAT, "0.0", "0.0", 3}},
{"StartAccelStock", {PERSISTENT, FLOAT, "0.0", "0.0", 3}},
{"StartupMessageBottom", {PERSISTENT, STRING, "Always keep hands on wheel and eyes on road", "Always keep hands on wheel and eyes on road", 0}},
@@ -257,7 +257,7 @@ class SystemSettingsManagerView(PanelManagerView):
self._toggle_defs = [
{
"title": tr("Standby Mode"),
"subtitle": tr("Only selected events wake the screen. Choose wake events below."),
"subtitle": tr("Touch and button presses always wake the screen. Choose additional wake events below."),
"get_state": lambda: self._controller._params.get_bool("StandbyMode"),
"set_state": lambda v: self._controller._params.put_bool("StandbyMode", v),
},
+1 -1
View File
@@ -300,7 +300,7 @@ class ScreenSettingsLayoutMici(NavScroller):
self._onroad_timeout = self._timeout_button("ScreenTimeoutOnroad", "onroad timeout")
standby = ScreenToggleMici("standby mode", self._params, "StandbyMode")
self._wake_controls = [ScreenWakeToggleMici("wake: " + label.lower(), self._params, key, default) for key, label, default in SCREEN_WAKE_OPTIONS]
explanation = GreyBigButton("", "Standby sleeps the screen onroad. Only selected wake events wake it.")
explanation = GreyBigButton("", "Standby sleeps the screen onroad. Touch and button presses always wake it.")
self._refresh_controls = [management, offroad_timeout, self._onroad_timeout, standby, *self._wake_controls]
for control in [offroad, onroad, offroad_timeout, self._onroad_timeout, standby, *self._wake_controls]:
control.set_enabled(lambda: self._params.get_bool("ScreenManagement", default=True))
@@ -47,9 +47,9 @@ class TestNativeScreenControls:
refresh = method(MICi, 'ScreenToggleMici', 'refresh')
params = self._real_params()
values = []
control = SimpleNamespace(_params=params, _key='StandbyWakeBrake', _default=False, set_checked=values.append)
control = SimpleNamespace(_params=params, _key='StandbyWakeTurnSignal', _default=False, set_checked=values.append)
refresh(control)
params.put_bool('StandbyWakeBrake', True)
params.put_bool('StandbyWakeTurnSignal', True)
refresh(control)
assert values == [False, True]
@@ -78,13 +78,13 @@ class TestNativeScreenControls:
namespace = {
'self': view,
'tr': lambda text: text,
'SCREEN_WAKE_DESCRIPTIONS': {'StandbyWakeEngage': 'Wake when engaged.', 'StandbyWakeBrake': 'Wake when braking.'},
'SCREEN_WAKE_OPTIONS': [('StandbyWakeEngage', 'Engagement', True), ('StandbyWakeBrake', 'Brake', False)],
'SCREEN_WAKE_DESCRIPTIONS': {'StandbyWakeEngage': 'Wake when engaged.', 'StandbyWakeTurnSignal': 'Wake when indicating.'},
'SCREEN_WAKE_OPTIONS': [('StandbyWakeEngage', 'Engagement', True), ('StandbyWakeTurnSignal', 'Turn signals', False)],
}
exec(compile(ast.Module(body=[assignment], type_ignores=[]), 'native-wake-callbacks', 'exec'), namespace)
assert [option['get_state']() for option in view._wake_toggle_defs] == [True, False]
assert [option['subtitle'] for option in view._wake_toggle_defs] == ['Wake when engaged.', 'Wake when braking.']
params.put_bool('StandbyWakeBrake', True)
assert [option['subtitle'] for option in view._wake_toggle_defs] == ['Wake when engaged.', 'Wake when indicating.']
params.put_bool('StandbyWakeTurnSignal', True)
params.put_bool('StandbyWakeEngage', False)
assert [option['get_state']() for option in view._wake_toggle_defs] == [False, True]
@@ -94,9 +94,9 @@ class TestNativeScreenControls:
helper_spec.loader.exec_module(helper)
save = method(MICi, 'ScreenToggleMici', '_save', SCREEN_WAKE_KEYS=helper.SCREEN_WAKE_KEYS, write_screen_setting=helper.write_screen_setting)
params = self._real_params()
for key in ('ScreenManagement', 'StandbyMode', 'StandbyWakeBrake'):
for key in ('ScreenManagement', 'StandbyMode', 'StandbyWakeTurnSignal'):
save(SimpleNamespace(_params=params, _key=key, refresh=lambda: None), True)
assert [params.get_bool(key) for key in ('ScreenManagement', 'StandbyMode', 'StandbyWakeBrake')] == [True, True, True]
assert [params.get_bool(key) for key in ('ScreenManagement', 'StandbyMode', 'StandbyWakeTurnSignal')] == [True, True, True]
def test_mici_value_save_remembers_brightness_and_accepts_existing_timeout(self):
helper_spec = importlib.util.spec_from_file_location('screen_settings', ROOT / 'starpilot/common/screen_settings.py')
@@ -126,11 +126,11 @@ class TestNativeScreenControls:
view = SimpleNamespace(
_controller=SimpleNamespace(_params=params),
_toggle_defs=[{'title': 'Standby'}, {'title': 'Uploads'}],
_wake_toggle_defs=[{'title': 'Engagement'}, {'title': 'Brake'}],
_wake_toggle_defs=[{'title': 'Engagement'}, {'title': 'Turn signals'}],
)
assert get_defs(view) == [{'title': 'Standby'}, {'title': 'Uploads'}]
params.get_bool = lambda key: True
assert get_defs(view) == [{'title': 'Standby'}, {'title': 'Engagement'}, {'title': 'Brake'}, {'title': 'Uploads'}]
assert get_defs(view) == [{'title': 'Standby'}, {'title': 'Engagement'}, {'title': 'Turn signals'}, {'title': 'Uploads'}]
def test_mici_brightness_slider_selects_offset_or_manual_with_correct_bounds(self):
for mode, want in [('auto', ('ScreenBrightnessOnroadOffset', -30, 30, -25, '%')), ('manual', ('ScreenBrightnessOnroad', 0, 100, 73, '%'))]:
@@ -35,7 +35,7 @@ class TestNativeScreenSaveErrors:
self.params.put_int('ScreenBrightness', 101)
self.params.put_int('ScreenBrightnessManual', 37)
self.params.put_int('ScreenBrightnessOffset', -12)
self.params.put_bool('StandbyWakeBrake', False)
self.params.put_bool('StandbyWakeTurnSignal', False)
self.errors = []
with (
patch.object(big, 'show_screen_save_error', lambda: self.errors.append('save failed')),
@@ -107,20 +107,20 @@ class TestNativeScreenSaveErrors:
compile(ast.Expression(definition), "wake-controls", "eval"),
dict(namespace, self=view, tr=lambda value: value, SCREEN_WAKE_OPTIONS=SCREEN_WAKE_OPTIONS, SCREEN_WAKE_DESCRIPTIONS=SCREEN_WAKE_DESCRIPTIONS),
)
brake = next(control for control in definitions if "Brake pedal" in control["title"])
self.assert_recoverable(lambda: brake["set_state"](True))
assert brake["get_state"]() is False
assert self.params.get_bool("StandbyWakeBrake") is False
turn_signal = next(control for control in definitions if "Turn signals" in control["title"])
self.assert_recoverable(lambda: turn_signal["set_state"](True))
assert turn_signal["get_state"]() is False
assert self.params.get_bool("StandbyWakeTurnSignal") is False
@pytest.mark.parametrize("error_type", [OSError, ValueError, UnknownKeyName])
def test_c4_failed_wake_toggle_restores_checked_state(self, error_type):
control = mici.ScreenToggleMici.__new__(mici.ScreenToggleMici)
control._params, control._key, control._default = (self.params, 'StandbyWakeBrake', False)
control._params, control._key, control._default = (self.params, 'StandbyWakeTurnSignal', False)
control._checked = True
with patch.object(mici, 'write_screen_setting', side_effect=error_type('Screen settings busy')):
self.assert_recoverable(lambda: control._save(True))
assert not control._checked
assert not self.params.get_bool('StandbyWakeBrake')
assert not self.params.get_bool('StandbyWakeTurnSignal')
@pytest.mark.parametrize('key,minimum,saved', [('ScreenBrightnessOffset', -30, -12), ('ScreenBrightness', 0, 37)])
def test_c4_slider_failure_restores_visible_value_and_keeps_slider_open(self, key, minimum, saved):
@@ -198,7 +198,7 @@ class TestNativeScreenSaveErrors:
return control._set_mode(1)
else:
control = mici.ScreenToggleMici.__new__(mici.ScreenToggleMici)
control._params, control._key, control._default = self.params, "StandbyWakeBrake", False
control._params, control._key, control._default = self.params, "StandbyWakeTurnSignal", False
control._checked = True
def action():
@@ -208,7 +208,7 @@ class TestNativeScreenSaveErrors:
self.assert_recoverable(action)
assert self.params.get_int("ScreenBrightness") == 101
assert self.params.get_int("ScreenBrightnessManual") == 37
assert self.params.get_bool("StandbyWakeBrake") is False
assert self.params.get_bool("StandbyWakeTurnSignal") is False
if interface == "c3":
assert control._mode_index() == 0
else:
+48 -55
View File
@@ -13,9 +13,9 @@ from openpilot.selfdrive.ui.lib.prime_state import PrimeState
from openpilot.selfdrive.ui.lib.ui_param_cache import shared_ui_params
from openpilot.system.ui.lib.application import gui_app
from openpilot.starpilot.common.lateral_only_experimental import lateral_only_experimental_available
from openpilot.system.hardware import HARDWARE, PC, TICI
from openpilot.system.hardware import HARDWARE, PC
from openpilot.starpilot.common.screen_settings import (
StandbyWakeTracker, standby_alert_wake_key, brightness_preferences, calculate_screen_brightness, enabled_wake_keys, standby_button_press_time,
alert_wake_key, brightness_preferences, calculate_screen_brightness, enabled_wake_keys, standby_button_press_time,
)
BACKLIGHT_OFFROAD = 65 if HARDWARE.get_device_type() == "mici" else 50
@@ -304,8 +304,8 @@ class Device:
def __init__(self):
self._ignition = False
self._started = ui_state.started
self._last_button_press = standby_button_press_time(ui_state.params_memory)
self._last_turn_signal = None
self._interaction_time: float = -1
self._override_interactive_timeout: int | None = None
self._interactive_timeout_callbacks: list[Callable] = []
@@ -320,10 +320,9 @@ class Device:
self._screen_timeout = 30
self._screen_timeout_onroad = 30
self._standby_mode = False
self._screen_offset = 0
self._screen_offset_onroad = 0
self._last_status = ui_state.status
self._screen_offset = self._screen_offset_onroad = 0
self._wake_keys = frozenset()
self._wake_tracker = StandbyWakeTracker()
self._refresh_screen_settings(force=True)
self._offroad_brightness: int = BACKLIGHT_OFFROAD
@@ -338,11 +337,11 @@ class Device:
def set_override_interactive_timeout(self, timeout: int | None) -> None:
# Override the interactive timeout duration temporarily
self._override_interactive_timeout = timeout
self.reset_interactive_timeout()
self._reset_interactive_timeout()
@property
def interactive_timeout(self) -> int:
if self._override_interactive_timeout is not None and not ((ui_state.started or ui_state.ignition) and self._standby_mode):
if self._override_interactive_timeout is not None:
return self._override_interactive_timeout
timeout_onroad = self._screen_timeout_onroad
@@ -359,9 +358,7 @@ class Device:
self._interaction_time = time.monotonic() + self.interactive_timeout
def reset_interactive_timeout(self) -> None:
# Page lifecycle callbacks must not wake Standby without a selected event.
if not ((ui_state.started or ui_state.ignition) and self._standby_mode):
self._reset_interactive_timeout()
self._reset_interactive_timeout()
def add_interactive_timeout_callback(self, callback: Callable):
self._interactive_timeout_callbacks.append(callback)
@@ -424,7 +421,7 @@ class Device:
self._screen_offset_onroad,
self._wake_keys,
)
if previous != current and self._interaction_time > 0 and not ((ui_state.started or ui_state.ignition) and (previous[5] or self._standby_mode)):
if previous != current and self._interaction_time > 0:
self._reset_interactive_timeout()
def set_offroad_brightness(self, brightness: int | None):
@@ -463,23 +460,25 @@ class Device:
self._screen_offset_onroad if ui_state.started else self._screen_offset,
interactive=interactive,
awake=self._awake,
standby_timed_out=(ui_state.started or ui_state.ignition) and self._standby_mode and not interactive,
standby_timed_out=ui_state.started and self._standby_mode and not interactive,
)
def _update_wakefulness(self):
# Handle interactive timeout
drive_state_changed = ui_state.ignition != self._ignition or ui_state.started != self._started
standby_active = self._standby_mode and (ui_state.started or self._started or ui_state.ignition or self._ignition)
self._ignition, self._started = ui_state.ignition, ui_state.started
ignition_state_changed = ui_state.ignition != self._ignition
self._ignition = ui_state.ignition
events = self._wake_input_events() | self._active_standby_alerts()
touched = any(ev.left_down for ev in gui_app.mouse_events)
if touched:
events.add("StandbyWakeTouch")
if drive_state_changed:
events.add("StandbyWakeDriveState")
should_wake = bool(events & self._wake_keys) if standby_active else touched or drive_state_changed
if should_wake:
status_changed = ui_state.status != self._last_status and ui_state.status != UIStatus.OVERRIDE
self._last_status = ui_state.status
status_key = "StandbyWakeEngage" if ui_state.status == UIStatus.ENGAGED else "StandbyWakeDisengage"
input_events = self._wake_input_events()
selected_status_change = status_changed and status_key in self._wake_keys
selected_turn_signal = bool(input_events & self._wake_keys)
button_pressed = self._standby_mode and (ui_state.started or ui_state.ignition) and "button" in input_events
wake_for_onroad_event = (ui_state.started and self._standby_mode and self._screen_brightness_onroad != 0 and
(selected_status_change or self._visible_onroad_alert() or selected_turn_signal))
if ignition_state_changed or any(ev.left_down for ev in gui_app.mouse_events) or button_pressed or wake_for_onroad_event:
self._reset_interactive_timeout()
interaction_timeout = time.monotonic() > self._interaction_time
@@ -503,44 +502,38 @@ class Device:
def _wake_input_events(self):
button_time = standby_button_press_time(ui_state.params_memory)
external_press = button_time > self._last_button_press and 0 <= time.monotonic() - button_time / 1e9 < 2
button_pressed = button_time > self._last_button_press and 0 <= time.monotonic() - button_time / 1e9 < 2
self._last_button_press = button_time
if not ui_state.started:
events = self._wake_tracker.update()
if external_press:
events.add("StandbyWakeButton")
return events
selfdrive_state = self._fresh_message("selfdriveState")
car_state = self._fresh_message("carState")
gear = str(car_state.gearShifter) if car_state is not None else None
events = self._wake_tracker.update(
engaged=bool(selfdrive_state.enabled) if selfdrive_state is not None else None,
turn_signal=(int(car_state.leftBlinker) | (int(car_state.rightBlinker) << 1)) if car_state is not None else None,
brake=bool(car_state.brakePressed) if car_state is not None else None,
accelerator=bool(car_state.gasPressed) if car_state is not None else None,
drive_state=gear if gear not in ("unknown", "0") else None,
)
if external_press:
events.add("StandbyWakeButton")
events = {"button"} if button_pressed else set()
car_state = self._fresh_message("carState") if ui_state.started else None
turn_signal = (int(car_state.leftBlinker) | (int(car_state.rightBlinker) << 1)) if car_state is not None else None
if self._last_turn_signal is not None and turn_signal and turn_signal != self._last_turn_signal:
events.add("StandbyWakeTurnSignal")
self._last_turn_signal = turn_signal
return events
def _active_standby_alerts(self):
# Match Dom's alert predicate and primary-message precedence. The toggle
# selects the category; it does not introduce another alert renderer.
if not ui_state.started:
return set()
sm = ui_state.sm
try:
key = standby_alert_wake_key(
sm["selfdriveState"], sm["starpilotSelfdriveState"],
now=time.monotonic(), started_time=ui_state.started_time, started_frame=ui_state.started_frame,
updated=sm.updated["selfdriveState"], recv_frame=sm.recv_frame["selfdriveState"], recv_time=sm.recv_time["selfdriveState"],
primary_fresh=self._fresh_message("selfdriveState") is not None,
secondary_fresh=self._fresh_message("starpilotSelfdriveState") is not None,
tici=TICI, mici=HARDWARE.get_device_type() == "mici",
hide_alerts=getattr(ui_state, "starpilot_toggles", {}).get("hide_alerts", False),
)
except (KeyError, AttributeError):
return set()
return {key} if key is not None else set()
key = alert_wake_key(ui_state.sm["selfdriveState"])
if key is not None:
return {key}
except Exception:
pass
try:
alert = ui_state.sm["starpilotSelfdriveState"]
if str(alert.alertSize) not in ("none", "0"):
key = alert_wake_key(alert)
return {key} if key is not None else set()
except Exception:
pass
return set()
def _visible_onroad_alert(self):
return bool(self._active_standby_alerts() & self._wake_keys)
def _set_awake(self, on: bool):
if on != self._awake:
@@ -4751,8 +4751,8 @@
{
"key": "StandbyMode",
"label": "Standby Mode",
"description": "Turn the driving screen off after the onroad timeout. While Standby Mode is active, only the selected wake events turn it back on.",
"picker_description": "Turns the driving screen off and wakes it only for selected events.",
"description": "Turn off the screen after the onroad timeout. Touch and recognised Bluetooth or steering wheel button presses always wake it. Choose additional wake events below.",
"picker_description": "Turn off the driving screen after inactivity and choose additional wake events.",
"data_type": "bool",
"ui_type": "toggle",
"parent_key": "ScreenManagement",
@@ -4842,76 +4842,6 @@
"StandbyMode"
],
"settings_tier": "simple"
},
{
"key": "StandbyWakeBrake",
"label": "Brake pedal",
"description": "Wake the screen from Standby when the brake pedal is pressed.",
"data_type": "bool",
"ui_type": "toggle",
"default": false,
"parent_key": "StandbyMode",
"visible_when_all_true": [
"ScreenManagement",
"StandbyMode"
],
"settings_tier": "simple"
},
{
"key": "StandbyWakeAccelerator",
"label": "Accelerator pedal",
"description": "Wake the screen from Standby when the accelerator pedal is pressed.",
"data_type": "bool",
"ui_type": "toggle",
"default": false,
"parent_key": "StandbyMode",
"visible_when_all_true": [
"ScreenManagement",
"StandbyMode"
],
"settings_tier": "simple"
},
{
"key": "StandbyWakeTouch",
"label": "Touch screen",
"description": "Wake the screen from Standby when the screen is touched.",
"data_type": "bool",
"ui_type": "toggle",
"default": true,
"parent_key": "StandbyMode",
"visible_when_all_true": [
"ScreenManagement",
"StandbyMode"
],
"settings_tier": "simple"
},
{
"key": "StandbyWakeDriveState",
"label": "Car drive state changed",
"description": "Wake the screen from Standby when the gear changes between Park, Reverse, Neutral or Drive, or when ignition or driving state changes.",
"data_type": "bool",
"ui_type": "toggle",
"default": true,
"parent_key": "StandbyMode",
"visible_when_all_true": [
"ScreenManagement",
"StandbyMode"
],
"settings_tier": "simple"
},
{
"key": "StandbyWakeButton",
"label": "Bluetooth or steering wheel button",
"description": "Wake the screen from Standby when a connected Bluetooth or USB controller button, or a supported steering wheel button, is pressed.",
"data_type": "bool",
"ui_type": "toggle",
"default": false,
"parent_key": "StandbyMode",
"visible_when_all_true": [
"ScreenManagement",
"StandbyMode"
],
"settings_tier": "simple"
}
]
},
-63
View File
@@ -16,30 +16,20 @@ BRIGHTNESS_KEYS = ('ScreenBrightness', 'ScreenBrightnessOnroad')
SCREEN_INT_KEYS = frozenset(key + suffix for key in BRIGHTNESS_KEYS for suffix in ('', 'Manual', 'Offset'))
STANDBY_BUTTON_PRESS_PARAM = 'StandbyButtonPressTime'
SCREEN_WAKE_OPTIONS = (
('StandbyWakeTouch', 'Touch screen', True),
('StandbyWakeDriveState', 'Car drive state changed', True),
('StandbyWakeButton', 'Bluetooth or steering wheel button', False),
('StandbyWakeEngage', 'Engagement', True),
('StandbyWakeDisengage', 'Disengagement', True),
('StandbyWakeInfoAlert', 'Informational alerts', True),
('StandbyWakeWarningAlert', 'Warning alerts', True),
('StandbyWakeCriticalAlert', 'Critical / takeover alerts', True),
('StandbyWakeTurnSignal', 'Turn signals', False),
('StandbyWakeBrake', 'Brake pedal', False),
('StandbyWakeAccelerator', 'Accelerator pedal', False),
)
SCREEN_WAKE_DESCRIPTIONS = {
'StandbyWakeTouch': 'Wake the screen from Standby when you touch it.',
'StandbyWakeDriveState': 'Wake the screen from Standby when the gear, ignition or driving state changes.',
'StandbyWakeButton': 'Wake the screen from Standby when a connected Bluetooth or USB controller, or a supported steering wheel button, is pressed.',
'StandbyWakeEngage': 'Wake the screen from Standby when StarPilot engages.',
'StandbyWakeDisengage': 'Wake the screen from Standby when StarPilot disengages.',
'StandbyWakeInfoAlert': 'Wake the screen from Standby and keep it awake while an informational alert is displayed.',
'StandbyWakeWarningAlert': 'Wake the screen from Standby and keep it awake while a warning alert is displayed.',
'StandbyWakeCriticalAlert': 'Wake the screen from Standby and keep it awake while a critical or takeover alert is displayed.',
'StandbyWakeTurnSignal': 'Wake the screen from Standby when a turn signal is activated or its direction changes.',
'StandbyWakeBrake': 'Wake the screen from Standby when the brake pedal is pressed.',
'StandbyWakeAccelerator': 'Wake the screen from Standby when the accelerator pedal is pressed.',
}
SCREEN_WAKE_KEYS = frozenset(key for key, _, _ in SCREEN_WAKE_OPTIONS)
SCREEN_SETTING_KEYS = SCREEN_INT_KEYS | SCREEN_WAKE_KEYS
@@ -191,8 +181,6 @@ def calculate_screen_brightness(automatic, manual, offset=0, *, interactive=Fals
def alert_wake_key(alert):
status = str(getattr(alert, 'alertStatus', 'normal'))
size = str(getattr(alert, 'alertSize', 'none'))
if size in ('none', '0'):
return None
if status in ('critical', '2'):
return 'StandbyWakeCriticalAlert'
if status in ('userPrompt', '1'):
@@ -202,57 +190,6 @@ def alert_wake_key(alert):
return None
def standby_alert_wake_key(primary, secondary, *, now, started_time, started_frame, updated, recv_frame, recv_time,
primary_fresh, secondary_fresh, tici, mici, hide_alerts=False):
"""Resolve the current renderer alert without constructing a UI widget.
Both renderers generate startup/unresponsive alerts before reading normal
messages. C4 keeps the reboot alert critical; C3 renders it informational.
Raw stale messages never wake Standby, even if still cached by the renderer.
"""
waiting_for_startup = recv_frame < started_frame
if not updated:
if waiting_for_startup and now - started_time > 5:
return 'StandbyWakeInfoAlert'
missing = now - recv_time
if tici and not waiting_for_startup and missing > 5:
if getattr(primary, 'enabled', False) and missing - 5 < 10:
return 'StandbyWakeCriticalAlert'
return 'StandbyWakeCriticalAlert' if mici else 'StandbyWakeInfoAlert'
if waiting_for_startup:
return None
# Primary alerts take display precedence, even when that category is disabled.
if alert_wake_key(primary) is not None:
alert = primary if primary_fresh else None
else:
alert = secondary if secondary_fresh else None
if not mici and hide_alerts and str(getattr(alert, 'alertStatus', 'normal')) in ('normal', '0'):
return None
return alert_wake_key(alert)
class StandbyWakeTracker:
"""Detect new driver inputs; missing samples reset each input's history."""
def __init__(self):
self.previous = {}
def update(self, *, engaged=None, turn_signal=None, brake=None, accelerator=None, drive_state=None):
current = dict(engaged=engaged, turn_signal=turn_signal, brake=brake, accelerator=accelerator, drive_state=drive_state)
events = set()
for key, value in current.items():
previous = self.previous.get(key)
if value is not None and previous is not None and value != previous:
if key == 'engaged':
events.add('StandbyWakeEngage' if value else 'StandbyWakeDisengage')
elif key == 'drive_state':
events.add('StandbyWakeDriveState')
elif value:
events.add({'turn_signal': 'StandbyWakeTurnSignal', 'brake': 'StandbyWakeBrake', 'accelerator': 'StandbyWakeAccelerator'}[key])
self.previous = current
return events
def standby_button_press_time(params):
try:
return max(0, int(_raw(params, STANDBY_BUTTON_PRESS_PARAM) or 0))
@@ -1,113 +0,0 @@
"""Compare wake policy with real renderer methods without constructing widgets."""
import ast
from enum import IntEnum
from pathlib import Path
from types import SimpleNamespace
import pytest
from openpilot.starpilot.common.screen_settings import alert_wake_key, standby_alert_wake_key
class ProtoEnum(IntEnum):
@property
def raw(self):
return int(self)
class Size(ProtoEnum):
none = 0
small = 1
mid = 2
full = 3
class Status(ProtoEnum):
normal = 0
userPrompt = 1
critical = 2
starpilot = 3
def message():
return SimpleNamespace(enabled=True, alertSize=Size.none, alertStatus=Status.normal,
alertText1='', alertText2='', alertType='', alertHudVisual=0)
def renderer_method(mici, state, sm, tici):
root = Path(__file__).resolve().parents[3]
source = root / ('selfdrive/ui/mici/onroad/alert_renderer.py' if mici else 'selfdrive/ui/onroad/alert_renderer.py')
tree = ast.parse(source.read_text())
constants = {'ALERT_STARTUP_PENDING', 'ALERT_CRITICAL_TIMEOUT', 'ALERT_CRITICAL_REBOOT',
'SELFDRIVE_STATE_TIMEOUT', 'SELFDRIVE_UNRESPONSIVE_TIMEOUT'}
body = [node for node in tree.body if isinstance(node, ast.Assign) and any(
isinstance(target, ast.Name) and target.id in constants for target in node.targets)]
renderer = next(node for node in tree.body if isinstance(node, ast.ClassDef) and node.name == 'AlertRenderer')
body.append(next(node for node in renderer.body if isinstance(node, ast.FunctionDef) and node.name == 'get_alert'))
namespace = dict(Alert=SimpleNamespace, AlertSize=Size, AlertStatus=Status,
custom=SimpleNamespace(StarPilotSelfdriveState=SimpleNamespace(AlertSize=Size)),
messaging=SimpleNamespace(SubMaster=object), tr=lambda text: text, ui_state=state,
time=SimpleNamespace(monotonic=lambda: 100), TICI=tici)
exec(compile(ast.Module(body=body, type_ignores=[]), str(source), 'exec'), namespace)
return namespace['get_alert'](SimpleNamespace(), sm)
@pytest.mark.parametrize('mici', [False, True])
@pytest.mark.parametrize('scenario', [
'startup', 'startup_boundary', 'takeover', 'timeout_boundary', 'reboot', 'reboot_boundary', 'reboot_disengaged',
'nonhardware', 'updated_old_timestamp', 'info', 'warning', 'critical', 'no_size', 'secondary',
'primary_precedence', 'hidden_normal', 'visible_starpilot', 'hidden_generated_startup',
])
def test_wake_category_matches_actual_renderer_alert(mici, scenario):
class Messages(dict):
pass
sm = Messages(selfdriveState=message(), starpilotSelfdriveState=message())
sm.updated = {'selfdriveState': True}
sm.recv_frame = {'selfdriveState': 6}
sm.recv_time = {'selfdriveState': 100}
state = SimpleNamespace(started_time=90, started_frame=5, starpilot_toggles={})
tici = True
primary = sm['selfdriveState']
secondary = sm['starpilotSelfdriveState']
if scenario in ('startup', 'startup_boundary', 'hidden_generated_startup'):
sm.updated['selfdriveState'] = False
sm.recv_frame['selfdriveState'] = 4
if scenario == 'startup_boundary':
state.started_time = 95
if scenario == 'hidden_generated_startup':
state.starpilot_toggles['hide_alerts'] = True
elif scenario in ('takeover', 'timeout_boundary', 'reboot', 'reboot_boundary', 'reboot_disengaged', 'nonhardware', 'updated_old_timestamp'):
sm.updated['selfdriveState'] = scenario == 'updated_old_timestamp'
sm.recv_time['selfdriveState'] = {'timeout_boundary': 95, 'reboot': 84, 'reboot_boundary': 85}.get(scenario, 94)
primary.enabled = scenario != 'reboot_disengaged'
tici = scenario != 'nonhardware'
elif scenario == 'secondary':
secondary.alertSize = Size.small
elif scenario == 'visible_starpilot':
secondary.alertSize, secondary.alertStatus = Size.small, Status.starpilot
state.starpilot_toggles['hide_alerts'] = True
else:
primary.alertSize = Size.none if scenario == 'no_size' else Size.small
primary.alertStatus = (Status.userPrompt if scenario == 'warning' else
Status.critical if scenario in ('critical', 'no_size', 'primary_precedence') else Status.normal)
if scenario == 'primary_precedence':
secondary.alertSize = Size.small
if scenario == 'hidden_normal':
state.starpilot_toggles['hide_alerts'] = True
rendered = renderer_method(mici, state, sm, tici)
expected = alert_wake_key(SimpleNamespace(alertSize=rendered.size, alertStatus=rendered.status)) if rendered else None
actual = standby_alert_wake_key(primary, secondary, now=100,
started_time=state.started_time, started_frame=state.started_frame,
updated=sm.updated['selfdriveState'], recv_frame=sm.recv_frame['selfdriveState'], recv_time=sm.recv_time['selfdriveState'],
primary_fresh=True, secondary_fresh=True, tici=tici, mici=mici, hide_alerts=state.starpilot_toggles.get('hide_alerts', False))
assert actual == expected
@pytest.mark.parametrize('primary_stale', [False, True])
def test_stale_raw_alert_is_ignored_before_generated_timeout(primary_stale):
primary, secondary = message(), message()
alert = primary if primary_stale else secondary
alert.alertStatus, alert.alertSize = Status.critical, Size.full
assert standby_alert_wake_key(primary, secondary, now=100, started_time=90, started_frame=5,
updated=False, recv_frame=6, recv_time=99, primary_fresh=not primary_stale, secondary_fresh=primary_stale,
tici=True, mici=True) is None
@@ -106,7 +106,7 @@ def test_only_selected_alert_categories_wake(key, field, value):
@pytest.mark.parametrize('key,field', [
('StandbyWakeBrake', 'brakePressed'), ('StandbyWakeAccelerator', 'gasPressed'), ('StandbyWakeTurnSignal', 'leftBlinker'),
('StandbyWakeTurnSignal', 'leftBlinker'),
])
def test_selected_driver_input_wakes_once_and_can_sleep_while_held(key, field):
device, state, _ = make_device(**dict.fromkeys(screen.SCREEN_WAKE_KEYS, False))
@@ -121,31 +121,6 @@ def test_selected_driver_input_wakes_once_and_can_sleep_while_held(key, field):
assert device._calculate_brightness() == 0
def test_unselected_engagement_and_stale_alerts_do_not_wake():
device, state, _ = make_device(**dict.fromkeys(screen.SCREEN_WAKE_KEYS, False))
device._update_wakefulness()
state.sm['selfdriveState'].enabled = True
state.status = type(state.status).ENGAGED
device._update_wakefulness()
assert device._calculate_brightness() == 0
device._params.values['StandbyWakeCriticalAlert'] = True
device._refresh_screen_settings(force=True)
device._interaction_time = 90
state.sm['selfdriveState'].alertStatus = 'critical'
state.sm['selfdriveState'].alertSize = 'full'
state.sm.alive['selfdriveState'] = False
device._update_wakefulness()
assert device._calculate_brightness() == 0
def test_selected_event_recovers_manual_zero_brightness():
device, state, _ = make_device(ScreenBrightnessOnroad=0)
state.sm['selfdriveState'].alertSize = 'full'
state.sm['selfdriveState'].alertStatus = 'critical'
device._update_wakefulness()
assert device._calculate_brightness() == 5
def test_runtime_limits_legacy_offsets_to_thirty_percent():
device, state, _ = make_device(StandbyMode=False, ScreenBrightnessOffset=-100, ScreenBrightnessOnroadOffset=-100)
assert device._calculate_brightness() == 46
@@ -162,23 +137,19 @@ def test_every_wake_choice_controls_its_own_event(key, selected):
device._update_wakefulness() # Seed signals; startup is not a driver event.
if key == 'StandbyWakeEngage':
state.sm['selfdriveState'].enabled = True
state.status = type(state.status).ENGAGED
elif key == 'StandbyWakeDisengage':
state.sm['selfdriveState'].enabled = True
state.status = type(state.status).ENGAGED
device._update_wakefulness()
state.sm['selfdriveState'].enabled = False
state.status = type(state.status).DISENGAGED
elif key.endswith('Alert'):
alert = state.sm['selfdriveState']
alert.alertSize = 'small'
alert.alertStatus = {'StandbyWakeInfoAlert': 'normal', 'StandbyWakeWarningAlert': 'userPrompt', 'StandbyWakeCriticalAlert': 'critical'}[key]
elif key in ('StandbyWakeBrake', 'StandbyWakeAccelerator', 'StandbyWakeTurnSignal'):
field = {'StandbyWakeBrake': 'brakePressed', 'StandbyWakeAccelerator': 'gasPressed', 'StandbyWakeTurnSignal': 'leftBlinker'}[key]
setattr(state.sm['carState'], field, True)
elif key == 'StandbyWakeTouch':
app.mouse_events = [SimpleNamespace(left_down=True)]
elif key == 'StandbyWakeDriveState':
state.sm['carState'].gearShifter = 'reverse'
elif key == 'StandbyWakeButton':
state.params_memory.values['StandbyButtonPressTime'] = 99_500_000_000
elif key == 'StandbyWakeTurnSignal':
state.sm['carState'].leftBlinker = True
else:
pytest.fail('No stimulus for wake option ' + key)
device._interaction_time = 90
@@ -186,15 +157,13 @@ def test_every_wake_choice_controls_its_own_event(key, selected):
assert (device._calculate_brightness() > 0) is selected
@pytest.mark.parametrize('selected', [False, True])
def test_external_button_press_is_fresh_and_consumed_once(selected):
def test_external_button_press_is_fresh_and_consumed_once():
settings = dict.fromkeys(screen.SCREEN_WAKE_KEYS, False)
settings['StandbyWakeButton'] = selected
device, state, _ = make_device(**settings)
device._update_wakefulness()
state.params_memory.values['StandbyButtonPressTime'] = 99_500_000_000
device._update_wakefulness()
assert (device._calculate_brightness() > 0) is selected
assert (device._calculate_brightness() > 0) is True
device._interaction_time = 90
device._update_wakefulness()
assert device._calculate_brightness() == 0
@@ -203,60 +172,8 @@ def test_external_button_press_is_fresh_and_consumed_once(selected):
assert device._calculate_brightness() == 0
@pytest.mark.parametrize('selected', [False, True])
def test_ignition_changes_obey_drive_state_choice(selected):
settings = dict.fromkeys(screen.SCREEN_WAKE_KEYS, False)
settings['StandbyWakeDriveState'] = selected
device, state, _ = make_device(**settings)
state.ignition = False
device._update_wakefulness()
assert (device._calculate_brightness() > 0) is selected
def test_settings_and_programmatic_timeouts_cannot_bypass_standby_choices():
device, _, _ = make_device(**dict.fromkeys(screen.SCREEN_WAKE_KEYS, False))
device._params.values['ScreenBrightnessOnroadOffset'] = 20
device._refresh_screen_settings(force=True)
assert device._interaction_time == 90
device.reset_interactive_timeout()
device.set_override_interactive_timeout(300)
assert device._interaction_time == 90
assert device.interactive_timeout == 30
assert device._calculate_brightness() == 0
def test_stale_button_messages_and_releases_do_not_wake():
device, state, _ = make_device(StandbyWakeButton=True)
device._update_wakefulness()
state.sm['carState'].buttonEvents = [SimpleNamespace(pressed=False, type='accelCruise')]
state.sm.logMonoTime['carState'] += 1
device._update_wakefulness()
assert device._calculate_brightness() == 0
state.sm['carState'].buttonEvents[0].pressed = True
state.sm.logMonoTime['carState'] += 1
state.sm.alive['carState'] = False
device._update_wakefulness()
state.sm.alive['carState'] = True
device._update_wakefulness()
assert device._calculate_brightness() == 0
@pytest.mark.parametrize('selected', [False, True])
def test_entering_ignition_before_onroad_obeys_drive_state_selection(selected):
device, state, _ = make_device(**{**dict.fromkeys(screen.SCREEN_WAKE_KEYS, False), 'StandbyWakeDriveState': selected})
state.started = state.ignition = device._started = device._ignition = False
device._update_wakefulness()
assert not device.awake
state.ignition = True
device._update_wakefulness()
assert (device._calculate_brightness() > 0) is selected
state.started = True
device._update_wakefulness()
assert (device._calculate_brightness() > 0) is selected
def test_consumed_button_press_is_not_replayed_between_ui_frames():
device, state, _ = make_device(StandbyWakeButton=True)
device, state, _ = make_device()
device._update_wakefulness()
state.params_memory.values['StandbyButtonPressTime'] = 99_500_000_000
device._update_wakefulness()
@@ -266,58 +183,6 @@ def test_consumed_button_press_is_not_replayed_between_ui_frames():
assert device._calculate_brightness() == 0
@pytest.mark.parametrize('operation', ['settings_change', 'public_reset', 'timeout_override'])
def test_force_offroad_cannot_wake_through_settings_or_timeout_overrides(operation):
device, state, _ = make_device(**dict.fromkeys(screen.SCREEN_WAKE_KEYS, False))
device._update_wakefulness()
state.started = False
assert state.ignition
device._update_wakefulness()
assert device._calculate_brightness() == 0
if operation == 'settings_change':
device._params.values['ScreenBrightnessOffset'] = 15
device._refresh_screen_settings(force=True)
elif operation == 'public_reset':
device.reset_interactive_timeout()
else:
device.set_override_interactive_timeout(300)
device._update_wakefulness()
assert device._interaction_time == 90
assert device.interactive_timeout == 30
assert device._calculate_brightness() == 0
@pytest.mark.parametrize('device_type', ['tici', 'mici'])
@pytest.mark.parametrize('fallback', ['startup', 'takeover', 'reboot'])
@pytest.mark.parametrize('selected', [False, True])
def test_rendered_system_alerts_obey_their_selected_category(device_type, fallback, selected):
category = 'StandbyWakeCriticalAlert' if fallback == 'takeover' or (fallback == 'reboot' and device_type == 'mici') else 'StandbyWakeInfoAlert'
settings = dict.fromkeys(screen.SCREEN_WAKE_KEYS, False)
settings[category] = selected
device, state, _ = make_device(device_type=device_type, **settings)
state.started_frame, state.started_time = 5, 90
state.sm.updated['selfdriveState'] = False
state.sm.alive['selfdriveState'] = False
state.sm.recv_frame['selfdriveState'] = 1 if fallback == 'startup' else 6
state.sm.recv_time['selfdriveState'] = 84 if fallback == 'reboot' else 94
state.sm['selfdriveState'].enabled = True
# Generated fallback alerts are returned before the normal hide-alerts filter.
state.starpilot_toggles['hide_alerts'] = True
device._update_wakefulness()
assert device._active_standby_alerts() == {category}
assert (device._calculate_brightness() > 0) is selected
def test_alert_with_no_displayed_size_does_not_wake():
device, state, _ = make_device(StandbyWakeCriticalAlert=True)
state.sm['selfdriveState'].alertStatus = 'critical'
state.sm['selfdriveState'].alertSize = 'none'
device._update_wakefulness()
assert device._active_standby_alerts() == set()
assert device._calculate_brightness() == 0
def test_hidden_secondary_alert_does_not_bypass_primary_category_selection():
settings = dict.fromkeys(screen.SCREEN_WAKE_KEYS, False)
settings['StandbyWakeInfoAlert'] = True
@@ -331,21 +196,20 @@ def test_hidden_secondary_alert_does_not_bypass_primary_category_selection():
@pytest.mark.parametrize('device_type', ['tici', 'mici'])
def test_normal_alert_hide_setting_matches_active_renderer(device_type):
def test_dom_alert_predicate_does_not_depend_on_renderer_hide_setting(device_type):
device, state, _ = make_device(device_type=device_type)
state.starpilot_toggles['hide_alerts'] = True
state.sm['selfdriveState'].alertSize = 'small'
device._update_wakefulness()
assert (device._calculate_brightness() > 0) is (device_type == 'mici')
assert (device._calculate_brightness() > 0) is True
@pytest.mark.parametrize('selected', [False, True])
def test_bluetooth_wake_during_ignition_only_standby(selected):
settings = {**dict.fromkeys(screen.SCREEN_WAKE_KEYS, False), 'StandbyWakeButton': selected}
def test_bluetooth_wake_during_ignition_only_standby():
settings = dict.fromkeys(screen.SCREEN_WAKE_KEYS, False)
device, state, _ = make_device(**settings)
state.started = device._started = False
state.started = False
state.ignition = device._ignition = True
device._update_wakefulness()
state.params_memory.values['StandbyButtonPressTime'] = 99_500_000_000
device._update_wakefulness()
assert (device._calculate_brightness() > 0) is selected
assert (device._calculate_brightness() > 0) is True
+2 -18
View File
@@ -104,7 +104,7 @@ def test_sleep_and_standby_override_positive_offsets():
@pytest.mark.parametrize('status,size,expected', [
('normal', 'none', None), ('normal', 'small', 'StandbyWakeInfoAlert'),
('userPrompt', 'small', 'StandbyWakeWarningAlert'), ('critical', 'full', 'StandbyWakeCriticalAlert'),
('critical', 'none', None),
('critical', 'none', 'StandbyWakeCriticalAlert'),
])
def test_alerts_are_classified_by_priority(status, size, expected):
assert screen().alert_wake_key(SimpleNamespace(alertStatus=status, alertSize=size)) == expected
@@ -113,26 +113,10 @@ def test_alerts_are_classified_by_priority(status, size, expected):
def test_defaults_preserve_alert_and_engagement_waking_only():
enabled = screen().enabled_wake_keys(Params())
assert enabled == {'StandbyWakeEngage', 'StandbyWakeDisengage', 'StandbyWakeInfoAlert', 'StandbyWakeWarningAlert',
'StandbyWakeCriticalAlert', 'StandbyWakeTouch', 'StandbyWakeDriveState'}
'StandbyWakeCriticalAlert'}
assert 'StandbyWakeCriticalAlert' not in screen().enabled_wake_keys(Params({'StandbyWakeCriticalAlert': False}))
def test_vehicle_triggers_are_edges_not_continuous_pedal_or_signal_states():
tracker = screen().StandbyWakeTracker()
assert tracker.update(engaged=False, turn_signal=0, brake=False, accelerator=False) == set()
assert tracker.update(engaged=True, turn_signal=1, brake=True, accelerator=True) == {
'StandbyWakeEngage', 'StandbyWakeTurnSignal', 'StandbyWakeBrake', 'StandbyWakeAccelerator'}
assert tracker.update(engaged=True, turn_signal=1, brake=True, accelerator=True) == set()
assert tracker.update(engaged=False, turn_signal=2, brake=False, accelerator=False) == {'StandbyWakeDisengage', 'StandbyWakeTurnSignal'}
def test_stale_vehicle_data_does_not_create_wake_events_when_it_returns():
tracker = screen().StandbyWakeTracker()
tracker.update(engaged=False, turn_signal=0, brake=False, accelerator=False)
assert tracker.update() == set()
assert tracker.update(engaged=True, turn_signal=1, brake=True, accelerator=True) == set()
@pytest.mark.parametrize('key', ['ScreenBrightness', 'ScreenBrightnessOnroad'])
@pytest.mark.parametrize('stored,expected', [(-100, -30), (100, 30), (-30, -30), (30, 30)])
def test_saved_offsets_are_limited_without_rewriting_preferences(key, stored, expected):
@@ -0,0 +1,105 @@
"""Requested standby contract, exercised through the actual Device class."""
import json
from pathlib import Path
from types import SimpleNamespace
import pytest
from openpilot.starpilot.common import screen_settings as screen
from test_screen_device_runtime import make_device
EXPECTED = {
'StandbyWakeEngage', 'StandbyWakeDisengage', 'StandbyWakeInfoAlert',
'StandbyWakeWarningAlert', 'StandbyWakeCriticalAlert', 'StandbyWakeTurnSignal',
}
def disabled():
return dict.fromkeys(EXPECTED, False)
def test_only_requested_wake_options_are_exposed():
assert screen.SCREEN_WAKE_KEYS == EXPECTED
root = Path(__file__).resolve().parents[3]
layout = json.loads((root / 'starpilot/common/assets/device_settings_layout.json').read_text())
keys = {p['key'] for section in layout for p in section['params'] if p['key'].startswith('StandbyWake')}
assert keys == EXPECTED
@pytest.mark.parametrize('source', ['touch', 'button'])
def test_touch_and_unassigned_button_always_wake_even_with_old_disabled_settings(source):
device, state, app = make_device(**disabled(), StandbyWakeTouch=False, StandbyWakeButton=False)
device._update_wakefulness()
if source == 'touch':
app.mouse_events = [SimpleNamespace(left_down=True)]
else:
state.params_memory.values['StandbyButtonPressTime'] = 99_500_000_000
device._update_wakefulness()
assert device.awake
assert device._calculate_brightness() > 0
@pytest.mark.parametrize('field,value,old_key', [
('brakePressed', True, 'StandbyWakeBrake'),
('gasPressed', True, 'StandbyWakeAccelerator'),
('gearShifter', 'reverse', 'StandbyWakeDriveState'),
])
def test_removed_driver_triggers_do_not_wake_even_with_old_enabled_settings(field, value, old_key):
device, state, _ = make_device(**disabled(), **{old_key: True})
device._update_wakefulness()
setattr(state.sm['carState'], field, value)
device._update_wakefulness()
assert device._calculate_brightness() == 0
@pytest.mark.parametrize('target,key', [('ENGAGED', 'StandbyWakeEngage'), ('DISENGAGED', 'StandbyWakeDisengage')])
@pytest.mark.parametrize('selected', [False, True])
def test_dom_status_transition_out_of_override_obeys_corresponding_toggle(target, key, selected):
device, state, _ = make_device(**{**disabled(), key: selected})
state.status = type(state.status).OVERRIDE
device._update_wakefulness()
assert device._calculate_brightness() == 0
state.status = getattr(type(state.status), target)
device._update_wakefulness()
assert (device._calculate_brightness() > 0) is selected
@pytest.mark.parametrize('status,key', [('userPrompt', 'StandbyWakeWarningAlert'), ('critical', 'StandbyWakeCriticalAlert')])
def test_dom_non_normal_alert_status_wakes_even_without_alert_size(status, key):
device, state, _ = make_device(**{**disabled(), key: True})
state.sm['selfdriveState'].alertStatus = status
device._update_wakefulness()
assert device._calculate_brightness() > 0
def test_dom_standby_powers_display_down_until_touch():
device, _, app = make_device(**disabled())
device._update_wakefulness()
assert not device.awake
app.mouse_events = [SimpleNamespace(left_down=True)]
device._update_wakefulness()
assert device.awake
@pytest.mark.parametrize('event', ['engage', 'alert'])
def test_dom_manual_zero_suppresses_automatic_status_and_alert_wakes(event):
device, state, app = make_device(ScreenBrightnessOnroad=0)
if event == 'engage':
state.status = type(state.status).ENGAGED
state.sm['selfdriveState'].enabled = True
else:
state.sm['selfdriveState'].alertStatus = 'critical'
state.sm['selfdriveState'].alertSize = 'full'
device._update_wakefulness()
assert device._calculate_brightness() == 0
app.mouse_events = [SimpleNamespace(left_down=True)]
device._update_wakefulness()
assert device._calculate_brightness() == 5
def test_dom_ignition_transition_remains_unconditional():
device, state, _ = make_device(**disabled(), StandbyWakeDriveState=False)
state.ignition = False
device._update_wakefulness()
assert device._calculate_brightness() > 0
+14 -18
View File
@@ -8,7 +8,7 @@ Auto remains the default. It uses the existing automatic brightness calculation
The existing onroad calculation follows camera exposure and filters changes. Parked Auto retains the existing base level: 50% on comma 3/3X and 65% on comma 4, with existing screen overrides still applied. This feature does not add an offroad ambient-light sensor.
Manual allows 0–100% and starts at 100% when there is no previous manual choice. Each context remembers its own manual value when switched to Auto. Existing manual selections remain selected. A selected wake event temporarily makes manual 0% visible at 5%.
Manual allows 0–100% and starts at 100% when there is no previous manual choice. Each context remembers its own manual value when switched to Auto. Existing manual selections remain selected. Touch and recognised button presses temporarily make manual 0% visible at 5%. As in Dom, manual onroad 0% suppresses automatic engagement and alert wakes; the same rule applies to the optional turn-signal wake.
New Galaxy shows a mode selector and the slider for that mode. Standby uses the normal Galaxy toggle styling and an enabled-only Manage/Close submenu containing the onroad timeout and wake choices. Native settings provide the same preferences using their existing screen sizes and navigation patterns.
@@ -16,29 +16,26 @@ New Galaxy shows a mode selector and the slider for that mode. Standby uses the
Both timeout readouts use seconds, with a 5–60 second range and 5 second steps. The parked timeout controls normal offroad sleep. The onroad timeout and wake choices are visible when Standby is enabled; the onroad timeout also remains the internal temporary-visibility duration for manual 0%.
During Standby, only selected conditions reset the wake timer. Changing settings, opening a page, or requesting a page-specific timeout does not independently wake it. Once Standby ends, ordinary parked touch and drive-transition behavior applies again.
Touch and recognised Bluetooth/USB or steering-wheel button presses always wake Standby, including buttons without an assigned action. There are no touch or button wake toggles and no additional controller actions to configure.
| Wake choice | Default | Trigger |
| --- | --- | --- |
| Touch screen | On | Screen touch |
| Car drive state changed | On | Gear, ignition or onroad state changes |
| Bluetooth or steering wheel button | Off | Connected Bluetooth/USB controller button or supported vehicle button press |
| Engagement | On | Assistance becomes enabled |
| Disengagement | On | Assistance becomes disabled |
| Informational alerts | On | A displayed informational alert |
| Warning alerts | On | A displayed warning alert |
| Critical / takeover alerts | On | A displayed critical or takeover alert |
| Engagement | On | UI status changes to engaged, including returning from override |
| Disengagement | On | UI status changes to disengaged, including returning from override |
| Informational alerts | On | Dom reports an informational onroad alert |
| Warning alerts | On | Dom reports a warning onroad alert |
| Critical / takeover alerts | On | Dom reports a critical or takeover onroad alert |
| Turn signals | Off | Signal activation or direction change |
| Brake pedal | Off | Brake press |
| Accelerator pedal | Off | Accelerator press |
Selected alerts keep the screen awake while displayed. Category selection follows the comma 3/4 renderers, including generated startup and unresponsive-system alerts and primary-alert precedence. Hidden or stale raw alerts do not bypass selection. Pedals, signals and buttons use transitions, so holding one does not continually refresh the timer. Disabling an alert wake controls the screen only; it does not change the underlying alert or sound.
Engagement and alert detection use Dom's existing status and alert predicates. Entering override alone does not wake. A selected alert keeps resetting the timer while it remains reported. Primary alerts take precedence over the secondary StarPilot alert state, as in Dom; this code does not duplicate renderer-generated alerts or add a separate freshness policy. Wake preferences only affect the display, not the alert or its sound.
Vehicle button coverage depends on the car interface. Generic button events are drained through a nonconflating carState subscription because a single-frame event can disappear between UI frames. Tesla Model 3/Y additionally use a passive subscriber to the existing UI_warning/scrollWheelPressed signal. The Tesla signal detects left/right/down presses, not wheel rotation or a second overlapping press while its aggregate pressed bit remains set. Neither observer transmits CAN nor changes driving button events. Controller actions retain their separate enable toggle; an unmapped button can wake the screen without executing an action.
Dom's ignition transitions, screen-setting changes and page timeout handling are retained. There are no gear, brake-pedal or accelerator wake triggers. Standby powers the display down after the timeout using Dom's existing display-power path.
Vehicle buttons use the car interface's existing decoded `carState.buttonEvents`. The listener drains every message so short presses survive between UI refreshes. Controller buttons use the existing input-device reader. Fresh presses wake once; releases, key repeat and held buttons do not keep extending the timer. Mapped actions retain their separate enable setting and continue to work normally. Button coverage depends on what the existing vehicle interface and supported input devices expose; this PR introduces no manufacturer-specific CAN decoding.
## Persistence and compatibility
The existing brightness keys retain 101 as Auto and 0–100 as Manual. Four additional persistent integers store manual memory and relative offsets. Eleven persistent booleans store wake selections. StandbyButtonPressTime carries fresh button timestamps in RAM, clears on manager start, and is excluded from logging.
The existing brightness keys retain 101 as Auto and 0–100 as Manual. Four additional persistent integers store manual memory and relative offsets. Six persistent booleans store wake selections. StandbyButtonPressTime carries fresh button timestamps in RAM, clears on manager start, and is excluded from logging.
Native UI and Galaxy writes use one shared validator and an advisory nonblocking file lock outside the Params key directory. Snapshot, write, readback and rollback run within that transaction; UI caches invalidate inside and after it. A busy or failed save is reported and can be retried. Other direct Params writers must use the shared helper to participate in this transaction contract.
@@ -46,7 +43,7 @@ The Params registry source must be included in the normal device build before in
## Focused verification
From a configured Linux checkout with the project Python dependencies, including the compiled opendbc parser/packer:
From a configured Linux checkout with the project Python dependencies:
```sh
PYTHONPATH=. python -m pytest -q -c /dev/null --confcutdir=starpilot/common/tests \
@@ -62,9 +59,8 @@ PYTHONPATH=. python -m pytest -q -c /dev/null --confcutdir=starpilot/system/the_
starpilot/system/the_galaxy/tests/test_frontend_module_graph.py
node starpilot/system/the_galaxy/tests/test_screen_settings_dom.cjs
node starpilot/system/the_galaxy/tests/test_tesla_can_wake_frontend.cjs
```
The DOM test requires Playwright and Chromium. PLAYWRIGHT_MODULE and CHROMIUM_EXECUTABLE can point to an existing installation; GALAXY_DOM_SCREENSHOT optionally saves previews. It loads the real Vue components with synthetic API responses and no device writes. The Python commands bypass unrelated manager-wide fixtures and explicitly include starpilot tests, which are outside the repository's default testpaths. A fully built environment can additionally run selfdrive/ui/tests/test_device_screen_settings.py through the normal pytest configuration.
Automated tests cover every wake choice enabled and disabled, freshness, held inputs, generated-alert parity, minimum brightness, write failures, cross-process saves, native controls, browser interactions and existing controller actions. Physical screen readability and actual car input coverage still require checks on the relevant hardware.
Automated tests cover every wake choice enabled and disabled, generic-button freshness, held inputs, Dom status and alert behavior, minimum brightness, write failures, cross-process saves, native controls, browser interactions and existing controller actions. Physical screen readability and actual car input coverage still require checks on the relevant hardware.
@@ -10,8 +10,7 @@ const layout = JSON.parse(fs.readFileSync(path.join(repo, 'starpilot/common/asse
const section = layout.find(s => s.params.some(p => p.key === 'ScreenManagement'))
const wakeDefaults = {
StandbyWakeEngage:true, StandbyWakeDisengage:true, StandbyWakeInfoAlert:true, StandbyWakeWarningAlert:true,
StandbyWakeCriticalAlert:true, StandbyWakeTurnSignal:false, StandbyWakeBrake:false, StandbyWakeAccelerator:false,
StandbyWakeTouch:true, StandbyWakeDriveState:true, StandbyWakeButton:false,
StandbyWakeCriticalAlert:true, StandbyWakeTurnSignal:false,
}
const wakes = Object.keys(wakeDefaults)
const fixture = `
@@ -155,7 +154,7 @@ createApp({components:{SettingTree},setup:()=>({values}),
assert.equal(await page.locator('.gx-wake-choice, [data-wake-choice]').count(),0,'wake events use ordinary Galaxy switches')
for(const key of wakes) {
const param=section.params.find(p=>p.key===key)
assert.ok(param && param.ui_type==='toggle','all eleven wake events use standard toggles')
assert.ok(param && param.ui_type==='toggle','all six wake events use standard toggles')
assert.equal(param.default,wakeDefaults[key],key+' retains its required default')
assert.match(param.description,/wake.*standby|standby.*wake/i,key+' explains waking Standby')
assert.equal(await wakeRow(key).getByText(param.description,{exact:true}).count(),1,key+' description is visible')
@@ -164,11 +163,8 @@ createApp({components:{SettingTree},setup:()=>({values}),
await checkbox.setChecked(!wakeDefaults[key])
await page.waitForFunction(({key,expected})=>window.values[key]===expected,{key,expected:!wakeDefaults[key]})
}
assert.equal(await page.getByText(/Touch always wakes|ignition changes also wake/).count(),0)
assert.match(section.params.find(p=>p.key==='StandbyMode').description,/only.*selected/i)
assert.match(section.params.find(p=>p.key==='StandbyWakeDriveState').description,/Park.*Reverse.*Neutral.*Drive/)
assert.match(section.params.find(p=>p.key==='StandbyWakeDriveState').description,/ignition.*driving/)
assert.match(section.params.find(p=>p.key==='StandbyWakeButton').description,/Bluetooth.*steering wheel/)
assert.match(section.params.find(p=>p.key==='StandbyMode').description,/Touch.*Bluetooth.*steering wheel.*always wake/i)
assert.deepEqual(section.params.filter(p=>p.key.startsWith('StandbyWake')).map(p=>p.key).sort(), [...wakes].sort())
await page.waitForFunction(()=>!Array.from(document.querySelectorAll('.gx-switch input')).some(input=>input.disabled))
await page.evaluate(()=>{window.values.ScreenBrightnessOffset=80})
assert.equal(await offSlider.inputValue(),'30','older saved offsets stay within the new display range')
@@ -196,6 +192,6 @@ createApp({components:{SettingTree},setup:()=>({values}),
await page.evaluate(()=>{window.values.StandbyMode=true;window.values.ScreenManagement=false})
assert.equal(await wakeRow(wakes[0]).count(),0,'disabled Screen Settings hides wake choices')
assert.deepEqual(errors,[])
console.log('PASS: real Vue Auto/Manual controls, manual100 default/reset, +/-30% offsets, memory/zero, independent contexts, drag/pending stability, save rollback, eleven described standard wake toggles, Manage/Close submenu, seconds readouts, mobile layout, zero page errors')
console.log('PASS: real Vue Auto/Manual controls, manual100 default/reset, +/-30% offsets, memory/zero, independent contexts, drag/pending stability, save rollback, six described standard wake toggles, Manage/Close submenu, seconds readouts, mobile layout, zero page errors')
} finally {await browser.close()}
})().catch(error=>{console.error(error);process.exitCode=1})
@@ -1,53 +0,0 @@
"""Observe Tesla wheel presses for the display without changing vehicle button events."""
from opendbc.can import CANParser
from opendbc.car import Bus
from opendbc.car.tesla.values import CANBUS, CAR, DBC
MAX_SAMPLE_AGE_NS = 2_000_000_000
UI_WARNING_ADDRESS = 0x311
UI_WARNING_SIZE = 7
def tesla_button_dbc(cp) -> str | None:
if cp.brand != "tesla" or cp.carFingerprint not in (CAR.TESLA_MODEL_3, CAR.TESLA_MODEL_Y):
return None
dbc = DBC[cp.carFingerprint][Bus.party]
return dbc if dbc == "tesla_model3_party" else None
class TeslaStandbyButtonObserver:
def __init__(self, dbc: str):
self._parser = CANParser(dbc, [("UI_warning", 0)], CANBUS.party)
self._pressed: bool | None = None
self._last_sample_ns: int | None = None
def update(self, messages, now_ns: int) -> int:
"""Return the latest fresh press's CAN boot-clock timestamp, or zero."""
if self._last_sample_ns is not None and not 0 <= now_ns - self._last_sample_ns < MAX_SAMPLE_AGE_NS:
self._pressed = None
pressed_at = 0
for message in messages:
timestamp = int(message.logMonoTime)
if not message.valid or not 0 <= now_ns - timestamp < MAX_SAMPLE_AGE_NS:
continue
if self._last_sample_ns is not None and timestamp <= self._last_sample_ns:
continue
frames = [(frame.address, frame.dat, frame.src) for frame in message.can
if frame.address == UI_WARNING_ADDRESS and frame.src == CANBUS.party and len(frame.dat) == UI_WARNING_SIZE]
if not frames:
continue
self._parser.update([(timestamp, frames)])
values = self._parser.vl_all["UI_warning"]["scrollWheelPressed"]
if not values:
continue
if self._last_sample_ns is not None and timestamp - self._last_sample_ns >= MAX_SAMPLE_AGE_NS:
self._pressed = None
for value in values:
pressed = bool(value)
if self._pressed is False and pressed:
pressed_at = timestamp
self._pressed = pressed
self._last_sample_ns = timestamp
return pressed_at
@@ -0,0 +1,28 @@
"""An unmapped controller listener must stay idle without persistent writes."""
import pytest
from openpilot.starpilot.system.wheel_controls import wheel_controlsd
from test_wheel_controlsd import FakeParams
@pytest.mark.parametrize('initially_enabled,expected_writes', [(False, []), (True, [(wheel_controlsd.ENABLED_PARAM, False)])])
def test_standby_listener_disables_unused_mappings_at_most_once(initially_enabled, expected_writes):
class RecordingParams(FakeParams):
def __init__(self):
super().__init__({'IsOffroad': False, 'ScreenManagement': True, 'StandbyMode': True,
wheel_controlsd.ENABLED_PARAM: initially_enabled})
self.writes = []
def put_bool(self, key, value):
self.writes.append((key, value))
super().put_bool(key, value)
params = RecordingParams()
daemon = wheel_controlsd.WheelControlsDaemon(params, FakeParams())
try:
for frame in range(20):
daemon._update_learning(frame / 10)
assert not params.get_bool(wheel_controlsd.ENABLED_PARAM)
assert params.writes == expected_writes
finally:
daemon.close()
@@ -24,7 +24,7 @@ def packet(timestamp, pressed=None, *, valid=True, button_type='accelCruise'):
@pytest.fixture
def car_buttons(monkeypatch):
params = FakeParams({'ScreenManagement': True, 'StandbyMode': True, 'StandbyWakeButton': True, 'IsOnroad': True})
params = FakeParams({'ScreenManagement': True, 'StandbyMode': True, 'IsOnroad': True})
memory = FakeParams()
daemon = wheel_controlsd.WheelControlsDaemon(params, memory)
queued, subscriptions = [], []
@@ -65,7 +65,7 @@ def test_short_button_edge_survives_later_empty_frame_and_is_consumed_once(car_b
assert len(subscriptions) == 1
@pytest.mark.parametrize('disabled', ['ScreenManagement', 'StandbyMode', 'StandbyWakeButton', 'IsOnroad'])
@pytest.mark.parametrize('disabled', ['ScreenManagement', 'StandbyMode', 'IsOnroad'])
def test_subscription_only_runs_when_needed_and_reopens_cleanly(car_buttons, disabled):
daemon, params, memory, queued, subscriptions, _boot, now = car_buttons
params.put_bool(disabled, False)
@@ -15,7 +15,7 @@ PRESS_PARAM = "StandbyButtonPressTime"
@pytest.fixture
def input_pipe():
params = FakeParams({"ScreenManagement": True, "StandbyMode": True, "StandbyWakeButton": True})
params = FakeParams({"ScreenManagement": True, "StandbyMode": True})
memory = FakeParams()
daemon = wheel_controlsd.WheelControlsDaemon(params, memory)
read_fd, write_fd = os.pipe()
@@ -60,8 +60,8 @@ def test_selected_joystick_buttons_wake_without_executing_mappings(input_pipe, m
assert actions == []
@pytest.mark.parametrize("disabled_key", ["ScreenManagement", "StandbyMode", "StandbyWakeButton"])
def test_button_wake_is_opt_in_and_enabling_it_while_held_does_not_create_a_press(input_pipe, monkeypatch, disabled_key):
@pytest.mark.parametrize("disabled_key", ["ScreenManagement", "StandbyMode"])
def test_enabling_standby_while_held_does_not_create_a_press(input_pipe, monkeypatch, disabled_key):
_daemon, params, memory, _fd, send = input_pipe
params.put_bool(disabled_key, False)
monkeypatch.setattr(wheel_controlsd.time, "monotonic_ns", lambda: 100)
@@ -162,10 +162,10 @@ def test_disconnected_device_does_not_suppress_next_press_on_reused_descriptor(i
(False, True, True, True, True),
(False, False, True, True, False),
(False, True, False, True, False),
(False, True, True, False, False),
(False, True, True, False, True),
(True, False, False, False, True),
])
def test_manager_runs_listener_for_enabled_mappings_or_selected_standby_wake(started, mapping, management, standby, button, expected):
def test_manager_runs_listener_for_enabled_mappings_or_standby(started, mapping, management, standby, button, expected):
# The manager module creates native processes at import, so load its real predicate only.
path = Path(__file__).resolve().parents[4] / "system/manager/process_config.py"
tree = ast.parse(path.read_text())
@@ -1,185 +0,0 @@
"""Decode real Tesla DBC frames; no CAN sockets or transmitters are opened."""
import importlib
import sys
from pathlib import Path
from types import SimpleNamespace
import pytest
from cereal import car, log
from opendbc.can import CANPacker
from opendbc.car.tesla.values import CAR
from openpilot.starpilot.system.wheel_controls import wheel_controlsd
from test_wheel_controlsd import FakeParams
def observer_module():
path = Path(__file__).resolve().parents[1] / "tesla_standby_buttons.py"
assert path.exists(), "The passive Tesla standby button observer is not implemented"
return importlib.import_module("openpilot.starpilot.system.wheel_controls.tesla_standby_buttons")
@pytest.fixture
def observer():
return observer_module().TeslaStandbyButtonObserver("tesla_model3_party")
def packet(timestamp, *pressed_values, bus=0, valid=True, address=None):
packer = CANPacker("tesla_model3_party")
event = log.Event.new_message(logMonoTime=timestamp, valid=valid)
event.init("can", len(pressed_values))
for frame, pressed in zip(event.can, pressed_values, strict=True):
addr, data, src = packer.make_can_msg("UI_warning", bus, {"scrollWheelPressed": pressed})
frame.address, frame.dat, frame.src = addr if address is None else address, data, src
return event
@pytest.mark.parametrize("model", [CAR.TESLA_MODEL_3, CAR.TESLA_MODEL_Y])
def test_only_supported_tesla_fingerprints_select_existing_party_dbc(model):
cp = car.CarParams.new_message(brand="tesla", carFingerprint=model.value)
assert observer_module().tesla_button_dbc(cp) == "tesla_model3_party"
cp.brand = "toyota"
assert observer_module().tesla_button_dbc(cp) is None
cp.brand, cp.carFingerprint = "tesla", CAR.TESLA_MODEL_S_PREAP.value
assert observer_module().tesla_button_dbc(cp) is None
def test_first_held_sample_seeds_then_only_new_presses_emit_original_timestamp(observer):
assert observer.update([packet(100, 1)], 100) == 0
assert observer.update([packet(200, 1)], 200) == 0
assert observer.update([packet(300, 0)], 300) == 0
assert observer.update([packet(400, 1)], 500) == 400
assert observer.update([packet(600, 1), packet(700, 0)], 800) == 0
def test_press_and_release_in_one_received_packet_are_not_lost(observer):
assert observer.update([packet(100, 0)], 100) == 0
assert observer.update([packet(200, 1, 0)], 250) == 200
assert observer.update([packet(300, 1)], 350) == 300
def test_wrong_bus_invalid_unrelated_truncated_and_future_messages_do_not_wake(observer):
assert observer.update([packet(100, 0)], 100) == 0
assert observer.update([packet(200, 1, bus=2)], 300) == 0
assert observer.update([packet(300, 1, valid=False)], 400) == 0
assert observer.update([packet(400, 1, address=0x312)], 500) == 0
assert observer.update([packet(600, 1)], 500) == 0
truncated = packet(600, 1)
truncated.can[0].dat = truncated.can[0].dat[:2]
assert observer.update([truncated], 700) == 0
assert observer.update([packet(800, 1)], 900) == 800
def test_stale_gap_reseeds_held_state_instead_of_waking(observer):
assert observer.update([packet(100, 0)], 100) == 0
assert observer.update([], 2_000_000_100) == 0
assert observer.update([packet(2_000_000_200, 1)], 2_000_000_300) == 0
assert observer.update([packet(2_000_000_400, 0), packet(2_000_000_500, 1)], 2_000_000_600) == 2_000_000_500
def test_old_packets_never_replay_a_button_press(observer):
assert observer.update([packet(100, 0), packet(200, 1)], 300) == 200
assert observer.update([packet(100, 0), packet(200, 1)], 400) == 0
assert observer.update([packet(500, 0)], 500) == 0
assert observer.update([packet(600, 1)], 2_000_000_600) == 0
assert observer.update([packet(2_000_000_700, 1)], 2_000_000_800) == 0
@pytest.fixture
def daemon_subscription(monkeypatch):
params = FakeParams({"ScreenManagement": True, "StandbyMode": True, "StandbyWakeButton": True,
"IsOnroad": True, "CarParams": car.CarParams.new_message(
brand="tesla", carFingerprint=CAR.TESLA_MODEL_3.value).to_bytes()})
memory = FakeParams()
daemon = wheel_controlsd.WheelControlsDaemon(params, memory)
subscriptions, queued, drains = [], [], []
def subscribe(endpoint, **kwargs):
if endpoint == "carState":
assert kwargs == {"conflate": False}
return "carState-subscription"
assert endpoint == "can"
sock = object()
subscriptions.append(sock)
return sock
def drain(sock, wait_for_one=False):
assert wait_for_one is False
if sock == "carState-subscription":
return []
drains.append(sock)
messages, queued[:] = list(queued), []
return messages
messaging = SimpleNamespace(sub_sock=subscribe, drain_sock=drain)
import cereal
monkeypatch.setitem(sys.modules, "cereal.messaging", messaging)
monkeypatch.setattr(cereal, "messaging", messaging, raising=False)
monkeypatch.setattr(wheel_controlsd.time, "monotonic_ns", lambda: 1_000_000_000)
monkeypatch.setattr(wheel_controlsd.time, "clock_gettime_ns", lambda _clock: 1_000_000_000)
yield daemon, params, memory, subscriptions, queued, drains
daemon.close()
def test_daemon_opens_can_only_for_enabled_onroad_supported_tesla_and_reseeds_changes(daemon_subscription):
daemon, params, memory, subscriptions, queued, drains = daemon_subscription
configure = getattr(daemon, "_configure_tesla_buttons", None)
assert callable(configure), "The passive Tesla subscription is not connected to the daemon"
params.put_bool("StandbyWakeButton", False)
configure()
assert subscriptions == []
params.put_bool("StandbyWakeButton", True)
params.put_bool("IsOnroad", False)
configure()
assert subscriptions == []
params.put_bool("IsOnroad", True)
configure()
configure()
assert len(subscriptions) == 1
queued[:] = [packet(100, 0), packet(200, 1)]
daemon._poll_tesla_buttons()
assert memory.get_int("StandbyButtonPressTime") == 200
params.put_bool("StandbyMode", False)
configure()
daemon._poll_tesla_buttons()
assert len(drains) == 1
params.put_bool("StandbyMode", True)
configure()
queued[:] = [packet(300, 1)]
daemon._poll_tesla_buttons()
assert len(subscriptions) == 2
assert memory.get_int("StandbyButtonPressTime") == 200
params.put("CarParams", car.CarParams.new_message(brand="tesla", carFingerprint=CAR.TESLA_MODEL_Y.value).to_bytes())
configure()
queued[:] = [packet(400, 1)]
daemon._poll_tesla_buttons()
assert len(subscriptions) == 3
assert memory.get_int("StandbyButtonPressTime") == 200
params.put("CarParams", car.CarParams.new_message(brand="toyota", carFingerprint="unsupported").to_bytes())
configure()
daemon._poll_tesla_buttons()
assert len(subscriptions) == 3
assert len(drains) == 3
def test_daemon_run_loop_publishes_passive_tesla_press_without_external_inputs(daemon_subscription, monkeypatch):
daemon, _params, memory, subscriptions, queued, _drains = daemon_subscription
queued[:] = [packet(100, 0), packet(200, 1)]
# Only hardware enumeration and the blocking selector are replaced; run/configure/poll stay real.
monkeypatch.setattr(daemon, "_scan_devices", lambda: None)
selections = iter([[]])
monkeypatch.setattr(daemon.selector, "select", lambda timeout: next(selections))
with pytest.raises(StopIteration):
daemon.run()
assert memory.get_int("StandbyButtonPressTime") == 200
assert len(subscriptions) == 1
def test_can_boot_clock_is_converted_to_ui_monotonic_clock_after_suspend(daemon_subscription, monkeypatch):
daemon, _params, memory, _subscriptions, queued, _drains = daemon_subscription
daemon._configure_tesla_buttons()
monkeypatch.setattr(wheel_controlsd.time, "clock_gettime_ns", lambda _clock: 6_000_000_000)
queued[:] = [packet(5_000_000_100, 0), packet(5_000_000_200, 1)]
daemon._poll_tesla_buttons()
assert memory.get_int("StandbyButtonPressTime") == 200
@@ -371,7 +371,7 @@ def start_learning(slot: int, params_memory: Params | None = None, params: Param
def cancel_learning(params_memory: Params | None = None, params: Params | None = None) -> None:
(params_memory or Params(memory=True)).remove(LEARN_SLOT_PARAM)
if params is not None and not load_mappings(params):
if params is not None and params.get_bool(ENABLED_PARAM) and not load_mappings(params):
params.put_bool(ENABLED_PARAM, False)
@@ -516,14 +516,9 @@ class WheelControlsDaemon:
self._car_state_sock = None
self._car_state_messaging = None
self._last_car_button_frame = 0
self._tesla_car_params: bytes | None = None
self._tesla_can_sock = None
self._tesla_messaging = None
self._tesla_button_observer = None
def close(self) -> None:
self._close_car_buttons()
self._close_tesla_buttons()
for fd in list(self.sources):
self._remove(fd)
self.selector.close()
@@ -669,7 +664,7 @@ class WheelControlsDaemon:
def _publish_button_press(self, timestamp: int) -> None:
try:
if not all(self.params.get_bool(key) for key in ("ScreenManagement", "StandbyMode", "StandbyWakeButton")):
if not all(self.params.get_bool(key) for key in ("ScreenManagement", "StandbyMode")):
return
self.params_memory.put_int(STANDBY_BUTTON_PRESS_PARAM, timestamp)
except Exception:
@@ -682,7 +677,7 @@ class WheelControlsDaemon:
self._last_car_button_frame = 0
def _configure_car_buttons(self) -> None:
if not all(self.params.get_bool(key) for key in ("ScreenManagement", "StandbyMode", "StandbyWakeButton", "IsOnroad")):
if not all(self.params.get_bool(key) for key in ("ScreenManagement", "StandbyMode", "IsOnroad")):
self._close_car_buttons()
return
if self._car_state_sock is not None:
@@ -719,54 +714,6 @@ class WheelControlsDaemon:
self._close_car_buttons()
cloudlog.exception("wheel controls: car button read failed")
def _close_tesla_buttons(self) -> None:
# SubSocket releases its native subscription in __dealloc__.
self._tesla_can_sock = None
self._tesla_button_observer = None
self._tesla_messaging = None
self._tesla_car_params = None
def _configure_tesla_buttons(self) -> None:
if not all(self.params.get_bool(key) for key in ("ScreenManagement", "StandbyMode", "StandbyWakeButton", "IsOnroad")):
self._close_tesla_buttons()
return
cp_bytes = self.params.get("CarParams")
if cp_bytes == self._tesla_car_params:
return
self._close_tesla_buttons()
self._tesla_car_params = cp_bytes
if not cp_bytes:
return
try:
from cereal import car
from openpilot.starpilot.system.wheel_controls.tesla_standby_buttons import TeslaStandbyButtonObserver, tesla_button_dbc
with car.CarParams.from_bytes(cp_bytes) as cp:
dbc = tesla_button_dbc(cp)
if dbc is not None:
from cereal import messaging
self._tesla_button_observer = TeslaStandbyButtonObserver(dbc)
self._tesla_can_sock = messaging.sub_sock("can")
self._tesla_messaging = messaging
except Exception:
self._close_tesla_buttons()
cloudlog.exception("wheel controls: passive Tesla button observer unavailable")
def _poll_tesla_buttons(self) -> None:
if self._tesla_can_sock is None:
return
try:
messages = self._tesla_messaging.drain_sock(self._tesla_can_sock, wait_for_one=False)
now_boot_ns = time.clock_gettime_ns(time.CLOCK_BOOTTIME)
now_ns = time.monotonic_ns()
timestamp = self._tesla_button_observer.update(messages, now_boot_ns)
if timestamp:
# pandad timestamps include suspend time; the UI and external inputs use monotonic().
self._publish_button_press(now_ns - (now_boot_ns - timestamp))
except Exception:
self._close_tesla_buttons()
cloudlog.exception("wheel controls: passive Tesla button read failed")
def _publish_status(self, now: float) -> None:
remaining = max(0, round(self.learning_deadline - now, 1)) if self.learning_slot is not None else 0
@@ -791,7 +738,6 @@ class WheelControlsDaemon:
if now - self.last_scan >= DEVICE_SCAN_INTERVAL_SECONDS:
self._scan_devices()
self._configure_car_buttons()
self._configure_tesla_buttons()
self.last_scan = now
for key, _mask in self.selector.select(timeout=0.1):
try:
@@ -799,7 +745,6 @@ class WheelControlsDaemon:
except (KeyError, OSError):
self._remove(key.fd)
self._poll_car_buttons()
self._poll_tesla_buttons()
now = time.monotonic()
if now - self.last_status >= STATUS_INTERVAL_SECONDS:
self._publish_status(now)
+1 -1
View File
@@ -135,7 +135,7 @@ def soundd_run(started: bool, params: Params, CP: car.CarParams, starpilot_toggl
def wheel_controls_enabled(started: bool, params: Params, CP: car.CarParams, starpilot_toggles: SimpleNamespace) -> bool:
return params.get_bool("WheelControlsEnabled") or (
params.get_bool("ScreenManagement") and params.get_bool("StandbyMode") and params.get_bool("StandbyWakeButton")
params.get_bool("ScreenManagement") and params.get_bool("StandbyMode")
)