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

This commit is contained in:
rav4kumar
2026-08-15 01:06:35 -07:00
parent 89327eef45
commit 170eef8a2a
15 changed files with 675 additions and 14 deletions
+1
View File
@@ -234,6 +234,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
@@ -138,6 +138,7 @@ def initialize_params(params) -> list[dict[str, Any]]:
"ToyotaStopAndGoHack",
"ToyotaEnhancedBsm",
"ToyotaAutoHold",
"ToyotaVirtualCruiseSpeed",
])
return [{k: params.get(k, return_default=True)} for k in keys]
@@ -1,10 +1,14 @@
import pytest
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_class
from openpilot.common.params import Params
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
ButtonEvent = car.CarState.ButtonEvent
@@ -148,3 +152,290 @@ 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:
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., 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., 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 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
@pytest.mark.parametrize("hold_frames", (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
@pytest.mark.parametrize(("canonical_kph", "cluster_kph", "button_type"), (
(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)
assert self.v_cruise_helper.v_cruise_kph == pytest.approx(27)
assert self.v_cruise_helper.v_cruise_cluster_kph == pytest.approx(31)
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 == pytest.approx(27)
assert self.v_cruise_helper.v_cruise_cluster_kph == pytest.approx(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)
assert self.v_cruise_helper.v_cruise_kph == pytest.approx(28)
assert self.v_cruise_helper.v_cruise_cluster_kph == pytest.approx(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)
assert self.v_cruise_helper.v_cruise_kph == pytest.approx(28)
assert self.v_cruise_helper.v_cruise_cluster_kph == pytest.approx(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)
assert self.v_cruise_helper.v_cruise_kph == pytest.approx(28)
assert self.v_cruise_helper.v_cruise_cluster_kph == pytest.approx(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)
assert self.v_cruise_helper.v_cruise_kph == pytest.approx(28)
assert self.v_cruise_helper.v_cruise_cluster_kph == pytest.approx(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)
assert self.v_cruise_helper.v_cruise_kph == pytest.approx(29)
assert self.v_cruise_helper.v_cruise_cluster_kph == pytest.approx(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
@@ -14,6 +14,10 @@ from __future__ import annotations
import pytest
from openpilot.cereal import custom
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,
@@ -27,6 +31,33 @@ 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):
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 FakeParams({
"CarParamsPersistent": CP.to_bytes(),
"CarParamsSPPersistent": CP_SP.to_bytes(),
})
@pytest.fixture(scope="module")
def caps():
return generate_capabilities()
@@ -77,6 +108,58 @@ class TestOpaquePerBrandFlags:
assert caps["hyundai_alpha_long_available"] is False
class TestToyotaVirtualCruiseSpeedCapability:
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(FakeParams())
assert caps["toyota_virtual_cruise_speed_available"] is False
@pytest.mark.parametrize(("platform", "expected"), (
(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(params)
assert caps["toyota_virtual_cruise_speed_available"] is expected
@pytest.mark.parametrize(("platform", "sp_flags", "expected"), (
(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
@pytest.mark.parametrize(("bundle_platform", "persistent_platform", "expected"), (
(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:
def test_all_fields_present(self, caps):
for field in CAPABILITY_FIELDS:
@@ -105,6 +105,34 @@ 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 [])
@pytest.fixture(scope="module")
def schema():
return generate_schema()
@@ -217,3 +245,25 @@ class TestNotEngagedReplacement:
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:
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"))
@@ -300,10 +300,11 @@ class TestKnownVehicleSettings:
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"))}