feat(toyota): let supported cars own cruise set speed

This commit is contained in:
rav4kumar
2026-08-15 03:16:25 -07:00
parent 35d2485525
commit be88a29ace
15 changed files with 721 additions and 49 deletions
+1
View File
@@ -233,6 +233,7 @@ inline static std::unordered_map<std::string, ParamKeyAttributes> keys = {
{"TeslaMadsScreenButton", {PERSISTENT | BACKUP, INT, "0"}},
{"ToyotaEnforceStockLongitudinal", {PERSISTENT | BACKUP, BOOL, "0"}},
{"ToyotaStopAndGoHack", {PERSISTENT | BACKUP, BOOL, "0"}},
{"ToyotaVirtualCruiseSpeed", {PERSISTENT | BACKUP, BOOL, "0"}},
{"DynamicExperimentalControl", {PERSISTENT | BACKUP, BOOL, "0"}},
{"BlindSpot", {PERSISTENT | BACKUP, BOOL, "0"}},
+64 -7
View File
@@ -19,6 +19,7 @@ IMPERIAL_INCREMENT = round(CV.MPH_TO_KPH, 1) # round here to avoid rounding err
ButtonEvent = car.CarState.ButtonEvent
ButtonType = car.CarState.ButtonEvent.Type
CRUISE_LONG_PRESS = 50
TOYOTA_VIRTUAL_CRUISE_LONG_PRESS = 65
CRUISE_NEAREST_FUNC = {
ButtonType.accelCruise: math.ceil,
ButtonType.decelCruise: math.floor,
@@ -43,6 +44,30 @@ class VCruiseHelper(VCruiseHelperSP):
def v_cruise_initialized(self):
return self.v_cruise_kph != V_CRUISE_UNSET
@property
def software_pcm_cruise_speed(self) -> bool:
return self.CP.brand == "toyota" and self.CP.pcmCruise and self.CP.openpilotLongitudinalControl and not self.CP_SP.pcmCruiseSpeed
@property
def cruise_long_press_frames(self) -> int:
return TOYOTA_VIRTUAL_CRUISE_LONG_PRESS if self.software_pcm_cruise_speed else CRUISE_LONG_PRESS
@property
def software_pcm_cruise_initialized(self) -> bool:
return 0 < self.v_cruise_kph < V_CRUISE_UNSET and 0 < self.v_cruise_cluster_kph < V_CRUISE_UNSET
def _apply_software_pcm_cruise_delta(self, delta_kph: float, is_metric: bool) -> None:
"""Move Toyota's planner/display targets together while respecting both targets' bounds."""
cluster_min_kph = self.v_cruise_min if is_metric else self.v_cruise_min * CV.MPH_TO_KPH
min_delta = max(V_CRUISE_MIN - self.v_cruise_kph, cluster_min_kph - self.v_cruise_cluster_kph)
max_delta = min(V_CRUISE_MAX - self.v_cruise_kph, V_CRUISE_MAX - self.v_cruise_cluster_kph)
if delta_kph > 0:
applied_delta = min(delta_kph, max(0., max_delta))
else:
applied_delta = max(delta_kph, min(0., min_delta))
self.v_cruise_kph = round(self.v_cruise_kph + applied_delta, 1)
self.v_cruise_cluster_kph = round(self.v_cruise_cluster_kph + applied_delta, 1)
def update_v_cruise(self, CS, enabled, is_metric):
self.v_cruise_kph_last = self.v_cruise_kph
@@ -51,11 +76,21 @@ class VCruiseHelper(VCruiseHelperSP):
_enabled = self.update_enabled_state(CS, enabled)
if CS.cruiseState.available:
if not self.CP.pcmCruise or (not self.CP_SP.pcmCruiseSpeed and _enabled):
software_pcm_enabled = not self.CP_SP.pcmCruiseSpeed and _enabled
if self.software_pcm_cruise_speed:
software_pcm_enabled = software_pcm_enabled and self.software_pcm_cruise_initialized
if not self.CP.pcmCruise or software_pcm_enabled:
# if stock cruise is completely disabled, then we can use our own set speed logic
self._update_v_cruise_non_pcm(CS, _enabled, is_metric)
v_cruise_kph_before_sla = self.v_cruise_kph
self.update_speed_limit_assist_v_cruise_non_pcm()
self.v_cruise_cluster_kph = self.v_cruise_kph
if self.software_pcm_cruise_speed:
sla_delta_kph = self.v_cruise_kph - v_cruise_kph_before_sla
self.v_cruise_kph = v_cruise_kph_before_sla
self._apply_software_pcm_cruise_delta(sla_delta_kph, is_metric)
else:
self.v_cruise_cluster_kph = self.v_cruise_kph
else:
self.v_cruise_kph = CS.cruiseState.speed * CV.MS_TO_KPH
self.v_cruise_cluster_kph = CS.cruiseState.speedCluster * CV.MS_TO_KPH
@@ -85,13 +120,13 @@ class VCruiseHelper(VCruiseHelperSP):
for b in CS.buttonEvents:
if b.type.raw in self.button_timers and not b.pressed:
if self.button_timers[b.type.raw] > CRUISE_LONG_PRESS:
if self.button_timers[b.type.raw] > self.cruise_long_press_frames:
return # end long press
button_type = b.type.raw
break
else:
for k, timer in self.button_timers.items():
if timer and timer % CRUISE_LONG_PRESS == 0:
if timer and timer % self.cruise_long_press_frames == 0:
button_type = k
long_press = True
break
@@ -115,10 +150,26 @@ class VCruiseHelper(VCruiseHelperSP):
return
long_press, v_cruise_delta = VCruiseHelperSP.update_v_cruise_delta(self, long_press, v_cruise_delta)
if long_press and self.v_cruise_kph % v_cruise_delta != 0: # partial interval
self.v_cruise_kph = CRUISE_NEAREST_FUNC[button_type](self.v_cruise_kph / v_cruise_delta) * v_cruise_delta
# Toyota's canonical PCM set speed and displayed cluster set speed can differ. In
# software-owned PCM mode, round the value the driver sees and apply the same delta
# to both targets so the planner/cluster calibration offset remains intact.
v_cruise_reference = self.v_cruise_cluster_kph if self.software_pcm_cruise_speed else self.v_cruise_kph
if long_press and v_cruise_reference % v_cruise_delta != 0: # partial interval
v_cruise_reference_new = CRUISE_NEAREST_FUNC[button_type](v_cruise_reference / v_cruise_delta) * v_cruise_delta
else:
self.v_cruise_kph += v_cruise_delta * CRUISE_INTERVAL_SIGN[button_type]
v_cruise_reference_new = v_cruise_reference + v_cruise_delta * CRUISE_INTERVAL_SIGN[button_type]
if self.software_pcm_cruise_speed:
delta_kph = v_cruise_reference_new - v_cruise_reference
# If SET is pressed while overriding, do not lower the target below the current speed.
if CS.gasPressed and button_type in (ButtonType.decelCruise, ButtonType.setCruise):
delta_kph = max(delta_kph, CS.vEgo * CV.MS_TO_KPH - self.v_cruise_kph)
self._apply_software_pcm_cruise_delta(delta_kph, is_metric)
return
self.v_cruise_kph += v_cruise_reference_new - v_cruise_reference
# If set is pressed while overriding, clip cruise speed to minimum of vEgo
if CS.gasPressed and button_type in (ButtonType.decelCruise, ButtonType.setCruise):
@@ -127,6 +178,12 @@ class VCruiseHelper(VCruiseHelperSP):
self.v_cruise_kph = np.clip(round(self.v_cruise_kph, 1), self.v_cruise_min, V_CRUISE_MAX)
def update_button_timers(self, CS, enabled):
if self.software_pcm_cruise_speed and (not enabled or not CS.cruiseState.available or not self.software_pcm_cruise_initialized):
for k in self.button_timers:
self.button_timers[k] = 0
self.button_change_states[k] = {"standstill": False, "enabled": False}
return
# increment timer for buttons still pressed
for k in self.button_timers:
if self.button_timers[k] > 0:
@@ -143,7 +143,8 @@ class CruiseLayout(Widget):
self.icbm_toggle.show_description(True)
if has_long or has_icbm:
self.custom_acc_toggle.action_item.set_enabled(((has_long and not ui_state.CP.pcmCruise) or has_icbm) and ui_state.is_offroad())
software_cruise_speed = has_long and (not ui_state.CP.pcmCruise or not ui_state.CP_SP.pcmCruiseSpeed)
self.custom_acc_toggle.action_item.set_enabled((software_cruise_speed or has_icbm) and ui_state.is_offroad())
self.dec_toggle.action_item.set_enabled(has_long)
self.scc_v_toggle.action_item.set_enabled(True)
self.scc_m_toggle.action_item.set_enabled(True)
@@ -169,7 +170,7 @@ class CruiseLayout(Widget):
show_custom_acc_desc = True
else:
if has_long or has_icbm:
if has_long and ui_state.CP.pcmCruise:
if has_long and ui_state.CP.pcmCruise and ui_state.CP_SP.pcmCruiseSpeed:
new_custom_acc_desc = tr(ACC_PCMCRUISE_DISABLED_DESCRIPTION)
show_custom_acc_desc = True
else:
@@ -11,10 +11,12 @@ from openpilot.system.ui.lib.multilang import tr, tr_noop
from openpilot.system.ui.widgets import DialogResult
from openpilot.system.ui.widgets.confirm_dialog import ConfirmDialog
from openpilot.system.ui.sunnypilot.widgets.list_view import toggle_item_sp
from opendbc.sunnypilot.car.toyota.values import ToyotaFlagsSP
ONROAD_ONLY_DESCRIPTION = tr_noop("Start the vehicle to check vehicle compatibility.")
SNG_HACK_UNAVAILABLE = tr_noop("sunnypilot Longitudinal Control must be available and enabled for your vehicle to use this feature.")
VIRTUAL_CRUISE_UNAVAILABLE = tr_noop("Virtual Cruise Speed is available only on supported Toyota TSS2 configurations with sunnypilot Longitudinal Control.")
DESCRIPTIONS = {
'enforce_stock_longitudinal': tr_noop(
@@ -23,7 +25,14 @@ DESCRIPTIONS = {
'stop_and_go_hack': tr_noop(
'sunnypilot will allow some Toyota/Lexus cars to auto resume during stop and go traffic. ' +
'This feature is only applicable to certain models that are able to use longitudinal control. This is an alpha feature. Use at your own risk.'
)
),
'virtual_cruise_speed': tr_noop(
'Use a sunnypilot-owned cruise target with the Toyota RES/SET buttons while sunnypilot longitudinal control is active. ' +
'This unlocks Custom ACC Speed Increments; set the short interval to 5 for next-5-unit tap behavior. ' +
'The Toyota cluster will continue to show the factory target and may differ from sunnypilot. ' +
'The direct button signals are route-validated on Corolla Cross and Prius TSS2, but held-button timing differs by platform. ' +
'This is an alpha feature; validate acceleration above the factory target in a controlled setting.'
),
}
@@ -47,8 +56,17 @@ class ToyotaSettings(BrandSettings):
enabled=lambda: not ui_state.engaged,
)
self.virtual_cruise_speed = toggle_item_sp(
lambda: tr("Virtual Cruise Speed (Alpha)"),
description=lambda: tr(DESCRIPTIONS["virtual_cruise_speed"]),
initial_state=ui_state.params.get_bool("ToyotaVirtualCruiseSpeed"),
callback=self._on_enable_virtual_cruise_speed,
enabled=lambda: not ui_state.engaged,
)
self.items = [
self.enforce_stock_longitudinal,
self.virtual_cruise_speed,
self.stop_and_go_hack,
]
@@ -60,7 +78,9 @@ class ToyotaSettings(BrandSettings):
if ui_state.params.get_bool("AlphaLongitudinalEnabled"):
ui_state.params.put_bool("AlphaLongitudinalEnabled", False)
ui_state.params.put_bool("ToyotaStopAndGoHack", False)
ui_state.params.put_bool("ToyotaVirtualCruiseSpeed", False)
self.stop_and_go_hack.action_item.set_state(False)
self.virtual_cruise_speed.action_item.set_state(False)
ui_state.params.put_bool("OnroadCycleRequested", True)
else:
self.enforce_stock_longitudinal.action_item.set_state(False)
@@ -94,10 +114,46 @@ class ToyotaSettings(BrandSettings):
ui_state.params.put_bool("ToyotaStopAndGoHack", False)
ui_state.params.put_bool("OnroadCycleRequested", True)
def _on_enable_virtual_cruise_speed(self, state: bool):
if state:
def confirm_callback(result: int):
enabled = result == DialogResult.CONFIRM
ui_state.params.put_bool("ToyotaVirtualCruiseSpeed", enabled)
self.virtual_cruise_speed.action_item.set_state(enabled)
if enabled:
ui_state.params.put_bool("OnroadCycleRequested", True)
content = (f"<h1>{self.virtual_cruise_speed.title}</h1><br>" +
f"<p>{self.virtual_cruise_speed.description}</p>")
dlg = ConfirmDialog(content, tr("Enable"), rich=True, callback=confirm_callback)
gui_app.push_widget(dlg)
else:
ui_state.params.put_bool("ToyotaVirtualCruiseSpeed", False)
ui_state.params.put_bool("OnroadCycleRequested", True)
def update_settings(self):
if ui_state.CP is not None:
longitudinal = ui_state.CP.openpilotLongitudinalControl
enforce_stock = self.enforce_stock_longitudinal.action_item.get_state()
virtual_cruise_available = bool(ui_state.CP_SP is not None and
ui_state.CP_SP.flags & ToyotaFlagsSP.VIRTUAL_CRUISE_SPEED_AVAILABLE)
if longitudinal and virtual_cruise_available:
self.virtual_cruise_speed.action_item.set_enabled(not ui_state.engaged)
virtual_cruise_desc = tr(DESCRIPTIONS["virtual_cruise_speed"])
show_virtual_cruise_desc = False
else:
self.virtual_cruise_speed.action_item.set_enabled(False)
if self.virtual_cruise_speed.action_item.get_state():
self.virtual_cruise_speed.action_item.set_state(False)
ui_state.params.put_bool("ToyotaVirtualCruiseSpeed", False)
virtual_cruise_desc = "<b>" + tr(VIRTUAL_CRUISE_UNAVAILABLE) + "</b>\n\n" + tr(DESCRIPTIONS["virtual_cruise_speed"])
show_virtual_cruise_desc = True
if self.virtual_cruise_speed.description != virtual_cruise_desc:
self.virtual_cruise_speed.set_description(virtual_cruise_desc)
if show_virtual_cruise_desc:
self.virtual_cruise_speed.show_description(True)
if longitudinal and not enforce_stock:
self.stop_and_go_hack.action_item.set_enabled(not ui_state.engaged)
@@ -114,6 +170,12 @@ class ToyotaSettings(BrandSettings):
if show_desc:
self.stop_and_go_hack.show_description(True)
else:
self.virtual_cruise_speed.action_item.set_enabled(False)
virtual_cruise_desc = "<b>" + tr(ONROAD_ONLY_DESCRIPTION) + "</b>\n\n" + tr(DESCRIPTIONS["virtual_cruise_speed"])
if self.virtual_cruise_speed.description != virtual_cruise_desc:
self.virtual_cruise_speed.set_description(virtual_cruise_desc)
self.virtual_cruise_speed.show_description(True)
self.stop_and_go_hack.action_item.set_enabled(False)
new_desc = "<b>" + tr(ONROAD_ONLY_DESCRIPTION) + "</b>\n\n" + tr(DESCRIPTIONS["stop_and_go_hack"])
if self.stop_and_go_hack.description != new_desc:
@@ -115,7 +115,7 @@ class IntelligentCruiseButtonManagement:
self.is_ready = ready and not button_pressed
def run(self, CS: car.CarState, CC: car.CarControl, LP_SP: custom.LongitudinalPlanSP, is_metric: bool) -> None:
if self.CP_SP.pcmCruiseSpeed:
if self.CP_SP.pcmCruiseSpeed or not self.CP_SP.intelligentCruiseButtonManagementAvailable:
return
self.is_metric = is_metric
@@ -139,6 +139,7 @@ def initialize_params(params) -> list[dict[str, Any]]:
"ToyotaTSS2Long",
"ToyotaEnhancedBsm",
"ToyotaAutoHold",
"ToyotaVirtualCruiseSpeed",
])
return [{k: params.get(k, return_default=True)} for k in keys]
@@ -1,9 +1,13 @@
from opendbc.can.parser import CANParser
from opendbc.car import create_button_events
from opendbc.car.structs import car
from opendbc.car.toyota.carstate import get_virtual_cruise_button, VIRTUAL_CRUISE_BUTTONS
from openpilot.cereal import custom
from openpilot.common.constants import CV
from openpilot.common.parameterized import parameterized, parameterized_class
from openpilot.common.params import Params
from openpilot.common.test import OpenpilotTestCase
from openpilot.selfdrive.car.cruise import V_CRUISE_INITIAL
from openpilot.selfdrive.car.cruise import TOYOTA_VIRTUAL_CRUISE_LONG_PRESS, VCruiseHelper, V_CRUISE_INITIAL, V_CRUISE_UNSET
from openpilot.selfdrive.car.tests.test_cruise_speed import TestVCruiseHelper
from openpilot.sunnypilot.selfdrive.car.interfaces import initialize_params
@@ -122,8 +126,8 @@ class TestCustomAccIncrements(TestVCruiseHelper):
def test_rounding_behavior(self):
"""Test rounding behavior for 5 and 10 increments"""
test_cases = [
(47, 5, 50), # 47 -> 50 (round up to next 5)
(45, 5, 50), # 45 -> 50 (already at 5, increment by 5)
(47, 5, 50), # 47 -> 50 (round up to next 5)
(45, 5, 50), # 45 -> 50 (already at 5, increment by 5)
(43, 10, 50), # 43 -> 50 (round up to next 10)
(40, 10, 50), # 40 -> 50 (already at 10, increment by 10)
]
@@ -154,3 +158,302 @@ class TestCustomAccIncrements(TestVCruiseHelper):
initial_speed = self.v_cruise_helper.v_cruise_kph
self.press_button_long(ButtonType.accelCruise)
assert self.v_cruise_helper.v_cruise_kph == initial_speed + 10 # Should fallback to 10
class TestToyotaVirtualCruiseSpeed(OpenpilotTestCase):
def setup_method(self):
self.params = Params()
self.params.put_bool("CustomAccIncrementsEnabled", True, block=True)
self.params.put("CustomAccShortPressIncrement", 5, block=True)
self.params.put("CustomAccLongPressIncrement", 5, block=True)
CP = car.CarParams(brand="toyota", pcmCruise=True, openpilotLongitudinalControl=True)
CP_SP = custom.CarParamsSP(pcmCruiseSpeed=False)
self.v_cruise_helper = VCruiseHelper(CP, CP_SP)
self.v_cruise_helper.read_custom_set_speed_params()
self.route_parser = CANParser("toyota_nodsu_pt_generated", [("CLUTCH", 16)], 0)
self.route_button = 0
@staticmethod
def car_state(canonical_kph, cluster_kph, *, available=True, standstill=False, gas_pressed=False, v_ego_kph=0.0, button_events=None):
CS = car.CarState(
gasPressed=gas_pressed,
vEgo=v_ego_kph * CV.KPH_TO_MS,
cruiseState={
"available": available,
"speed": canonical_kph * CV.KPH_TO_MS,
"speedCluster": cluster_kph * CV.KPH_TO_MS,
"standstill": standstill,
},
)
CS.buttonEvents = button_events or []
return CS
def seed_enabled(self, canonical_kph, cluster_kph, *, is_metric=True):
CS = self.car_state(canonical_kph, cluster_kph)
self.v_cruise_helper.update_v_cruise(CS, enabled=False, is_metric=is_metric)
self.v_cruise_helper.update_v_cruise(CS, enabled=True, is_metric=is_metric)
self.v_cruise_helper.update_v_cruise(CS, enabled=True, is_metric=is_metric)
assert self.v_cruise_helper.v_cruise_kph == canonical_kph
assert self.v_cruise_helper.v_cruise_cluster_kph == cluster_kph
def press(self, button_type, canonical_kph, cluster_kph, hold_frames=0, *, standstill=False, gas_pressed=False, v_ego_kph=0.0, is_metric=True):
pressed = [ButtonEvent(type=button_type, pressed=True)]
self.v_cruise_helper.update_v_cruise(
self.car_state(canonical_kph, cluster_kph, standstill=standstill, gas_pressed=gas_pressed, v_ego_kph=v_ego_kph, button_events=pressed),
enabled=True,
is_metric=is_metric,
)
for _ in range(hold_frames):
self.v_cruise_helper.update_v_cruise(
self.car_state(canonical_kph, cluster_kph, standstill=standstill, gas_pressed=gas_pressed, v_ego_kph=v_ego_kph),
enabled=True,
is_metric=is_metric,
)
released = [ButtonEvent(type=button_type, pressed=False)]
self.v_cruise_helper.update_v_cruise(
self.car_state(canonical_kph, cluster_kph, standstill=standstill, gas_pressed=gas_pressed, v_ego_kph=v_ego_kph, button_events=released),
enabled=True,
is_metric=is_metric,
)
def set_increments(self, short_increment, long_increment):
self.params.put("CustomAccShortPressIncrement", short_increment, block=True)
self.params.put("CustomAccLongPressIncrement", long_increment, block=True)
self.v_cruise_helper.read_custom_set_speed_params()
def assert_kph_almost_equal(self, actual, expected):
self.assertAlmostEqual(actual, expected, delta=abs(expected) * 1e-6)
def route_button_events(self, payload):
self.route_parser.update((1, [(0x361, bytes.fromhex(payload), 0)]))
current = get_virtual_cruise_button(
self.route_parser.vl["CLUTCH"]["CRUISE_RES"],
self.route_parser.vl["CLUTCH"]["CRUISE_SET"],
)
events = create_button_events(current, self.route_button, VIRTUAL_CRUISE_BUTTONS)
self.route_button = current
return events
def test_short_press_rounds_display_target_and_preserves_offset(self):
self.seed_enabled(27, 31)
self.press(ButtonType.accelCruise, 28, 32)
assert self.v_cruise_helper.v_cruise_kph == 31
assert self.v_cruise_helper.v_cruise_cluster_kph == 35
def test_decel_at_display_minimum_does_not_increase_target(self):
self.seed_enabled(26, 30)
self.press(ButtonType.decelCruise, 25, 29)
assert self.v_cruise_helper.v_cruise_kph == 26
assert self.v_cruise_helper.v_cruise_cluster_kph == 30
@parameterized.expand((52, TOYOTA_VIRTUAL_CRUISE_LONG_PRESS - 1))
def test_route_length_short_press_is_not_a_long_press(self, hold_frames):
self.set_increments(short_increment=2, long_increment=5)
self.seed_enabled(27, 31)
self.press(ButtonType.accelCruise, 28, 32, hold_frames=hold_frames)
assert self.v_cruise_helper.v_cruise_kph == 29
assert self.v_cruise_helper.v_cruise_cluster_kph == 33
def test_toyota_long_press_uses_route_validated_cadence_and_suppresses_release(self):
self.set_increments(short_increment=2, long_increment=5)
self.seed_enabled(27, 31)
pressed = [ButtonEvent(type=ButtonType.accelCruise, pressed=True)]
self.v_cruise_helper.update_v_cruise(self.car_state(31, 35, button_events=pressed), enabled=True, is_metric=True)
for _ in range(TOYOTA_VIRTUAL_CRUISE_LONG_PRESS):
self.v_cruise_helper.update_v_cruise(self.car_state(31, 35), enabled=True, is_metric=True)
assert self.v_cruise_helper.v_cruise_kph == 31
assert self.v_cruise_helper.v_cruise_cluster_kph == 35
released = [ButtonEvent(type=ButtonType.accelCruise, pressed=False)]
self.v_cruise_helper.update_v_cruise(self.car_state(31, 35, button_events=released), enabled=True, is_metric=True)
assert self.v_cruise_helper.v_cruise_kph == 31
assert self.v_cruise_helper.v_cruise_cluster_kph == 35
def test_route_4_32_second_hold_repeats_six_times(self):
self.seed_enabled(26, 30)
self.press(ButtonType.accelCruise, 30, 34, hold_frames=432)
assert self.v_cruise_helper.v_cruise_kph == 56
assert self.v_cruise_helper.v_cruise_cluster_kph == 60
def test_maximum_boundary_caps_pair_and_preserves_offset(self):
self.seed_enabled(141, 145)
self.press(ButtonType.accelCruise, 142, 146)
assert self.v_cruise_helper.v_cruise_kph == 141
assert self.v_cruise_helper.v_cruise_cluster_kph == 145
self.press(ButtonType.accelCruise, 143, 147)
assert self.v_cruise_helper.v_cruise_kph == 141
assert self.v_cruise_helper.v_cruise_cluster_kph == 145
@parameterized.expand(
(
(25, 29, ButtonType.decelCruise),
(141, 147, ButtonType.accelCruise),
)
)
def test_out_of_range_raw_pair_is_not_moved_in_opposite_direction(self, canonical_kph, cluster_kph, button_type):
self.seed_enabled(canonical_kph, cluster_kph)
self.press(button_type, canonical_kph, cluster_kph)
assert self.v_cruise_helper.v_cruise_kph == canonical_kph
assert self.v_cruise_helper.v_cruise_cluster_kph == cluster_kph
def test_imperial_increment_preserves_canonical_cluster_pair(self):
self.seed_enabled(45, 50, is_metric=False)
self.press(ButtonType.accelCruise, 46, 51, is_metric=False)
assert self.v_cruise_helper.v_cruise_kph == 51
assert self.v_cruise_helper.v_cruise_cluster_kph == 56
def test_engagement_button_held_does_not_change_target(self):
initial = self.car_state(27, 31)
self.v_cruise_helper.update_v_cruise(initial, enabled=False, is_metric=True)
pressed = [ButtonEvent(type=ButtonType.decelCruise, pressed=True)]
self.v_cruise_helper.update_v_cruise(self.car_state(27, 31, button_events=pressed), enabled=False, is_metric=True)
for _ in range(TOYOTA_VIRTUAL_CRUISE_LONG_PRESS + 10):
self.v_cruise_helper.update_v_cruise(self.car_state(28, 32), enabled=True, is_metric=True)
released = [ButtonEvent(type=ButtonType.decelCruise, pressed=False)]
self.v_cruise_helper.update_v_cruise(self.car_state(28, 32, button_events=released), enabled=True, is_metric=True)
self.v_cruise_helper.update_v_cruise(self.car_state(28, 32), enabled=True, is_metric=True)
assert self.v_cruise_helper.v_cruise_kph == 28
assert self.v_cruise_helper.v_cruise_cluster_kph == 32
def test_delayed_pcm_target_seeds_before_software_ownership(self):
invalid = self.car_state(0, 0)
self.v_cruise_helper.update_v_cruise(invalid, enabled=False, is_metric=True)
release = [ButtonEvent(type=ButtonType.decelCruise, pressed=False)]
for _ in range(4):
self.v_cruise_helper.update_v_cruise(self.car_state(0, 0, button_events=release), enabled=True, is_metric=True)
assert self.v_cruise_helper.v_cruise_kph == V_CRUISE_UNSET
assert self.v_cruise_helper.v_cruise_cluster_kph == V_CRUISE_UNSET
self.v_cruise_helper.update_v_cruise(self.car_state(27, 31), enabled=True, is_metric=True)
self.assert_kph_almost_equal(self.v_cruise_helper.v_cruise_kph, 27)
self.assert_kph_almost_equal(self.v_cruise_helper.v_cruise_cluster_kph, 31)
self.v_cruise_helper.update_v_cruise(self.car_state(28, 32), enabled=True, is_metric=True)
self.assert_kph_almost_equal(self.v_cruise_helper.v_cruise_kph, 27)
self.assert_kph_almost_equal(self.v_cruise_helper.v_cruise_cluster_kph, 31)
def test_route_payload_short_press_drives_virtual_target(self):
self.seed_enabled(27, 31)
pressed = self.route_button_events("a61a0000561a1a81")
self.v_cruise_helper.update_v_cruise(self.car_state(27, 31, button_events=pressed), enabled=True, is_metric=True)
for _ in range(52):
self.v_cruise_helper.update_v_cruise(self.car_state(28, 32), enabled=True, is_metric=True)
released = self.route_button_events("861a0000561b1a81")
self.v_cruise_helper.update_v_cruise(self.car_state(28, 32, button_events=released), enabled=True, is_metric=True)
assert self.v_cruise_helper.v_cruise_kph == 31
assert self.v_cruise_helper.v_cruise_cluster_kph == 35
def test_prius_route_payload_short_set_drives_virtual_target(self):
self.seed_enabled(31, 35)
pressed = self.route_button_events("965f000056666585")
self.v_cruise_helper.update_v_cruise(self.car_state(31, 35, button_events=pressed), enabled=True, is_metric=True)
for _ in range(45):
self.v_cruise_helper.update_v_cruise(self.car_state(30, 34), enabled=True, is_metric=True)
released = self.route_button_events("865f000056666585")
self.v_cruise_helper.update_v_cruise(self.car_state(30, 34, button_events=released), enabled=True, is_metric=True)
assert self.v_cruise_helper.v_cruise_kph == 26
assert self.v_cruise_helper.v_cruise_cluster_kph == 30
def test_prius_route_payload_standstill_res_does_not_change_target(self):
self.seed_enabled(27, 31)
pressed = self.route_button_events("a61b0000561c1c80")
self.v_cruise_helper.update_v_cruise(
self.car_state(27, 31, standstill=True, button_events=pressed),
enabled=True,
is_metric=True,
)
for _ in range(TOYOTA_VIRTUAL_CRUISE_LONG_PRESS):
self.v_cruise_helper.update_v_cruise(self.car_state(27, 31, standstill=True), enabled=True, is_metric=True)
released = self.route_button_events("865f000056666585")
self.v_cruise_helper.update_v_cruise(
self.car_state(27, 31, standstill=True, button_events=released),
enabled=True,
is_metric=True,
)
assert self.v_cruise_helper.v_cruise_kph == 27
assert self.v_cruise_helper.v_cruise_cluster_kph == 31
def test_route_payload_disengage_mid_hold_clears_pending_action(self):
self.seed_enabled(27, 31)
pressed = self.route_button_events("a61a0000561a1a81")
self.v_cruise_helper.update_v_cruise(self.car_state(27, 31, button_events=pressed), enabled=True, is_metric=True)
for _ in range(30):
self.v_cruise_helper.update_v_cruise(self.car_state(28, 32), enabled=True, is_metric=True)
self.v_cruise_helper.update_v_cruise(self.car_state(28, 32), enabled=False, is_metric=True)
released = self.route_button_events("861a0000561b1a81")
self.v_cruise_helper.update_v_cruise(self.car_state(28, 32, available=False, button_events=released), enabled=False, is_metric=True)
self.v_cruise_helper.update_v_cruise(self.car_state(28, 32), enabled=False, is_metric=True)
self.v_cruise_helper.update_v_cruise(self.car_state(28, 32), enabled=True, is_metric=True)
self.v_cruise_helper.update_v_cruise(self.car_state(28, 32), enabled=True, is_metric=True)
self.assert_kph_almost_equal(self.v_cruise_helper.v_cruise_kph, 28)
self.assert_kph_almost_equal(self.v_cruise_helper.v_cruise_cluster_kph, 32)
def test_standstill_resume_does_not_change_target(self):
self.seed_enabled(27, 31)
self.press(ButtonType.accelCruise, 27, 31, standstill=True)
assert self.v_cruise_helper.v_cruise_kph == 27
assert self.v_cruise_helper.v_cruise_cluster_kph == 31
def test_disengagement_discards_virtual_target_and_reseeds_raw_pair(self):
self.seed_enabled(27, 31)
self.press(ButtonType.accelCruise, 28, 32)
assert self.v_cruise_helper.v_cruise_kph == 31
raw = self.car_state(28, 32)
self.v_cruise_helper.update_v_cruise(raw, enabled=False, is_metric=True)
self.assert_kph_almost_equal(self.v_cruise_helper.v_cruise_kph, 28)
self.assert_kph_almost_equal(self.v_cruise_helper.v_cruise_cluster_kph, 32)
self.v_cruise_helper.update_v_cruise(raw, enabled=True, is_metric=True)
self.v_cruise_helper.update_v_cruise(raw, enabled=True, is_metric=True)
self.assert_kph_almost_equal(self.v_cruise_helper.v_cruise_kph, 28)
self.assert_kph_almost_equal(self.v_cruise_helper.v_cruise_cluster_kph, 32)
def test_unavailable_and_mads_handback_discard_virtual_target(self):
self.seed_enabled(27, 31)
self.press(ButtonType.accelCruise, 28, 32)
assert self.v_cruise_helper.v_cruise_kph == 31
self.v_cruise_helper.update_v_cruise(self.car_state(28, 32), enabled=False, is_metric=True)
self.assert_kph_almost_equal(self.v_cruise_helper.v_cruise_kph, 28)
self.assert_kph_almost_equal(self.v_cruise_helper.v_cruise_cluster_kph, 32)
self.v_cruise_helper.update_v_cruise(self.car_state(0, 0, available=False), enabled=False, is_metric=True)
assert self.v_cruise_helper.v_cruise_kph == V_CRUISE_UNSET
assert self.v_cruise_helper.v_cruise_cluster_kph == V_CRUISE_UNSET
self.v_cruise_helper.update_v_cruise(self.car_state(29, 33), enabled=False, is_metric=True)
self.assert_kph_almost_equal(self.v_cruise_helper.v_cruise_kph, 29)
self.assert_kph_almost_equal(self.v_cruise_helper.v_cruise_cluster_kph, 33)
def test_set_during_gas_override_clips_target_to_ego_speed(self):
self.seed_enabled(27, 31)
self.press(ButtonType.decelCruise, 26, 30, gas_pressed=True, v_ego_kph=50)
assert self.v_cruise_helper.v_cruise_kph == 50
assert self.v_cruise_helper.v_cruise_cluster_kph == 54
@@ -11,6 +11,7 @@ from opendbc.car.structs import car
from opendbc.car.hyundai.values import CAR as HYUNDAI_CAR, UNSUPPORTED_LONGITUDINAL_CAR
from opendbc.car.subaru.values import CAR as SUBARU_CAR, SubaruFlags
from opendbc.sunnypilot.car.tesla.values import TeslaFlagsSP
from opendbc.sunnypilot.car.toyota.values import ToyotaFlagsSP, VIRTUAL_CRUISE_SPEED_CAR
from openpilot.common.params import Params
from openpilot.common.swaglog import cloudlog
from openpilot.common.hardware import HARDWARE
@@ -19,6 +20,7 @@ from openpilot.common.hardware import HARDWARE
# Wire-protocol version for the capabilities payload. Bump on breaking changes
# only; additive fields are backward-compatible and do not require a bump.
PROTOCOL_VERSION = 1
TOYOTA_VIRTUAL_CRUISE_SPEED_PLATFORMS = {str(platform) for platform in VIRTUAL_CRUISE_SPEED_CAR}
# All capability fields that rules may reference.
# Non-boolean fields must have defaults in CAPABILITY_DEFAULTS.
@@ -42,6 +44,7 @@ CAPABILITY_FIELDS = (
"device_type",
"subaru_has_sng",
"hyundai_alpha_long_available",
"toyota_virtual_cruise_speed_available",
)
CAPABILITY_LABELS: dict[str, str] = {
@@ -64,6 +67,7 @@ CAPABILITY_LABELS: dict[str, str] = {
"device_type": "Device type",
"subaru_has_sng": "Subaru Stop-and-Go available",
"hyundai_alpha_long_available": "Hyundai Alpha Longitudinal available",
"toyota_virtual_cruise_speed_available": "Toyota Virtual Cruise Speed available",
}
# Explicit defaults for non-boolean capability fields
@@ -110,6 +114,12 @@ def _resolve_brand_capabilities(caps: dict, bundle_platform: str, CP) -> None:
caps["subaru_has_sng"] = not bool(CP.flags & (SubaruFlags.GLOBAL_GEN2 | SubaruFlags.HYBRID))
caps["has_stop_and_go"] = caps["subaru_has_sng"]
elif brand == "toyota":
if bundle_platform:
caps["toyota_virtual_cruise_speed_available"] = bundle_platform in TOYOTA_VIRTUAL_CRUISE_SPEED_PLATFORMS
elif CP is not None:
caps["toyota_virtual_cruise_speed_available"] = str(CP.carFingerprint) in TOYOTA_VIRTUAL_CRUISE_SPEED_PLATFORMS
def generate_capabilities(params: Params | None = None) -> dict:
"""Generate a SettingsCapabilities dict from CarParams + boolean params.
@@ -174,6 +184,8 @@ def generate_capabilities(params: Params | None = None) -> dict:
caps["icbm_available"] = bool(CP_SP.intelligentCruiseButtonManagementAvailable)
caps["has_icbm"] = bool(CP_SP.intelligentCruiseButtonManagementAvailable) and params.get_bool("IntelligentCruiseButtonManagement")
caps["tesla_has_vehicle_bus"] = bool(CP_SP.flags & TeslaFlagsSP.HAS_VEHICLE_BUS)
if caps["brand"] == "toyota":
caps["toyota_virtual_cruise_speed_available"] = bool(CP_SP.flags & ToyotaFlagsSP.VIRTUAL_CRUISE_SPEED_AVAILABLE)
except Exception:
cloudlog.exception("capabilities: failed to deserialize CarParamsSPPersistent")
@@ -764,6 +764,21 @@
"type": "capability",
"field": "has_icbm",
"equals": true
},
{
"type": "all",
"conditions": [
{
"type": "capability",
"field": "toyota_virtual_cruise_speed_available",
"equals": true
},
{
"type": "param",
"key": "ToyotaVirtualCruiseSpeed",
"equals": true
}
]
}
]
}
@@ -802,6 +817,21 @@
"type": "capability",
"field": "has_icbm",
"equals": true
},
{
"type": "all",
"conditions": [
{
"type": "capability",
"field": "toyota_virtual_cruise_speed_available",
"equals": true
},
{
"type": "param",
"key": "ToyotaVirtualCruiseSpeed",
"equals": true
}
]
}
]
}
@@ -2423,6 +2453,40 @@
"enablement": [
{
"type": "not_engaged"
},
{
"type": "param",
"key": "ToyotaVirtualCruiseSpeed",
"equals": false
}
]
},
{
"key": "ToyotaVirtualCruiseSpeed",
"widget": "toggle",
"needs_onroad_cycle": true,
"title": "Toyota: Virtual Cruise Speed (Alpha)",
"description": "Uses a sunnypilot-owned cruise target with the Toyota RES/SET buttons and unlocks Custom ACC Speed Intervals. Set the short interval to 5 for next-5-unit tap behavior. The Toyota cluster continues to show the factory target and may differ from sunnypilot. The direct button signals are route-validated on Corolla Cross and Prius TSS2, but held-button timing differs by platform. Validate acceleration above the factory target in a controlled setting.",
"visibility": [
{
"type": "capability",
"field": "toyota_virtual_cruise_speed_available",
"equals": true
}
],
"enablement": [
{
"type": "not_engaged"
},
{
"type": "capability",
"field": "has_longitudinal_control",
"equals": true
},
{
"type": "param",
"key": "ToyotaEnforceStockLongitudinal",
"equals": false
}
]
},
@@ -103,6 +103,14 @@ sections:
- type: capability
field: has_icbm
equals: true
- type: all
conditions:
- type: capability
field: toyota_virtual_cruise_speed_available
equals: true
- type: param
key: ToyotaVirtualCruiseSpeed
equals: true
items:
- key: CustomAccIncrementsEnabled
widget: toggle
@@ -124,6 +132,14 @@ sections:
- type: capability
field: has_icbm
equals: true
- type: all
conditions:
- type: capability
field: toyota_virtual_cruise_speed_available
equals: true
- type: param
key: ToyotaVirtualCruiseSpeed
equals: true
sub_panels:
- id: custom_acc_intervals
label: Custom ACC Speed Intervals Settings
@@ -113,6 +113,28 @@ sections:
description: sunnypilot will not take over control of gas and brakes. Factory Toyota longitudinal control will be used.
enablement:
- $ref: '#/macros/not_engaged'
- type: param
key: ToyotaVirtualCruiseSpeed
equals: false
- key: ToyotaVirtualCruiseSpeed
widget: toggle
needs_onroad_cycle: true
title: 'Toyota: Virtual Cruise Speed (Alpha)'
description: Uses a sunnypilot-owned cruise target with the Toyota RES/SET buttons and unlocks Custom ACC Speed
Intervals. Set the short interval to 5 for next-5-unit tap behavior. The Toyota cluster continues to show the
factory target and may differ from sunnypilot. The direct button signals are route-validated on Corolla Cross
and Prius TSS2, but held-button timing differs by platform. Validate acceleration above the factory target in a
controlled setting.
visibility:
- type: capability
field: toyota_virtual_cruise_speed_available
equals: true
enablement:
- $ref: '#/macros/not_engaged'
- $ref: '#/macros/longitudinal'
- type: param
key: ToyotaEnforceStockLongitudinal
equals: false
- key: ToyotaStopAndGoHack
widget: toggle
needs_onroad_cycle: true
@@ -10,9 +10,18 @@ change and must be intentional. KNOWN_PROTOCOL_VERSIONS pins the set we
explicitly support — when the constant is bumped, this list must be edited in
the same commit so the bump shows up in code review.
"""
from __future__ import annotations
from typing import cast
from openpilot.cereal import custom
from openpilot.common.params import Params
from openpilot.common.parameterized import parameterized
from openpilot.common.test import OpenpilotTestCase
from opendbc.car.structs import car
from opendbc.car.toyota.values import CAR as TOYOTA_CAR
from opendbc.sunnypilot.car.toyota.values import ToyotaFlagsSP
from openpilot.sunnypilot.sunnylink.capabilities import (
CAPABILITY_DEFAULTS,
CAPABILITY_FIELDS,
@@ -20,13 +29,41 @@ from openpilot.sunnypilot.sunnylink.capabilities import (
PROTOCOL_VERSION,
generate_capabilities,
)
from openpilot.common.test import OpenpilotTestCase
KNOWN_PROTOCOL_VERSIONS = (1,)
LATEST_KNOWN = max(KNOWN_PROTOCOL_VERSIONS)
class FakeParams:
def __init__(self, values=None):
self.values = values or {}
def get(self, key, *args, **kwargs):
return self.values.get(key)
def get_bool(self, key):
return bool(self.values.get(key, False))
def build_persistent_toyota_params(platform, *, sp_flags=0) -> Params:
CP = car.CarParams.new_message()
CP.brand = "toyota"
CP.carFingerprint = str(platform)
CP.pcmCruise = True
CP.openpilotLongitudinalControl = True
CP_SP = custom.CarParamsSP.new_message()
CP_SP.flags = int(sp_flags)
return cast(Params, FakeParams(
{
"CarParamsPersistent": CP.to_bytes(),
"CarParamsSPPersistent": CP_SP.to_bytes(),
}
))
def caps():
return generate_capabilities()
@@ -52,14 +89,12 @@ class TestProtocolVersion(OpenpilotTestCase):
def test_protocol_version_is_known(self):
"""Sentinel against accidental bumps. Edit KNOWN_PROTOCOL_VERSIONS if intentional."""
assert PROTOCOL_VERSION in KNOWN_PROTOCOL_VERSIONS, (
f"PROTOCOL_VERSION={PROTOCOL_VERSION} is not in KNOWN_PROTOCOL_VERSIONS={KNOWN_PROTOCOL_VERSIONS}. " +
"If this bump is intentional, add it to KNOWN_PROTOCOL_VERSIONS."
f"PROTOCOL_VERSION={PROTOCOL_VERSION} is not in KNOWN_PROTOCOL_VERSIONS={KNOWN_PROTOCOL_VERSIONS}. "
+ "If this bump is intentional, add it to KNOWN_PROTOCOL_VERSIONS."
)
def test_protocol_version_matches_latest_known(self):
assert PROTOCOL_VERSION == LATEST_KNOWN, (
"Test invariant: PROTOCOL_VERSION must equal max(KNOWN_PROTOCOL_VERSIONS)."
)
assert PROTOCOL_VERSION == LATEST_KNOWN, "Test invariant: PROTOCOL_VERSION must equal max(KNOWN_PROTOCOL_VERSIONS)."
class TestOpaquePerBrandFlags(OpenpilotTestCase):
@@ -76,6 +111,66 @@ class TestOpaquePerBrandFlags(OpenpilotTestCase):
assert caps["hyundai_alpha_long_available"] is False
class TestToyotaVirtualCruiseSpeedCapability(OpenpilotTestCase):
def test_field_present_and_labeled(self):
assert "toyota_virtual_cruise_speed_available" in CAPABILITY_FIELDS
assert "toyota_virtual_cruise_speed_available" in CAPABILITY_LABELS
def test_default_false(self):
caps = generate_capabilities(cast(Params, FakeParams()))
assert caps["toyota_virtual_cruise_speed_available"] is False
@parameterized.expand(
(
(TOYOTA_CAR.TOYOTA_COROLLA_TSS2, True),
(TOYOTA_CAR.TOYOTA_PRIUS_TSS2, True),
(TOYOTA_CAR.TOYOTA_RAV4_TSS2, False),
)
)
def test_bundle_platform_gating(self, platform, expected):
params = FakeParams(
{
"CarPlatformBundle": {
"brand": "toyota",
"platform": str(platform),
},
}
)
caps = generate_capabilities(cast(Params, params))
assert caps["toyota_virtual_cruise_speed_available"] is expected
@parameterized.expand(
(
(TOYOTA_CAR.TOYOTA_COROLLA_TSS2, ToyotaFlagsSP.VIRTUAL_CRUISE_SPEED_AVAILABLE, True),
(TOYOTA_CAR.TOYOTA_COROLLA_TSS2, 0, True),
(TOYOTA_CAR.TOYOTA_PRIUS_TSS2, ToyotaFlagsSP.VIRTUAL_CRUISE_SPEED_AVAILABLE, True),
(TOYOTA_CAR.TOYOTA_PRIUS_TSS2, 0, True),
(TOYOTA_CAR.TOYOTA_RAV4_TSS2, ToyotaFlagsSP.VIRTUAL_CRUISE_SPEED_AVAILABLE, False),
)
)
def test_persistent_car_params_platform_gating(self, platform, sp_flags, expected):
caps = generate_capabilities(build_persistent_toyota_params(platform, sp_flags=sp_flags))
assert caps["toyota_virtual_cruise_speed_available"] is expected
@parameterized.expand(
(
(TOYOTA_CAR.TOYOTA_COROLLA_TSS2, TOYOTA_CAR.TOYOTA_RAV4_TSS2, True),
(TOYOTA_CAR.TOYOTA_PRIUS_TSS2, TOYOTA_CAR.TOYOTA_RAV4_TSS2, True),
(TOYOTA_CAR.TOYOTA_RAV4_TSS2, TOYOTA_CAR.TOYOTA_COROLLA_TSS2, False),
(TOYOTA_CAR.TOYOTA_RAV4_TSS2, TOYOTA_CAR.TOYOTA_PRIUS_TSS2, False),
)
)
def test_bundle_platform_takes_precedence_over_stale_persistent_params(self, bundle_platform, persistent_platform, expected):
params = build_persistent_toyota_params(persistent_platform, sp_flags=ToyotaFlagsSP.VIRTUAL_CRUISE_SPEED_AVAILABLE)
params.values["CarPlatformBundle"] = {
"brand": "toyota",
"platform": str(bundle_platform),
}
caps = generate_capabilities(params)
assert caps["toyota_virtual_cruise_speed_available"] is expected
class TestCapabilitiesShape(OpenpilotTestCase):
def test_all_fields_present(self, caps):
for field in CAPABILITY_FIELDS:
@@ -9,6 +9,7 @@ isolates one of the gating bugs that the design-overhaul branch fixes so a
future regression is loud and obvious. These tests are intentionally narrow
and additive — they do not replace the broader test_settings_schema.py.
"""
from __future__ import annotations
import json
@@ -24,14 +25,13 @@ from openpilot.sunnypilot.sunnylink.tools.generate_settings_schema import (
_load_torque_versions,
generate_schema,
)
from openpilot.sunnypilot.sunnylink.tools.validate_settings_ui import validate as validate_settings_ui
from openpilot.common.test import OpenpilotTestCase
SCHEMA_VALIDATOR_PATH = os.path.join(os.path.dirname(DEFINITION_PATH), "settings_ui.schema.json")
def _walk_items(schema: dict[str, Any]):
"""Yield every item dict from the schema."""
def _yield(item: dict[str, Any]):
yield item
for sub in item.get("sub_items", []):
@@ -106,6 +106,26 @@ def _references_capability_field(rules: list[dict[str, Any]] | None, field: str)
return found
def _has_toyota_virtual_cruise_gate(rules: list[dict[str, Any]] | None) -> bool:
def _walk(rule: dict[str, Any]) -> bool:
if rule.get("type") == "all":
conditions = rule.get("conditions", [])
has_capability = any(
c.get("type") == "capability" and c.get("field") == "toyota_virtual_cruise_speed_available" and c.get("equals") is True for c in conditions
)
has_param = any(c.get("type") == "param" and c.get("key") == "ToyotaVirtualCruiseSpeed" and c.get("equals") is True for c in conditions)
if has_capability and has_param:
return True
if rule.get("type") == "not" and "condition" in rule:
return _walk(rule["condition"])
if rule.get("type") in ("any", "all"):
return any(_walk(c) for c in rule.get("conditions", []))
return False
return any(_walk(rule) for rule in rules or [])
def schema():
return generate_schema()
@@ -149,22 +169,13 @@ class TestTestManeuversSection(OpenpilotTestCase):
assert "is_sp_release" in vis_refs
enablement = section.get("enablement") or []
enable_refs = json.dumps(enablement)
assert "ShowAdvancedControls" in enable_refs, \
"test_maneuvers must gate ShowAdvancedControls via enablement"
assert "ShowAdvancedControls" in enable_refs, "test_maneuvers must gate ShowAdvancedControls via enablement"
class TestValidator(OpenpilotTestCase):
def test_validator_accepts_real_json(self):
"""settings_ui.json validates against settings_ui.schema.json."""
try:
import jsonschema
except ImportError:
self.skipTest("jsonschema not installed")
with open(DEFINITION_PATH) as f:
data = json.load(f)
with open(SCHEMA_VALIDATOR_PATH) as f:
validator = json.load(f)
jsonschema.validate(instance=data, schema=validator)
"""settings_ui.json passes the repository's production schema validator."""
self.assertTrue(validate_settings_ui(DEFINITION_PATH))
class TestTorqueOptionGeneration(OpenpilotTestCase):
@@ -177,16 +188,17 @@ class TestTorqueOptionGeneration(OpenpilotTestCase):
assert item.get("options") == expected
def test_torque_versions_path_resolves(self):
assert os.path.exists(TORQUE_VERSIONS_PATH), (
f"latcontrol_torque_versions.json not found at {TORQUE_VERSIONS_PATH}"
)
assert os.path.exists(TORQUE_VERSIONS_PATH), f"latcontrol_torque_versions.json not found at {TORQUE_VERSIONS_PATH}"
class TestReleaseBranchGates(OpenpilotTestCase):
@parameterized.expand([
"EnableGithubRunner",
"QuickBootToggle",
], names=["key"])
@parameterized.expand(
[
"EnableGithubRunner",
"QuickBootToggle",
],
names=["key"],
)
def test_sp_dev_items_gate_on_is_sp_release(self, schema, key):
"""sunnypilot dev items must hide on sunnypilot release branches (is_sp_release gate)."""
item = _find_item(schema, key)
@@ -208,11 +220,14 @@ class TestSpuriousOffroadGatesDropped(OpenpilotTestCase):
class TestNotEngagedReplacement(OpenpilotTestCase):
@parameterized.expand([
"AlphaLongitudinalEnabled",
"ToyotaEnforceStockLongitudinal",
"ToyotaStopAndGoHack",
], names=["key"])
@parameterized.expand(
[
"AlphaLongitudinalEnabled",
"ToyotaEnforceStockLongitudinal",
"ToyotaStopAndGoHack",
],
names=["key"],
)
def test_offroad_only_replaced_with_not_engaged(self, schema, key):
"""These items should use not_engaged, not offroad_only."""
item = _find_item(schema, key)
@@ -220,3 +235,25 @@ class TestNotEngagedReplacement(OpenpilotTestCase):
rule_types = _flatten_rule_types(item.get("enablement"))
assert "offroad_only" not in rule_types, f"{key} still uses offroad_only"
assert "not_engaged" in rule_types, f"{key} missing not_engaged"
class TestToyotaVirtualCruiseSpeed(OpenpilotTestCase):
def test_vehicle_toggle_contract(self, schema):
toyota = schema["vehicle_settings"]["toyota"]
item = next((item for item in toyota["items"] if item.get("key") == "ToyotaVirtualCruiseSpeed"), None)
assert item is not None
assert item["widget"] == "toggle"
assert item.get("needs_onroad_cycle") is True
assert _references_capability_field(item.get("visibility"), "toyota_virtual_cruise_speed_available")
assert _references_capability_field(item.get("enablement"), "has_longitudinal_control")
assert "not_engaged" in _flatten_rule_types(item.get("enablement"))
def test_custom_acc_section_links_virtual_cruise_opt_in(self, schema):
section = _find_section(schema, "cruise", "custom_acc_increments")
assert section is not None
assert _has_toyota_virtual_cruise_gate(section.get("enablement"))
item = _find_item(schema, "CustomAccIncrementsEnabled")
assert item is not None
assert _has_toyota_virtual_cruise_gate(item.get("enablement"))
@@ -298,10 +298,11 @@ class TestKnownVehicleSettings(OpenpilotTestCase):
keys = {i["key"] for i in _brand_items(schema["vehicle_settings"].get("hyundai"))}
assert "HyundaiLongitudinalTuning" in keys
def test_toyota_has_enforce_stock_and_stop_go(self, schema):
def test_toyota_has_enforce_stock_stop_go_and_virtual_cruise(self, schema):
keys = {i["key"] for i in _brand_items(schema["vehicle_settings"].get("toyota"))}
assert "ToyotaEnforceStockLongitudinal" in keys
assert "ToyotaStopAndGoHack" in keys
assert "ToyotaVirtualCruiseSpeed" in keys
def test_tesla_has_coop_steering(self, schema):
keys = {i["key"] for i in _brand_items(schema["vehicle_settings"].get("tesla"))}