Global Feature Availability

This commit is contained in:
firestar5683
2026-10-19 16:16:22 -05:00
parent fdd6627985
commit bea3b340c9
18 changed files with 647 additions and 52 deletions
+4 -5
View File
@@ -35,7 +35,7 @@ from openpilot.starpilot.conditional_mode.traffic import TrafficOwner, TrafficVe
from openpilot.starpilot.controllers.coast import CoastRuntime
from openpilot.starpilot.controllers.mode_actions import ModeActionOwner, KINDS as CONTROLLER_MODE_KINDS, publish_switchback, apply_switchback_gesture
import secrets
from openpilot.starpilot.conditional_mode.manual import ioniq6_media_eligible
from openpilot.starpilot.feature_capabilities import longitudinal_supported
from openpilot.starpilot.conditional_mode.projection import paired_clocks_ns
from openpilot.starpilot.conditional_mode.ui_action import observation as ui_manual_observation
from openpilot.starpilot.longitudinal.cruise_ceiling import CurveCeiling
@@ -120,8 +120,8 @@ def _profile_frame_valid(sm, now_ns: int) -> bool:
def profile_host_needed(CP, *, profile_enabled: bool, conditional_enabled: bool) -> bool:
"""Traffic defaults need a resolver only on the reviewed media topology."""
return profile_enabled or (conditional_enabled and ioniq6_media_eligible(CP))
"""Conditional Traffic profiles follow acceleration ownership, not media buttons."""
return profile_enabled or (conditional_enabled and longitudinal_supported(CP))
def traffic_observation(verdict, *, model_ns: int, owner_present: bool = True):
@@ -358,8 +358,7 @@ def starpilot_main():
# A validated Traffic gesture uses its frozen default profile even when
# CustomPersonalities is off; saved custom category values still require
# that master switch in ProfileHost.
traffic_capable = (CP.openpilotLongitudinalControl and not CP.passive and not CP.dashcamOnly and not CP.notCar and
feature_requested(params, 'conditional'))
traffic_capable = longitudinal_supported(CP) and feature_requested(params, 'conditional')
traffic_settings = ConditionalSettingsOwner(params) if traffic_capable else None
profile_request_ns = -1_000_000_000
profile_host = ProfileHost(params) if profile_host_needed(CP, profile_enabled=profile_replay,
+2 -2
View File
@@ -49,7 +49,7 @@ from openpilot.starpilot.aol.vehicle import policy_for as axis_policy_for, ordin
from openpilot.starpilot.conditional_mode.consumer import ConsumerResult, ModeConsumer
from openpilot.starpilot.conditional_mode.effective_status import publish_ack
from openpilot.starpilot.conditional_mode.policy import ModeChoice
from openpilot.starpilot.conditional_mode.manual import ioniq6_media_eligible
from openpilot.starpilot.feature_capabilities import lateral_supported
from openpilot.starpilot.controllers.mode_actions import SwitchbackStatusOwner, SwitchbackCooldown, ModeActionOwner
from openpilot.starpilot.conditional_mode.projection import paired_clocks_ns
from openpilot.starpilot.conditional_mode.runtime_settings import (ConditionalSettingsOwner, experimental_requested, read_mode_sources)
@@ -126,7 +126,7 @@ class SelfdriveD:
self.conditional_status = 'disabled'
self.controller_action_sock = messaging.sub_sock('slcAction', conflate=False)
self.controller_action_owner = ModeActionOwner()
self.switchback_capable = ioniq6_media_eligible(self.CP) and not self.CP.passive and not self.CP.dashcamOnly and not self.CP.notCar
self.switchback_capable = lateral_supported(self.CP)
self.switchback_status = SwitchbackStatusOwner()
self.switchback_cooldown = SwitchbackCooldown()
self.low_speed_advisory = LowSpeedAdvisory()
@@ -130,12 +130,15 @@ class TrafficOwnerTests(unittest.TestCase):
self.assertEqual(verdict.source_mono_ns, self.sm.logMonoTime['modelV2'])
self.assertFalse(verdict.requested)
def test_conditional_only_startup_does_not_admit_ordinary_profiles_on_other_cars(self):
def test_conditional_startup_uses_longitudinal_capability_without_media_buttons(self):
honda = interfaces[HONDA_CAR.HONDA_CIVIC].get_non_essential_params(HONDA_CAR.HONDA_CIVIC)
self.assertFalse(profile_host_needed(honda, profile_enabled=False, conditional_enabled=True))
self.assertTrue(honda.openpilotLongitudinalControl)
self.assertTrue(profile_host_needed(honda, profile_enabled=False, conditional_enabled=True))
self.assertFalse(profile_host_needed(self.cp, profile_enabled=False, conditional_enabled=False))
self.assertTrue(profile_host_needed(self.cp, profile_enabled=False, conditional_enabled=True))
self.assertTrue(profile_host_needed(honda, profile_enabled=True, conditional_enabled=False))
honda.openpilotLongitudinalControl = False
self.assertFalse(profile_host_needed(honda, profile_enabled=False, conditional_enabled=True))
def test_profile_status_requires_post_update_application(self):
target = object()
@@ -4,7 +4,7 @@ import re
import secrets
from openpilot.cereal import messaging
from openpilot.starpilot.conditional_mode.manual import ioniq6_media_eligible
from openpilot.starpilot.feature_capabilities import lateral_supported
from openpilot.starpilot.conditional_mode.ui_action import current_authority, fresh_service
LIFETIME_NS = 250_000_000
@@ -20,7 +20,7 @@ def authority(sm, cp, now_ns: int, mode: str) -> bool:
drive = int(sm['deviceState'].startedMonoTime)
if mode in ('traffic', 'pulse'):
return current_authority(sm, cp, drive, now_ns)
if mode == 'switchback' and not ioniq6_media_eligible(cp):
if mode == 'switchback' and not lateral_supported(cp):
return False
if mode == 'disengage' and not fresh_service(sm, 'selfdriveState', drive, now_ns):
return False
@@ -206,9 +206,11 @@ class SwitchbackStatusOwner:
def apply_switchback_gesture(owner, event, *, params, settings, sm, cp, now_ns: int, now_boot_ns: int, mode: str = 'switchback') -> bool:
from openpilot.starpilot.conditional_mode.manual import Button, Gesture, Press, read_button_map
from openpilot.starpilot.conditional_mode.manual import Button, Gesture, Press, read_button_map, ioniq6_media_eligible
from openpilot.starpilot.conditional_mode.status import settings_fingerprint
try:
if not ioniq6_media_eligible(cp):
return False
drive = int(sm['deviceState'].startedMonoTime)
if owner.drive != drive:
owner.reset(drive)
@@ -0,0 +1,119 @@
"""Generic mode requests retain their original live source and acknowledgement guards."""
from unittest.mock import Mock
import pytest
from opendbc.car.car_helpers import interfaces
from opendbc.car.ford.values import CAR as FORD
from opendbc.car.honda.values import CAR as HONDA
from opendbc.car.mazda.values import CAR as MAZDA
from opendbc.car.nissan.values import CAR as NISSAN
from opendbc.car.subaru.values import CAR as SUBARU
from opendbc.car.toyota.values import CAR as TOYOTA
from opendbc.car.volkswagen.values import CAR as VW
from openpilot.cereal import messaging
from openpilot.starpilot.controllers.mode_actions import (
ModeActionOwner, ModeActionPublisher, ModeIntent, SwitchbackStatusOwner,
authority, producer_available, publish_switchback,
)
from openpilot.starpilot.controllers.tests.test_mode_actions import DRIVE, NOW, fixture, request
VEHICLES = (TOYOTA.TOYOTA_RAV4_TSS2, HONDA.HONDA_CIVIC, MAZDA.MAZDA_CX5_2022,
SUBARU.SUBARU_IMPREZA, FORD.FORD_ESCAPE_MK4, NISSAN.NISSAN_XTRAIL,
VW.VOLKSWAGEN_GOLF_MK7)
def prepared(candidate):
sm, _ = fixture()
cp = interfaces[candidate].get_non_essential_params(candidate)
# Lateral-only operation is allowed even when the car owns longitudinal.
cp.openpilotLongitudinalControl = False
sm['carControl'].longActive = sm['carControl'].enabled = False
sm['selfdriveState'].enabled = False
heartbeat = publish_switchback(None, ModeIntent(DRIVE, False, False), session='b' * 32,
sequence=1, now_ns=NOW, source_car_control_ns=sm.logMonoTime['carControl'])
sm['slcState'] = heartbeat.slcState
for name in ('seen', 'valid', 'alive'):
getattr(sm, name)['slcState'] = True
sm.logMonoTime['slcState'], sm.recv_time['slcState'] = NOW, NOW / 1e9
return sm, cp
@pytest.mark.parametrize('candidate', VEHICLES)
def test_generic_switchback_publisher_owner_and_status_accept_actual_vehicle(candidate):
sm, cp = prepared(candidate)
sender, publisher, owner = Mock(), ModeActionPublisher(), ModeActionOwner()
assert authority(sm, cp, NOW, 'switchback')
assert producer_available(sm, now_ns=NOW)
assert publisher.dispatch('switchback', sm, cp, sender, now_ns=NOW)
event = messaging.log_from_bytes(sender.send.call_args.args[1].to_bytes())
assert event.slcAction.controllerMode.carFingerprint == cp.carFingerprint
assert owner.update(event, sm, cp, now_ns=NOW)
assert not owner.update(event, sm, cp, now_ns=NOW) # No duplicate toggle.
intent = owner.sample('switchback', sm, cp, now_ns=NOW)
assert intent.requested and intent.effective
state = publish_switchback(None, intent, session='c' * 32, sequence=1,
now_ns=NOW, source_car_control_ns=sm.logMonoTime['carControl'])
serialized = messaging.log_from_bytes(state.to_bytes())
assert SwitchbackStatusOwner().sample(serialized, drive_id=DRIVE, now_ns=NOW)
assert not SwitchbackStatusOwner().sample(serialized, drive_id=DRIVE, now_ns=NOW + 100_000_001)
sm['carControl'].latActive = False
paused = owner.sample('switchback', sm, cp, now_ns=NOW)
assert paused.requested and not paused.effective
assert not publisher.dispatch('traffic', sm, cp, sender, now_ns=NOW + 101_000_000)
@pytest.mark.parametrize('fault', ('passive', 'dashcamOnly', 'notCar', 'canInvalid', 'canTimeout',
'latInactive', 'offroad', 'carStateStale', 'carStateInvalid',
'carControlStale', 'carControlUnseen', 'deviceStale', 'producerStale',
'producerWrongDrive'))
def test_generic_switchback_still_rejects_live_faults_and_missing_producer(fault):
sm, cp = prepared(TOYOTA.TOYOTA_RAV4_TSS2)
if fault in ('passive', 'dashcamOnly', 'notCar'):
setattr(cp, fault, True)
elif fault == 'canInvalid':
sm['carState'].canValid = False
elif fault == 'canTimeout':
sm['carState'].canTimeout = True
elif fault == 'latInactive':
sm['carControl'].latActive = False
elif fault == 'offroad':
sm['deviceState'].started = False
elif fault == 'carStateStale':
sm.logMonoTime['carState'] = NOW - 250_000_001
elif fault == 'carStateInvalid':
sm.valid['carState'] = False
elif fault == 'carControlStale':
sm.logMonoTime['carControl'] = NOW - 250_000_001
elif fault == 'carControlUnseen':
sm.seen['carControl'] = False
elif fault == 'deviceStale':
sm.logMonoTime['deviceState'] = NOW - 1_000_000_001
elif fault == 'producerStale':
sm.logMonoTime['slcState'] = NOW - 100_000_001
else:
sm['slcState'].switchbackMode.driveStartMonoTime = DRIVE + 1
sender = Mock()
assert not ModeActionPublisher().dispatch('switchback', sm, cp, sender, now_ns=NOW)
sender.send.assert_not_called()
def test_generic_mode_owner_rejects_wrong_identity_expiry_and_inactive_traffic():
sm, cp = prepared(TOYOTA.TOYOTA_RAV4_TSS2)
event = request(sm, cp, 'switchbackModeToggle')
wrong = request(sm, cp, 'switchbackModeToggle')
wrong.slcAction.controllerMode.carFingerprint = 'another-car'
assert not ModeActionOwner().update(wrong, sm, cp, now_ns=NOW)
assert not ModeActionOwner().update(event, sm, cp, now_ns=NOW + 250_000_001)
assert not ModeActionOwner().update(request(sm, cp, 'trafficModeToggle'), sm, cp, now_ns=NOW)
def test_actual_dashcam_mazda_cannot_dispatch_generic_switchback():
sm, cp = prepared(MAZDA.MAZDA_CX5)
assert cp.dashcamOnly
assert producer_available(sm, now_ns=NOW)
assert not authority(sm, cp, NOW, 'switchback')
sender = Mock()
assert not ModeActionPublisher().dispatch('switchback', sm, cp, sender, now_ns=NOW)
sender.send.assert_not_called()
@@ -41,7 +41,7 @@ def test_request_replay_expiry_wrong_vehicle_and_drive_never_toggle():
sm, cp = fixture()
owner = ModeActionOwner()
msg = request(sm, cp)
with patch('openpilot.starpilot.controllers.mode_actions.ioniq6_media_eligible', return_value=True):
with patch('openpilot.starpilot.controllers.mode_actions.lateral_supported', return_value=True):
assert owner.update(msg, sm, cp, now_ns=NOW)
assert not owner.update(msg, sm, cp, now_ns=NOW + 100_000_000)
assert owner.requested['traffic']
@@ -57,7 +57,7 @@ def test_request_replay_expiry_wrong_vehicle_and_drive_never_toggle():
def test_temporary_authority_loss_pauses_effective_intent_without_stale_display():
sm, cp = fixture()
owner = ModeActionOwner()
with patch('openpilot.starpilot.controllers.mode_actions.ioniq6_media_eligible', return_value=True):
with patch('openpilot.starpilot.controllers.mode_actions.lateral_supported', return_value=True):
assert owner.update(request(sm, cp), sm, cp, now_ns=NOW)
assert owner.sample('traffic', sm, cp, now_ns=NOW).effective
sm['carControl'].longActive = False
@@ -75,12 +75,12 @@ def test_switchback_allows_qualified_lateral_only_but_unknown_vehicle_denied():
sm['carControl'].longActive = False
sm['selfdriveState'].enabled = False
cp.openpilotLongitudinalControl = False
with patch('openpilot.starpilot.controllers.mode_actions.ioniq6_media_eligible', return_value=True):
with patch('openpilot.starpilot.controllers.mode_actions.lateral_supported', return_value=True):
assert owner.update(request(sm, cp, 'switchbackModeToggle'), sm, cp, now_ns=NOW)
assert owner.sample('switchback', sm, cp, now_ns=NOW).effective
assert not owner.update(request(sm, cp, sequence=2), sm, cp, now_ns=NOW + 100_000_000)
with patch('openpilot.starpilot.controllers.mode_actions.ioniq6_media_eligible', return_value=False):
assert not ModeActionOwner().update(request(sm, cp), sm, cp, now_ns=NOW)
with patch('openpilot.starpilot.controllers.mode_actions.lateral_supported', return_value=False):
assert not ModeActionOwner().update(request(sm, cp, 'switchbackModeToggle'), sm, cp, now_ns=NOW)
def test_switchback_only_rate_limits_two_advisories_and_resets_per_drive():
@@ -131,6 +131,15 @@ def test_actual_alpha_off_wheel_switchback_mapping_and_replay_denial():
assert owner.sample('switchback', f.sm, f.cp, now_ns=fixtures.NOW).effective
assert not apply_switchback_gesture(owner, msg, params=f.params, settings=f.settings, sm=f.sm,
cp=f.cp, now_ns=fixtures.NOW, now_boot_ns=fixtures.BOOT)
# Generic Switchback authority must not grant an Ioniq physical-media source.
from opendbc.car.car_helpers import interfaces
from opendbc.car.toyota.values import CAR
from openpilot.starpilot.controllers.mode_actions import authority
other_cp = interfaces[CAR.TOYOTA_RAV4_TSS2].get_non_essential_params(CAR.TOYOTA_RAV4_TSS2)
other_cp.openpilotLongitudinalControl = False
assert authority(f.sm, other_cp, fixtures.NOW, 'switchback')
assert not apply_switchback_gesture(ModeActionOwner(), msg, params=f.params, settings=f.settings, sm=f.sm,
cp=other_cp, now_ns=fixtures.NOW, now_boot_ns=fixtures.BOOT)
f.params.put('ModeButtonControl', 0, block=True)
assert not apply_switchback_gesture(ModeActionOwner(), msg, params=f.params, settings=f.settings, sm=f.sm,
cp=f.cp, now_ns=fixtures.NOW, now_boot_ns=fixtures.BOOT)
@@ -225,7 +234,7 @@ def test_controller_dispatch_requires_live_producer_and_binds_current_control_st
from openpilot.starpilot.controllers.mode_actions import ModeActionPublisher, ModeIntent, publish_switchback
sm, cp = fixture()
publisher, sender = ModeActionPublisher(), Mock()
with patch('openpilot.starpilot.controllers.mode_actions.ioniq6_media_eligible', return_value=True):
with patch('openpilot.starpilot.controllers.mode_actions.lateral_supported', return_value=True):
assert not publisher.dispatch('switchback', sm, cp, sender, now_ns=NOW)
sender.send.assert_not_called()
state = publish_switchback(None, ModeIntent(DRIVE, False, False), session='b' * 32, sequence=1,
@@ -36,15 +36,45 @@ def eligible(cp):
def button_reported(cp, key):
"""Match known parser omissions without changing other ports' button contracts."""
if cp.brand == 'toyota':
# Toyota CarState emits LKAS and distance events, never Cancel.
if key in KEYS[5:8]:
"""Expose only button events the current vehicle parser actually emits."""
brand = cp.brand
if key not in KEYS[:8]:
return True # Physical media rows have their separate receiver contract.
if brand in ('body', 'byd', 'mg', 'mock', 'psa', 'rivian', 'subaru'):
return False
if brand == 'toyota':
from opendbc.car.toyota.values import CAR, ToyotaFlags
if key == 'MainCruiseButtonControl' or key in KEYS[5:8]:
return False
if key == 'LKASButtonControl':
from opendbc.car.toyota.values import CAR, ToyotaFlags
return bool(cp.flags & (ToyotaFlags.TSS2 | ToyotaFlags.LONG_FILTER) or
cp.carFingerprint == CAR.TOYOTA_PRIUS_RETROFIT)
return bool(cp.flags & ToyotaFlags.TSS2 and not cp.flags & (ToyotaFlags.RADAR_ACC | ToyotaFlags.SECOC) or
cp.flags & ToyotaFlags.LONG_FILTER or cp.carFingerprint in
(CAR.TOYOTA_PRIUS, CAR.TOYOTA_PRIUS_RETROFIT, CAR.TOYOTA_SIENNA_4TH_GEN))
if brand == 'tesla':
from opendbc.car.tesla.values import CAR, TeslaFlags
if key == 'LKASButtonControl':
return bool(cp.carFingerprint not in (CAR.TESLA_MODEL_S_PREAP, CAR.TESLA_MODEL_S_HW1) and
cp.flags & TeslaFlags.AOL_SCREEN_BUTTON)
return bool(key in KEYS[5:8] and cp.carFingerprint == CAR.TESLA_MODEL_S_PREAP)
if key == 'LKASButtonControl':
if brand in ('chrysler', 'gm', 'mazda', 'nissan', 'volkswagen', 'volvo'):
return False
if brand == 'hyundai':
from opendbc.car.hyundai.values import CAR, HyundaiFlags, is_blended
return bool(cp.flags & (HyundaiFlags.CANFD | HyundaiFlags.HAS_LDA_BUTTON) or
cp.carFingerprint == CAR.HYUNDAI_ELANTRA_HEV_2024 or is_blended(cp))
if key == 'MainCruiseButtonControl' and brand in ('chrysler', 'ford', 'mazda', 'nissan', 'volkswagen'):
return False
if key in KEYS[5:8]:
if brand in ('chrysler', 'nissan'):
return False
if brand == 'ford':
return bool(cp.openpilotLongitudinalControl)
if brand == 'mazda':
from opendbc.car.mazda.aol import qualified
return qualified(cp, marked_only=True)
return True
@@ -30,3 +30,30 @@ def display_supported(cp) -> bool:
def longitudinal_supported(cp) -> bool:
return bool(display_supported(cp) and getattr(cp, 'openpilotLongitudinalControl', None) is True and
_native_configuration_valid(cp, longitudinal=True))
def lateral_supported(cp) -> bool:
"""Generic curvature shaping follows steering capability, not cruise ownership.
Creating the optional preference host does not grant steering permission.
Controls retains its current axis, source, fault and final curvature checks.
GM additionally keeps its existing finalized physical-configuration checks.
"""
from opendbc.car.structs import CarParams
try:
if (cp is None or not cp.carFingerprint or getattr(cp, 'passive', None) is not False or
getattr(cp, 'dashcamOnly', None) is not False or getattr(cp, 'notCar', None) is not False or
(getattr(cp, 'secOcRequired', False) and not getattr(cp, 'secOcKeyAvailable', False)) or
cp.steerControlType not in (CarParams.SteerControlType.torque, CarParams.SteerControlType.angle,
CarParams.SteerControlType.curvature) or
not any(config.safetyModel not in (CarParams.SafetyModel.silent, CarParams.SafetyModel.noOutput,
CarParams.SafetyModel.allOutput) for config in cp.safetyConfigs)):
return False
if cp.brand == 'gm':
from openpilot.starpilot.lateral.bolt_policy import supported_cp as bolt_supported_cp
from opendbc.car.gm.lateral import lane_centering_supported
return bool(bolt_supported_cp(cp) or lane_centering_supported(cp))
return True
except (AttributeError, IndexError, TypeError, ValueError):
return False
+2 -2
View File
@@ -97,8 +97,8 @@ def _qualified(ctx: AuthorityContext, group: str) -> bool:
from opendbc.car.toyota.interface import toyota_auto_hold_supported
return toyota_auto_hold_supported(cp)
if group == "switchback_wheel":
from openpilot.starpilot.conditional_mode.manual import ioniq6_media_eligible
return bool(ioniq6_media_eligible(cp) and not cp.notCar and not cp.passive and not cp.dashcamOnly)
from openpilot.starpilot.feature_capabilities import lateral_supported
return lateral_supported(cp)
if group in ("conditional", "conditional_wheel", "long_output"):
return bool((cp.openpilotLongitudinalControl or ctx.configuration_longitudinal) and not cp.notCar and not cp.dashcamOnly and not cp.passive)
if group == "slc":
+3 -6
View File
@@ -8,7 +8,7 @@ from openpilot.cereal.services import SERVICE_LIST
from openpilot.common.realtime import DT_CTRL
from openpilot.common.params import UnknownKeyName
from openpilot.starpilot.lateral.lane_centering import ControlMode, LaneCenteringRequest, LaneCenteringSettings, STRENGTH_BASELINE, STRENGTH_MAX
from openpilot.starpilot.longitudinal.ioniq6_start import eligible as ioniq6_long_eligible
from openpilot.starpilot.feature_capabilities import lateral_supported
def read_settings(params) -> LaneCenteringSettings:
@@ -43,11 +43,8 @@ def read_settings(params) -> LaneCenteringSettings:
def runtime_supported(CP) -> bool:
"""Keep the preference reader available when a supported drive starts Off."""
from openpilot.starpilot.lateral.bolt_policy import supported_cp as bolt_supported_cp
from opendbc.car.gm.lateral import lane_centering_supported as gm_supported_cp
return (os.getenv('LANE_CENTERING_REPLAY_RUNTIME') == '1' or
ioniq6_long_eligible(CP) or (CP is not None and bolt_supported_cp(CP)) or gm_supported_cp(CP))
"""Keep the preference reader available for every configured steering owner."""
return os.getenv('LANE_CENTERING_REPLAY_RUNTIME') == '1' or lateral_supported(CP)
class LaneCenteringHost:
@@ -0,0 +1,154 @@
"""Generic lane preferences reach each native controller without granting an axis."""
import os
from unittest import mock
import pytest
from opendbc.car import gen_empty_fingerprint
from opendbc.car.car_helpers import interfaces
from opendbc.car.ford.values import CAR as FORD
from opendbc.car.honda.values import CAR as HONDA
from opendbc.car.mazda.values import CAR as MAZDA
from opendbc.car.nissan.values import CAR as NISSAN
from opendbc.car.subaru.values import CAR as SUBARU
from opendbc.car.toyota.values import CAR as TOYOTA
from opendbc.car.volkswagen.values import CAR as VOLKSWAGEN
from opendbc.car.volkswagen.values import NetworkLocation
from openpilot.common.params import Params
from openpilot.common.prefix import OpenpilotPrefix
from openpilot.selfdrive.controls.controlsd import Controls
from openpilot.starpilot.feature_capabilities import lateral_supported
from openpilot.starpilot.lateral.lane_runtime import runtime_supported
from openpilot.starpilot.lateral.tests.test_lane_runtime import feed, publish_corolla_params
GENERIC_STEERING = (
HONDA.HONDA_CIVIC,
MAZDA.MAZDA_CX5_2022,
SUBARU.SUBARU_IMPREZA,
FORD.FORD_ESCAPE_MK4,
NISSAN.NISSAN_XTRAIL,
VOLKSWAGEN.VOLKSWAGEN_GOLF_MK7,
VOLKSWAGEN.VOLKSWAGEN_ID4_MK1,
)
def native_steering_params(identity):
if identity == VOLKSWAGEN.VOLKSWAGEN_ID4_MK1:
fingerprint = gen_empty_fingerprint()
fingerprint[1][0x13D] = 32 # QFK_01 on the gateway side, as in the native MEB interface tests.
cp = interfaces[identity].get_params(identity, fingerprint, [], False, False, False)
assert cp.networkLocation == NetworkLocation.gateway
assert not cp.dashcamOnly
return cp
return interfaces[identity].get_non_essential_params(identity)
@pytest.mark.parametrize('identity', GENERIC_STEERING)
def test_actual_controller_paths_admit_generic_lane_host_and_keep_original_output_when_off(identity):
with OpenpilotPrefix(), mock.patch.dict(os.environ, {'SIMULATION': '1', 'REPLAY': '1',
'LANE_CENTERING_REPLAY_RUNTIME': '0', 'AOL_REPLAY_RUNTIME': '0'}), \
mock.patch('openpilot.selfdrive.controls.controlsd.messaging.PubMaster'):
params = Params()
cp = native_steering_params(identity)
assert lateral_supported(cp)
params.put('CarParams', cp.to_bytes(), block=True)
candidate, reference = Controls(), Controls()
assert candidate.lane_centering_host is not None
reference.lane_centering_host = None
for tick in range(20):
now = 1_000_000_000 + tick * 10_000_000
feed(candidate, now, tick)
feed(reference, now, tick)
actual, _ = candidate.state_control()
expected, _ = reference.state_control()
assert actual.to_dict() == expected.to_dict()
@pytest.mark.parametrize('identity', GENERIC_STEERING)
def test_actual_controller_paths_apply_generic_lane_curvature_then_withdraw_for_fault(identity):
with OpenpilotPrefix(), mock.patch.dict(os.environ, {'SIMULATION': '1', 'REPLAY': '1',
'LANE_CENTERING_REPLAY_RUNTIME': '0', 'AOL_REPLAY_RUNTIME': '0'}), \
mock.patch('openpilot.selfdrive.controls.controlsd.messaging.PubMaster'):
params = Params()
cp = native_steering_params(identity)
params.put('CarParams', cp.to_bytes(), block=True)
params.put_bool('LaneCentering', True, block=True)
params.put('LaneCenteringE2EAuthority', 0., block=True)
params.put('LaneCenterOffset', .2, block=True)
controls = Controls()
assert controls.lane_centering_host is not None
for tick in range(60):
now = 2_000_000_000 + tick * 10_000_000
feed(controls, now, tick)
command, _ = controls.state_control()
assert command.latActive
assert controls.last_lane_centering_result.correction > 0.
assert controls.lane_centering_applied > 0.
for tick in range(60, 65):
now = 2_000_000_000 + tick * 10_000_000
feed(controls, now, tick, fault=True)
command, _ = controls.state_control()
assert not command.latActive
assert controls.lane_centering_applied == 0.
def test_toyota_full_card_startup_admits_normal_lane_host_without_replay_opt_in():
with OpenpilotPrefix(), mock.patch.dict(os.environ, {'SIMULATION': '1', 'REPLAY': '1',
'LANE_CENTERING_REPLAY_RUNTIME': '0', 'AOL_REPLAY_RUNTIME': '0'}), \
mock.patch('openpilot.selfdrive.controls.controlsd.messaging.PubMaster'):
params = Params()
cp = publish_corolla_params(params)
assert lateral_supported(cp)
params.put('CarParams', cp.to_bytes(), block=True)
controls = Controls()
assert controls.lane_centering_host is not None
def test_actual_secoc_configuration_requires_its_key_without_affecting_nonsecoc():
cp = interfaces[TOYOTA.TOYOTA_SIENNA_4TH_GEN].get_non_essential_params(TOYOTA.TOYOTA_SIENNA_4TH_GEN)
assert cp.secOcRequired and not cp.secOcKeyAvailable
assert not lateral_supported(cp)
cp.secOcKeyAvailable = True
assert lateral_supported(cp)
cp.dashcamOnly = True
assert not lateral_supported(cp)
def test_generic_native_configuration_and_replay_opt_in_are_distinct():
with mock.patch.dict(os.environ, {'LANE_CENTERING_REPLAY_RUNTIME': '0'}):
assert not runtime_supported(None)
cp = interfaces[TOYOTA.TOYOTA_RAV4_TSS2].get_non_essential_params(TOYOTA.TOYOTA_RAV4_TSS2)
assert runtime_supported(cp)
for flag in ('notCar', 'dashcamOnly', 'passive'):
bad = cp.as_reader().as_builder()
setattr(bad, flag, True)
assert not runtime_supported(bad)
for safety_model in ('noOutput', 'silent', 'allOutput'):
bad = cp.as_reader().as_builder()
bad.safetyConfigs[0].safetyModel = safety_model
assert not runtime_supported(bad)
bad = cp.as_reader().as_builder()
bad.safetyConfigs = []
assert not runtime_supported(bad)
with mock.patch.dict(os.environ, {'LANE_CENTERING_REPLAY_RUNTIME': '1'}):
assert runtime_supported(None)
@pytest.mark.parametrize('identity', (MAZDA.MAZDA_CX5, VOLKSWAGEN.VOLKSWAGEN_ID4_MK1))
def test_actual_dashcam_configurations_never_create_lane_host(identity):
with OpenpilotPrefix(), mock.patch.dict(os.environ, {'SIMULATION': '1', 'REPLAY': '1',
'LANE_CENTERING_REPLAY_RUNTIME': '0', 'AOL_REPLAY_RUNTIME': '0'}), \
mock.patch('openpilot.selfdrive.controls.controlsd.messaging.PubMaster'):
params = Params()
# The original empty-fingerprint configurations are real dashcam paths.
cp = interfaces[identity].get_non_essential_params(identity)
assert cp.dashcamOnly
assert not lateral_supported(cp)
assert not runtime_supported(cp)
params.put('CarParams', cp.to_bytes(), block=True)
params.put_bool('LaneCentering', True, block=True)
controls = Controls()
assert controls.CP.dashcamOnly
assert controls.lane_centering_host is None
@@ -294,20 +294,21 @@ class LaneRuntimeTests(unittest.TestCase):
self.assertIsNone(host.sample(Frame(now), now_ns=now, lateral_active=True, longitudinal_active=False))
def test_normal_reader_admission_keeps_exact_vehicle_and_saved_preference_gates(self):
def test_normal_reader_admission_uses_steering_and_saved_preference_gates(self):
with OpenpilotPrefix(), mock.patch.dict(os.environ, {'LANE_CENTERING_REPLAY_RUNTIME': '0'}):
params = Params()
for alternate in (False, True):
with self.subTest(alternate=alternate):
stock, tagged = ioniq_candidate(alternate=alternate)
self.assertFalse(runtime_supported(stock))
self.assertTrue(runtime_supported(stock))
self.assertTrue(runtime_supported(tagged))
self.assertFalse(read_settings(params).enabled)
params.put_bool('LaneCentering', True, block=True)
self.assertTrue(read_settings(params).enabled)
self.assertFalse(runtime_supported(interfaces[CAR.TOYOTA_COROLLA_TSS2].get_non_essential_params(CAR.TOYOTA_COROLLA_TSS2)))
tagged.safetyConfigs[0].safetyParam = 0
self.assertFalse(runtime_supported(tagged))
self.assertTrue(runtime_supported(interfaces[CAR.TOYOTA_COROLLA_TSS2].get_non_essential_params(CAR.TOYOTA_COROLLA_TSS2)))
for safety_model in ('noOutput', 'silent', 'allOutput'):
tagged.safetyConfigs[0].safetyModel = safety_model
self.assertFalse(runtime_supported(tagged))
params.put_bool('LaneCentering', False, block=True)
_, tagged = ioniq_candidate()
params.put_bool('LaneCentering', True, block=True)
@@ -348,7 +349,7 @@ class LaneRuntimeTests(unittest.TestCase):
self.assertIsNone(controls.last_lane_centering_result)
self.assertEqual(controls.lane_centering_applied, 0.0)
params.put('CarParams', stock.to_bytes(), block=True)
self.assertIsNone(Controls().lane_centering_host)
self.assertIsNotNone(Controls().lane_centering_host)
def test_tagged_ioniq_four_axes_and_stale_reset(self):
with OpenpilotPrefix(), mock.patch.dict(os.environ, {'LANE_CENTERING_REPLAY_RUNTIME': '0', 'REPLAY': '1',
@@ -490,7 +491,8 @@ class LaneRuntimeTests(unittest.TestCase):
baseline = Controls()
with mock.patch.dict(os.environ, {'LANE_CENTERING_REPLAY_RUNTIME': '1'}):
candidate = Controls()
self.assertIsNone(baseline.lane_centering_host)
self.assertIsNotNone(baseline.lane_centering_host)
baseline.lane_centering_host = None # Explicit no-feature reference; Toyota now uses the normal host.
previous = 0.0
for tick in range(120):
now = 1_000_000_000 + tick * 10_000_000
@@ -956,9 +956,9 @@ class FeatureSettingsOwner:
rows.extend(offset_rows)
title = "Speed Limit Controller"
elif page == FeaturePage.LANE:
allowed = (configurable and lateral_context) or (not parked and self.authority("lane_live"))
from openpilot.starpilot.lateral.lane_runtime import runtime_supported
strength_supported = runtime_supported(self.vehicle_params())
allowed = strength_supported and ((configurable and lateral_context) or (not parked and self.authority("lane_live")))
lane_enabled, _, _ = self._value("LaneCentering")
rows.extend((self._bool_row("LaneCentering", "Enable Lane Centering", allowed),
replace(self._bool_row("LaneCenteringPauseOnSignal", "Pause on signal", allowed and lane_enabled == "1"),
@@ -1286,7 +1286,7 @@ class FeatureSettingsOwner:
return False
if key in (SLC_PRIORITY, SLC_SECONDARY) and request.capability != self._source_capability():
return False
if key == "LaneCenteringStrength":
if key in LANE_LIVE_KEYS:
from openpilot.starpilot.lateral.lane_runtime import runtime_supported
if not runtime_supported(self.vehicle_params()):
return False
@@ -1410,7 +1410,7 @@ class FeatureSettingsOwner:
if request.value not in self._source_choices():
return False
encoded = "Map Data" if request.value == "Maps" else request.value
if key == "LaneCenteringStrength":
if key in LANE_LIVE_KEYS:
from openpilot.starpilot.lateral.lane_runtime import runtime_supported
if not runtime_supported(self.vehicle_params()):
return False
+2 -2
View File
@@ -662,8 +662,8 @@ class StarShellSession:
if group == "lane_change":
return not cp.notCar and not cp.passive and not cp.dashcamOnly
if group == "switchback_wheel":
from openpilot.starpilot.conditional_mode.manual import ioniq6_media_eligible
return bool(ioniq6_media_eligible(cp) and not cp.notCar and not cp.passive and not cp.dashcamOnly)
from openpilot.starpilot.feature_capabilities import lateral_supported
return lateral_supported(cp)
if group in ("conditional", "conditional_wheel", "long_output", "slc"):
return bool(cp.openpilotLongitudinalControl and not cp.notCar and not cp.passive and not cp.dashcamOnly)
if group in ("aol_wheel", "lane", "torque", "aol"):
+7 -5
View File
@@ -19,7 +19,8 @@ from openpilot.starpilot.ui.appearance_preferences import CameraViewChoice, Onro
from openpilot.starpilot.ui.brand import DISPLAY_VERSION, home_description
from openpilot.starpilot.curve_speed.status import observation as curve_observation
from openpilot.starpilot.conditional_mode.policy import ModeChoice
from openpilot.starpilot.conditional_mode.manual import TRAFFIC_MODE_ACTION, ioniq6_media_eligible, read_button_map
from openpilot.starpilot.conditional_mode.manual import TRAFFIC_MODE_ACTION, read_button_map
from openpilot.starpilot.feature_capabilities import lateral_supported, longitudinal_supported
from openpilot.starpilot.conditional_mode.projection import paired_clocks_ns
from openpilot.starpilot.conditional_mode.runtime_settings import ConditionalSettingsOwner
from openpilot.starpilot.conditional_mode.status import settings_fingerprint
@@ -500,16 +501,17 @@ class RuntimeSnapshotAdapter:
traffic_eligible = False
if getattr(traffic_wire, 'version', 0) == 1 and cp is not None:
try:
traffic_eligible = bool(ioniq6_media_eligible(cp) and cp.openpilotLongitudinalControl and not cp.passive)
traffic_eligible = longitudinal_supported(cp)
except (AttributeError, TypeError, ValueError):
pass # A partial or unreadable CP never qualifies a display receipt.
if traffic_eligible:
if (self._traffic_context_ns is None or now_ns < self._traffic_context_ns or
now_ns - self._traffic_context_ns >= 1_000_000_000):
buttons = read_button_map(params, include_ioniq_media=True)
buttons = read_button_map(params, include_ioniq_media=True, cp=cp)
self._traffic_map_fingerprint = buttons.fingerprint() if buttons is not None else None
self._traffic_map_assigned = bool(buttons is not None and TRAFFIC_MODE_ACTION in
(buttons.mode, buttons.mode_long, buttons.mode_very_long,
(buttons.lkas, buttons.distance, buttons.distance_long, buttons.distance_very_long,
buttons.mode, buttons.mode_long, buttons.mode_very_long,
buttons.custom, buttons.custom_long, buttons.custom_very_long))
try:
self._traffic_profile_valid = read_traffic_settings(params).valid
@@ -641,7 +643,7 @@ class RuntimeSnapshotAdapter:
experimental_available=bool(ui.has_longitudinal_control),
experimental_enabled=observed_experimental,
personality=list(Personality).index(persona),
switchback_mode=bool(cp is not None and ioniq6_media_eligible(cp) and display_control is not None and display_control.latActive and
switchback_mode=bool(lateral_supported(cp) and display_control is not None and display_control.latActive and
self._switchback.sample(slc, drive_id=drive_id, now_ns=now_ns,
event_ns=int(sm.logMonoTime['slcState']))),
traffic_mode=bool(traffic is not None and traffic.state == 'active'),
@@ -347,6 +347,11 @@ class FeatureSettingsOwnerTests(unittest.TestCase):
self.assertTrue(VehicleStartupPreferences.read(self.params, enabled=True).turn_assist)
def test_live_lane_owner_maps_only_explicit_keys(self):
from opendbc.car.car_helpers import interfaces
from opendbc.car.toyota.values import CAR
cp = interfaces[CAR.TOYOTA_COROLLA_TSS2].get_non_essential_params(CAR.TOYOTA_COROLLA_TSS2)
self.fingerprint = cp.carFingerprint
self.owner.vehicle_params = lambda: cp
self.owner.authority = lambda group: group == "lane_live"
view = self.owner.snapshot("lane", parked=False, system_long=False, lateral_context=False, metric=False)
row = next(row for row in view.rows if row.key == "LaneCentering")
@@ -425,6 +430,11 @@ class FeatureSettingsOwnerTests(unittest.TestCase):
self.assertIsNone(self.params.get("CustomPersonalities"))
def test_release_rechecks_parked_authority_and_source(self):
from opendbc.car.car_helpers import interfaces
from opendbc.car.toyota.values import CAR
cp = interfaces[CAR.TOYOTA_COROLLA_TSS2].get_non_essential_params(CAR.TOYOTA_COROLLA_TSS2)
self.fingerprint = cp.carFingerprint
self.owner.vehicle_params = lambda: cp
row = self._row("lane", "LaneCentering")
request = required_change(row)
self.assertIsNotNone(request)
@@ -463,6 +473,11 @@ class FeatureSettingsOwnerTests(unittest.TestCase):
self.assertTrue(self._row("profiles", "LeadApproachBuffer").available)
def test_parent_and_units_guards(self):
from opendbc.car.car_helpers import interfaces
from opendbc.car.toyota.values import CAR
cp = interfaces[CAR.TOYOTA_COROLLA_TSS2].get_non_essential_params(CAR.TOYOTA_COROLLA_TSS2)
self.fingerprint = cp.carFingerprint
self.owner.vehicle_params = lambda: cp
self.assertFalse(any(row.key in ('SLCConfirmationLower', 'SLCConfirmationHigher') for row in
self.owner.snapshot('slc', parked=True, system_long=True, lateral_context=True, metric=False).rows))
self.assertFalse(self.owner.apply(FeatureSettingsRequest('SLCConfirmationLower', None, 'On', vehicle_fingerprint=self.fingerprint)))
@@ -0,0 +1,238 @@
"""Generic feature displays follow actual capability and fresh owner receipts."""
from pathlib import Path
from types import SimpleNamespace as NS
import tempfile
import unittest
from unittest import mock
from opendbc.car import gen_empty_fingerprint
from opendbc.car.car_helpers import interfaces
from opendbc.car.ford.values import CAR as FordCar
from opendbc.car.honda.values import CAR as HondaCar
from opendbc.car.nissan.values import CAR as NissanCar
from opendbc.car.toyota.values import CAR as ToyotaCar
from opendbc.car.volkswagen.interface import CarInterface as VolkswagenInterface
from opendbc.car.volkswagen.values import CAR as VolkswagenCar
from openpilot.cereal import messaging
from openpilot.common.params import Params
from openpilot.starpilot.conditional_mode.manual import read_button_map
from openpilot.starpilot.conditional_mode.runtime_settings import ConditionalSettingsOwner
from openpilot.starpilot.conditional_mode.status import settings_fingerprint
from openpilot.starpilot.aol.vehicle import policy_for
from openpilot.starpilot.controllers.tests.test_wheel_actions import Publisher, services, state, TYPE
from openpilot.starpilot.controllers.wheel_actions import ACTIONS, WheelConsumer, WheelPublisher
from openpilot.starpilot.galaxy.settings import AuthorityContext, _qualified
from openpilot.starpilot.ui.feature_settings_owner import FeatureSettingsOwner, LANE_LIVE_KEYS
from openpilot.starpilot.ui.feature_settings_state import FeatureSettingsRequest, row_change
from openpilot.starpilot.ui.runtime_snapshot import RuntimeSnapshotAdapter
from openpilot.starpilot.ui.shell import ShellMode
from openpilot.starpilot.ui.tests.test_runtime_snapshot import BOOT_OFFSET_NS, ui_fake
from openpilot.starpilot.ui.tests.test_traffic_status import BOOT, DRIVE, NOW, receipt
from openpilot.starpilot.ui.wheel_feature import WheelFeature
def params_for(candidate):
return interfaces[candidate].get_non_essential_params(candidate)
class TestGlobalFeaturePresentation(unittest.TestCase):
def setUp(self):
directory = tempfile.TemporaryDirectory()
self.addCleanup(directory.cleanup)
self.params = Params(directory.name)
def traffic_ui(self, cp, *, controller_source):
ui = ui_fake()
ui.CP, ui.params = cp, self.params
ui.sm['deviceState'].startedMonoTime = DRIVE
ui.sm['carState'].canValid, ui.sm['carState'].canTimeout = True, False
ui.sm['carControl'].longActive = True
ui.sm['selfdriveState'].enabled = True
ui.sm.put('controlsState', NS(longControlState='pid'))
self.params.put('DistanceButtonControl', 6, block=True)
snapshot = ConditionalSettingsOwner(self.params).refresh(NOW)
mapping = read_button_map(self.params, include_ioniq_media=True, cp=cp)
self.assertIsNotNone(mapping)
event = receipt(settings_fingerprint(snapshot), '' if controller_source else mapping.fingerprint()).as_builder()
event.slcState.trafficMode.controllerSource = controller_source
if controller_source:
event.slcState.trafficMode.sourceBootTime = 0
ui.sm.put('slcState', messaging.log_from_bytes(event.to_bytes()).slcState)
# SLC is an independent owner. Its validity cannot veto a valid nested Traffic receipt.
ui.sm.valid['slcState'] = False
return ui
def test_actual_toyota_and_pid_honda_traffic_receipts_reach_onroad_state(self):
for candidate in (ToyotaCar.TOYOTA_RAV4_TSS2, HondaCar.HONDA_CIVIC):
for controller_source in (False, True):
with self.subTest(candidate=candidate, controller_source=controller_source):
cp = params_for(candidate)
self.assertTrue(cp.openpilotLongitudinalControl)
ui = self.traffic_ui(cp, controller_source=controller_source)
adapter = RuntimeSnapshotAdapter(ui, mono_clock=lambda: NOW, boot_clock=lambda: NOW + BOOT_OFFSET_NS)
with mock.patch('openpilot.starpilot.ui.runtime_snapshot.paired_clocks_ns', return_value=(NOW, BOOT, 0)):
state = adapter.build(ShellMode.ONROAD, now_ns=NOW).onroad
self.assertTrue(state.traffic_mode)
self.assertEqual((state.traffic_display.state, state.traffic_display.label), ('active', 'TRAFFIC'))
self.assertFalse(adapter.build(ShellMode.ONROAD, now_ns=NOW + 100_000_001).onroad.traffic_mode)
def test_traffic_never_claims_active_without_capability_or_current_effective_control(self):
for changed in ('stock_long', 'passive', 'dashcam', 'not_car', 'empty_identity', 'missing_cp',
'long_inactive', 'selfdrive_disabled', 'invalid_can', 'ineffective', 'wrong_settings', 'wrong_drive'):
with self.subTest(changed=changed):
cp = params_for(ToyotaCar.TOYOTA_RAV4_TSS2)
ui = self.traffic_ui(cp, controller_source=True)
if changed == 'stock_long':
cp.openpilotLongitudinalControl = False
elif changed == 'passive':
cp.passive = True
elif changed == 'dashcam':
cp.dashcamOnly = True
elif changed == 'not_car':
cp.notCar = True
elif changed == 'empty_identity':
cp.carFingerprint = ''
elif changed == 'missing_cp':
ui.CP = None
elif changed == 'long_inactive':
ui.sm['carControl'].longActive = False
elif changed == 'selfdrive_disabled':
ui.sm['selfdriveState'].enabled = False
elif changed == 'invalid_can':
ui.sm['carState'].canValid = False
else:
event = messaging.new_message('slcState', valid=False)
event.slcState = ui.sm['slcState'].to_dict()
if changed == 'ineffective':
event.slcState.trafficMode.effective = False
elif changed == 'wrong_settings':
event.slcState.trafficMode.settingsFingerprint = 'b' * 64
else:
event.slcState.trafficMode.driveStartMonoTime = DRIVE + 1
ui.sm.put('slcState', messaging.log_from_bytes(event.to_bytes()).slcState)
adapter = RuntimeSnapshotAdapter(ui, mono_clock=lambda: NOW, boot_clock=lambda: NOW + BOOT_OFFSET_NS)
with mock.patch('openpilot.starpilot.ui.runtime_snapshot.paired_clocks_ns', return_value=(NOW, BOOT, 0)):
self.assertFalse(adapter.build(ShellMode.ONROAD, now_ns=NOW).onroad.traffic_mode)
def test_switchback_badge_and_settings_are_generic_lateral_with_original_receipt_expiry(self):
for candidate in (ToyotaCar.TOYOTA_RAV4_TSS2, HondaCar.HONDA_CIVIC, FordCar.FORD_ESCAPE_MK4):
with self.subTest(candidate=candidate):
cp = params_for(candidate)
ui = ui_fake()
ui.CP = cp
ui.sm['deviceState'].startedMonoTime = DRIVE
ui.sm['carState'].canValid, ui.sm['carState'].canTimeout = True, False
event = messaging.new_message('slcState', valid=True, logMonoTime=NOW)
event.slcState.switchbackMode = {
'version': 1, 'sessionId': 'c' * 32, 'sequence': 1,
'observedMonoTime': NOW, 'validUntilMonoTime': NOW + 100_000_000,
'driveStartMonoTime': DRIVE, 'requested': True, 'effective': True,
'sourceCarControlMonoTime': NOW,
}
ui.sm.put('slcState', messaging.log_from_bytes(event.to_bytes()).slcState)
adapter = RuntimeSnapshotAdapter(ui)
self.assertTrue(adapter.build(ShellMode.ONROAD, now_ns=NOW).onroad.switchback_mode)
self.assertFalse(adapter.build(ShellMode.ONROAD, now_ns=NOW + 100_000_001).onroad.switchback_mode)
self.assertTrue(_qualified(AuthorityContext(False, cp, None), 'switchback_wheel'))
cp.passive = True
self.assertFalse(_qualified(AuthorityContext(False, cp, None), 'switchback_wheel'))
def test_every_lane_control_matches_runtime_and_rechecks_before_writing(self):
cp = params_for(ToyotaCar.TOYOTA_RAV4_TSS2)
owner = FeatureSettingsOwner(self.params, lambda _group: True,
vehicle_fingerprint=lambda: cp.carFingerprint, vehicle_params=lambda: cp)
rows = owner.snapshot('lane', parked=True, system_long=True, lateral_context=True, metric=False).rows
row = next(row for row in rows if row.key == 'LaneCentering')
request = row_change(row, 1)
self.assertIsNotNone(request)
self.assertTrue(row.available)
self.assertTrue(owner.apply(request))
self.assertTrue(self.params.get_bool('LaneCentering'))
rows = owner.snapshot('lane', parked=True, system_long=True, lateral_context=True, metric=False).rows
self.assertTrue(all(row.available for row in rows if row.key in LANE_LIVE_KEYS))
requests = [FeatureSettingsRequest(row.key, row.source, row.value, vehicle_fingerprint=row.vehicle_fingerprint)
for row in rows if row.key in LANE_LIVE_KEYS]
before = {key: Path(self.params.get_param_path(key)).read_bytes()
if Path(self.params.get_param_path(key)).exists() else None for key in LANE_LIVE_KEYS}
cp.dashcamOnly = True
blocked = owner.snapshot('lane', parked=True, system_long=True, lateral_context=True, metric=False).rows
self.assertTrue(all(not row.available for row in blocked if row.key in LANE_LIVE_KEYS))
self.assertTrue(all(not owner.apply(request) for request in requests))
after = {key: Path(self.params.get_param_path(key)).read_bytes()
if Path(self.params.get_param_path(key)).exists() else None for key in LANE_LIVE_KEYS}
self.assertEqual(before, after)
def test_reported_distance_actions_do_not_depend_on_aol_policy(self):
nissan = params_for(NissanCar.NISSAN_ROGUE)
fingerprint = gen_empty_fingerprint()
fingerprint[1][0x40] = 8
volkswagen = VolkswagenInterface.get_params(VolkswagenCar.VOLKSWAGEN_GOLF_MK7,
fingerprint, [], True, False, False)
self.assertFalse(nissan.openpilotLongitudinalControl)
self.assertTrue(volkswagen.alphaLongitudinalAvailable)
self.assertTrue(volkswagen.openpilotLongitudinalControl)
for cp, options in ((nissan, ('Bookmark', 'Switchback Mode')),
(volkswagen, ('Bookmark', 'Cycle personality', 'Experimental override', 'Traffic mode', 'Switchback Mode'))):
with self.subTest(candidate=cp.carFingerprint):
self.assertFalse(policy_for(cp).settings_supported)
owner = FeatureSettingsOwner(self.params, lambda _group: True,
vehicle_fingerprint=lambda cp=cp: cp.carFingerprint, vehicle_params=lambda cp=cp: cp)
wheel = WheelFeature(owner)
choices = wheel.choices('DistanceButtonControl')
self.assertTrue(set(options).issubset(choices))
self.assertTrue({'Pause steering', 'Pause longitudinal', 'Toggle AOL'}.isdisjoint(choices))
self.assertEqual(wheel.choices('LKASButtonControl'), ())
self.assertEqual(wheel.choices('MainCruiseButtonControl'), ())
self.assertEqual(wheel.choices('ModeButtonControl'), ())
self.assertEqual(wheel.choices('StarButtonControl'), ())
for option in options:
with self.subTest(option=option):
row = next(row for row in wheel.rows() if row.key == 'wheel:DistanceButtonControl')
request = FeatureSettingsRequest(row.key, row.source, option, capability=row.capability)
self.assertTrue(wheel.apply(request))
publisher = WheelPublisher(self.params)
publisher.observe(self.params, cp, state(), now_ns=NOW, drive_id=DRIVE)
publisher.observe(self.params, cp, state(True), now_ns=NOW + 10_000_000, drive_id=DRIVE)
when = NOW + 20_000_000
commands = publisher.observe(self.params, cp, state(False), now_ns=when, drive_id=DRIVE)
self.assertEqual(commands, (('DistanceButtonControl', ACTIONS[option]),))
output = Publisher()
publisher.publish(commands, cp, output, now_ns=when, drive_id=DRIVE,
source_car_ns=when, source_control_ns=when)
event = messaging.log_from_bytes(output.events[0][1])
sm = services(cp, now=when)
# Use this test's drive identifier in the exact original wire checks.
sm['deviceState'].startedMonoTime = DRIVE
command = WheelConsumer().accept(event, self.params, cp, sm, now_ns=when)
self.assertIsNotNone(command)
self.assertEqual((command.key, command.action), ('DistanceButtonControl', ACTIONS[option]))
self.params.put('DistanceButtonControl', 0, block=True)
self.assertIsNone(WheelConsumer().accept(event, self.params, cp, sm, now_ns=when))
def test_toyota_lkas_generic_bookmark_still_preserves_fixed_main_and_absent_cancel(self):
cp = params_for(ToyotaCar.TOYOTA_RAV4_TSS2)
owner = FeatureSettingsOwner(self.params, lambda _group: True,
vehicle_fingerprint=lambda: cp.carFingerprint, vehicle_params=lambda: cp)
wheel = WheelFeature(owner)
self.assertIn('Bookmark', wheel.choices('LKASButtonControl'))
self.assertEqual(wheel.choices('MainCruiseButtonControl'), ())
self.assertEqual(wheel.choices('CancelButtonControl'), ())
row = next(row for row in wheel.rows() if row.key == 'wheel:LKASButtonControl')
self.assertTrue(wheel.apply(FeatureSettingsRequest(row.key, row.source, 'Bookmark', capability=row.capability)))
publisher = WheelPublisher(self.params)
publisher.observe(self.params, cp, state(), now_ns=NOW, drive_id=DRIVE)
when = NOW + 10_000_000
commands = publisher.observe(self.params, cp, state(True, button=TYPE.lkas), now_ns=when, drive_id=DRIVE)
self.assertEqual(commands, (('LKASButtonControl', 8),))
output = Publisher()
publisher.publish(commands, cp, output, now_ns=when, drive_id=DRIVE,
source_car_ns=when, source_control_ns=when)
sm = services(cp, now=when)
sm['deviceState'].startedMonoTime = DRIVE
event = messaging.log_from_bytes(output.events[0][1])
self.assertEqual(WheelConsumer().accept(event, self.params, cp, sm, now_ns=when).action, 8)
if __name__ == '__main__':
unittest.main()
+2 -4
View File
@@ -2,6 +2,7 @@ import hashlib
from openpilot.starpilot.aol.vehicle import policy_for
from openpilot.starpilot.conditional_mode.manual import ioniq6_media_eligible
from openpilot.starpilot.feature_capabilities import lateral_supported
from openpilot.starpilot.controllers.wheel_actions import ACTIONS, DEFAULTS, KEYS, button_reported, eligible
from openpilot.starpilot.saved_document import commit_exact
from openpilot.starpilot.saved_source import read_saved
@@ -51,14 +52,11 @@ class WheelFeature:
return ()
if key in KEYS[8:] and not media:
return ()
if key in KEYS[:5] and not policy.settings_supported and not (
key == "LKASButtonControl" and policy.auxiliary_lkas_supported):
return ()
codes = {0, 8, 11, 12, 13}
if self.owner.longitudinal_available():
codes.update((1, 2, 5, 10, 14))
codes.add(6)
if media:
if lateral_supported(cp):
codes.add(7)
if not policy.fixed_cruise_buttons and policy.settings_supported:
if key in KEYS[:5] or not policy.explicit_latch: