mirror of
https://github.com/firestar5683/StarPilot.git
synced 2026-10-31 11:03:51 +08:00
Enable Global Driving Features
This commit is contained in:
firestar5683
@@ -34,14 +34,20 @@ class TestConditionalRuntimeLoop(unittest.TestCase):
|
||||
with tempfile.TemporaryDirectory() as directory:
|
||||
params = Params(directory)
|
||||
params.put('CarParams', cp.to_bytes(), block=True)
|
||||
params.put_bool('SpeedLimitController', False, block=True)
|
||||
params.put_bool('ShowSpeedLimits', False, block=True)
|
||||
params.put_bool('CurveSpeedController', False, block=True)
|
||||
for enabled in (False, True):
|
||||
params.put('ConditionalModeConfig', json.loads(encode_preferences(SavedPreferences(
|
||||
mode=ModeChoice.CEM if enabled else ModeChoice.STOCK))), block=True)
|
||||
params.put_bool('SafeMode', not enabled, block=True)
|
||||
params.put_bool('UseStarPilotLongitudinalPlanner', enabled, block=True)
|
||||
with self.subTest(enabled=enabled):
|
||||
sm = Mock()
|
||||
sm.update.side_effect = EndLoop
|
||||
settings = {'CONDITIONAL_MODE_REPLAY_RUNTIME': '1' if enabled else '0',
|
||||
settings = {'CONDITIONAL_MODE_REPLAY_RUNTIME': '0',
|
||||
'SLC_REPLAY_RUNTIME': '0', 'CURVE_REPLAY_RUNTIME': '0', 'LONG_PLANNER_REPLAY_RUNTIME': '0'}
|
||||
with patch.dict(os.environ, settings), patch.object(plannerd, 'Params', return_value=params), \
|
||||
with patch.dict(os.environ, settings, clear=True), patch.object(plannerd, 'Params', return_value=params), \
|
||||
patch.object(plannerd, 'config_realtime_process'), \
|
||||
patch.object(plannerd, 'ConditionalPlannerHost', wraps=ConditionalPlannerHost) as conditional_ctor, \
|
||||
patch.object(messaging, 'PubMaster') as publisher, \
|
||||
@@ -52,9 +58,10 @@ class TestConditionalRuntimeLoop(unittest.TestCase):
|
||||
call('carState', conflate=False),
|
||||
call('slcAction', conflate=False),
|
||||
call('slcCruiseEvent', conflate=False)])
|
||||
# The typed manual-mode owner publishes status even without CEM.
|
||||
# Status and transport are reserved independently of saved SLC/CEM
|
||||
# admission, so a later saved change can attach without restarting.
|
||||
self.assertEqual(publisher.call_args.args[0], ['longitudinalPlan', 'starpilotLongitudinalPlan',
|
||||
'driverAssistance', 'slcState'])
|
||||
'driverAssistance', 'slcState', 'slcCruiseCommand'])
|
||||
self.assertEqual(conditional_ctor.call_count, int(enabled))
|
||||
if enabled:
|
||||
self.assertIs(conditional_ctor.call_args.args[0], params)
|
||||
|
||||
@@ -15,6 +15,7 @@ class AolVehiclePolicy:
|
||||
distance_personality: bool = False
|
||||
fixed_cruise_buttons: bool = False
|
||||
lkas_button_supported: bool = True
|
||||
auxiliary_lkas_supported: bool = False
|
||||
physical_stalk_owner: bool = False
|
||||
distance_pause_only: bool = False
|
||||
paddle_pause: bool = False
|
||||
|
||||
@@ -12,6 +12,7 @@ class WheelSettingsPolicy:
|
||||
paddle_pause: bool = True
|
||||
runtime_supported: bool = False
|
||||
lkas_button_supported: bool = False
|
||||
auxiliary_lkas_supported: bool = False
|
||||
|
||||
|
||||
def supports_ioniq6_configuration(cp) -> bool:
|
||||
|
||||
@@ -52,10 +52,14 @@ def qualified(cp, *, marked_only=False):
|
||||
|
||||
|
||||
def policy_for(cp):
|
||||
from openpilot.starpilot.controllers.wheel_actions import button_reported, eligible
|
||||
|
||||
admitted = qualified(cp)
|
||||
auxiliary_lkas = bool(eligible(cp) and cp.brand == 'toyota' and button_reported(cp, 'LKASButtonControl'))
|
||||
return AolVehiclePolicy(intent_supported=admitted, settings_supported=admitted, runtime_supported=admitted,
|
||||
normal_runtime_supported=admitted, ordinary_axis_ack_required=qualified(cp, marked_only=True),
|
||||
full_axis_runtime_required=admitted, fixed_cruise_buttons=admitted, distance_pause_only=admitted,
|
||||
auxiliary_lkas_supported=auxiliary_lkas,
|
||||
alternative_experience_addition=32 if admitted else 0)
|
||||
|
||||
|
||||
|
||||
@@ -0,0 +1,334 @@
|
||||
"""Factory Toyota CAN evidence reaches existing auxiliary action owners."""
|
||||
import json
|
||||
from types import SimpleNamespace as NS
|
||||
|
||||
import pytest
|
||||
|
||||
from opendbc.can import CANPacker
|
||||
from opendbc.car import Bus, structs
|
||||
from opendbc.car.toyota.carstate import CarState
|
||||
from opendbc.car.toyota.interface import CarInterface
|
||||
from opendbc.car.toyota.values import CAR, DBC
|
||||
from openpilot.cereal import messaging
|
||||
from openpilot.common.params import Params
|
||||
from openpilot.starpilot.aol.intent import AolSettings
|
||||
from openpilot.starpilot.aol.vehicle import policy_for
|
||||
from openpilot.starpilot.car.toyota.aol import ToyotaCardIntent
|
||||
from openpilot.starpilot.conditional_mode.card_input import conditional_manual_candidate
|
||||
from openpilot.starpilot.conditional_mode.manual import ButtonTracker
|
||||
from openpilot.starpilot.conditional_mode.policy import ModeChoice
|
||||
from openpilot.starpilot.conditional_mode.preferences import SavedPreferences, encode_preferences
|
||||
from openpilot.starpilot.conditional_mode.runtime_settings import ConditionalSettingsOwner
|
||||
from openpilot.starpilot.controllers.coast import CoastRuntime
|
||||
from openpilot.starpilot.controllers.tests import test_native_wheel_callbacks as native_callbacks
|
||||
from openpilot.starpilot.controllers.tests.test_wheel_actions import Publisher, services, state, candidate, DRIVE, NOW
|
||||
from openpilot.starpilot.controllers.wheel_actions import WheelConsumer, WheelPublisher
|
||||
from openpilot.starpilot.ui.feature_settings_state import FeatureSettingsRequest
|
||||
from openpilot.starpilot.ui.wheel_feature import WheelFeature
|
||||
|
||||
native_session = native_callbacks.native_session
|
||||
|
||||
KEY = 'LKASButtonControl'
|
||||
TYPE = structs.CarState.ButtonEvent.Type
|
||||
|
||||
|
||||
def factory(candidate=CAR.TOYOTA_RAV4_TSS2):
|
||||
return CarInterface.get_params(candidate, {0: {}, 1: {}, 2: {}}, [], False, False, False)
|
||||
|
||||
|
||||
class ToyotaButtons:
|
||||
def __init__(self, cp):
|
||||
self.state = CarState(cp)
|
||||
self.parsers = CarState.get_can_parsers(cp)
|
||||
self.packer = CANPacker(DBC[cp.carFingerprint][Bus.pt])
|
||||
self.tick = 0
|
||||
|
||||
def sample(self, lda=0):
|
||||
self.tick += 1
|
||||
frame = self.packer.make_can_msg('LKAS_HUD', 2, {'LDA_ON_MESSAGE': lda})
|
||||
self.parsers[Bus.cam].update([(self.tick * 10_000_000, [frame])])
|
||||
result = self.state.update(self.parsers)
|
||||
# Isolated edge bench: transport health is a synthetic premise here.
|
||||
# The full CarInterface tests below verify real CAN validity without overrides.
|
||||
result.canValid, result.canTimeout = True, False
|
||||
return result
|
||||
|
||||
|
||||
def emit(params, cp, code):
|
||||
params.put(KEY, code, block=True)
|
||||
source, parser = WheelPublisher(params), ToyotaButtons(cp)
|
||||
source.observe(params, cp, parser.sample(), now_ns=NOW - 10_000_000, drive_id=DRIVE)
|
||||
physical = parser.sample(1)
|
||||
assert [(e.type, e.pressed) for e in physical.buttonEvents] == [(TYPE.lkas, True), (TYPE.lkas, False)]
|
||||
commands = source.observe(params, cp, physical, now_ns=NOW, drive_id=DRIVE)
|
||||
assert commands == ((KEY, code),)
|
||||
assert not any(action for _, action in source.observe(params, cp, parser.sample(1), now_ns=NOW + 1, drive_id=DRIVE))
|
||||
sender = Publisher()
|
||||
source.publish(commands, cp, sender, now_ns=NOW, drive_id=DRIVE, source_car_ns=NOW, source_control_ns=NOW)
|
||||
return messaging.log_from_bytes(sender.events[-1][1])
|
||||
|
||||
|
||||
def feature(params, cp):
|
||||
return WheelFeature(NS(params=params, vehicle_params=lambda: cp, configuration_vehicle=lambda: False,
|
||||
configuration_longitudinal=lambda: cp.openpilotLongitudinalControl,
|
||||
longitudinal_available=lambda: cp.openpilotLongitudinalControl, authority=lambda _: True))
|
||||
|
||||
|
||||
def test_factory_settings_allow_auxiliary_lkas_hide_unreported_cancel_and_keep_ccmain_fixed(tmp_path):
|
||||
params, cp = Params(str(tmp_path)), factory()
|
||||
policy = policy_for(cp)
|
||||
assert policy.fixed_cruise_buttons and policy.auxiliary_lkas_supported
|
||||
wheel = feature(params, cp)
|
||||
rows = {row.key: row for row in wheel.rows()}
|
||||
row = rows['wheel:' + KEY]
|
||||
assert {'Off', 'Bookmark', 'Traffic mode', 'Cycle personality', 'Experimental override'} <= set(row.choices)
|
||||
assert not {'Pause steering', 'Pause longitudinal', 'Toggle AOL'} & set(row.choices)
|
||||
assert not any('Cancel' in key or 'MainCruise' in key for key in rows)
|
||||
assert wheel.apply(FeatureSettingsRequest(row.key, row.source, 'Bookmark', capability=row.capability))
|
||||
assert params.get(KEY) == 8
|
||||
assert not wheel.apply(FeatureSettingsRequest(row.key, b'8', 'Toggle AOL', capability=row.capability))
|
||||
assert not wheel.apply(FeatureSettingsRequest('wheel:CancelButtonControl', None, 'Bookmark', capability=row.capability))
|
||||
|
||||
|
||||
@pytest.mark.parametrize('code', (1, 5, 6, 8))
|
||||
def test_factory_can_parser_to_publisher_to_consumer(tmp_path, code):
|
||||
params, cp = Params(str(tmp_path)), factory()
|
||||
event = emit(params, cp, code)
|
||||
consumer = WheelConsumer()
|
||||
command = consumer.accept(event, params, cp, services(cp), now_ns=NOW)
|
||||
assert (command.key, command.action) == (KEY, code)
|
||||
assert consumer.accept(event, params, cp, services(cp), now_ns=NOW) is None
|
||||
|
||||
|
||||
@pytest.mark.parametrize('code', (3, 4, 9))
|
||||
def test_axis_assignments_cannot_take_native_lkas_or_ccmain_ownership(tmp_path, code):
|
||||
params, cp = Params(str(tmp_path)), factory()
|
||||
params.put(KEY, code, block=True)
|
||||
params.put('MainCruiseButtonControl', code, block=True)
|
||||
native = ToyotaCardIntent(AolSettings(True, 0., code, code, (0, 0, 0), (0, 0, 0)))
|
||||
assert native.settings.lkas_action == native.settings.main_action == 0
|
||||
source, parser = WheelPublisher(params), ToyotaButtons(cp)
|
||||
source.observe(params, cp, parser.sample(), now_ns=NOW - 10_000_000, drive_id=DRIVE)
|
||||
assert not any(action for _, action in source.observe(params, cp, parser.sample(1), now_ns=NOW, drive_id=DRIVE))
|
||||
|
||||
|
||||
@pytest.mark.parametrize('key', ('CancelButtonControl', 'LongCancelButtonControl', 'VeryLongCancelButtonControl'))
|
||||
def test_unreported_toyota_button_receipts_are_rejected(tmp_path, key):
|
||||
params, cp = Params(str(tmp_path)), factory()
|
||||
code = 8
|
||||
params.put(key, code, block=True)
|
||||
sender, source = Publisher(), WheelPublisher(params)
|
||||
source.publish(((key, code),), cp, sender, now_ns=NOW, drive_id=DRIVE, source_car_ns=NOW, source_control_ns=NOW)
|
||||
event = messaging.log_from_bytes(sender.events[-1][1])
|
||||
assert WheelConsumer().accept(event, params, cp, services(cp), now_ns=NOW) is None
|
||||
source.observe(params, cp, state(), now_ns=NOW - 20_000_000, drive_id=DRIVE)
|
||||
button = TYPE.cancel
|
||||
source.observe(params, cp, state(True, button=button), now_ns=NOW - 10_000_000, drive_id=DRIVE)
|
||||
assert not any(action for _, action in source.observe(params, cp, state(False, button=button), now_ns=NOW, drive_id=DRIVE))
|
||||
|
||||
|
||||
def test_non_toyota_cancel_contract_preserved(tmp_path):
|
||||
params = Params(str(tmp_path))
|
||||
_, cp = candidate(False)
|
||||
assert 'Bookmark' in feature(params, cp).choices('CancelButtonControl')
|
||||
params.put('CancelButtonControl', 8, block=True)
|
||||
source = WheelPublisher(params)
|
||||
source.observe(params, cp, state(), now_ns=NOW - 20_000_000, drive_id=DRIVE)
|
||||
source.observe(params, cp, state(True, button=TYPE.cancel), now_ns=NOW - 10_000_000, drive_id=DRIVE)
|
||||
assert source.observe(params, cp, state(False, button=TYPE.cancel), now_ns=NOW, drive_id=DRIVE) == (('CancelButtonControl', 8),)
|
||||
|
||||
|
||||
@pytest.mark.parametrize('code', (5, 8))
|
||||
def test_factory_lkas_reaches_actual_native_action_callback(native_session, monkeypatch, code):
|
||||
runtime, session, ui, _, _ = native_session
|
||||
ui.CP = factory()
|
||||
ui.has_longitudinal_control = ui.CP.openpilotLongitudinalControl
|
||||
calls = []
|
||||
session._native_favorite_actions = lambda: session.native_favorite_actions(
|
||||
lambda: calls.append('bookmark') or True, lambda _: True, lambda: calls.append('experimental') or True)
|
||||
queue = [emit(ui.params, ui.CP, code)]
|
||||
monkeypatch.setattr(runtime.messaging, 'recv_one_or_none', lambda _: queue.pop(0) if queue else None)
|
||||
session._poll_wheel(NOW)
|
||||
session._poll_wheel(NOW)
|
||||
assert calls == ['bookmark' if code == 8 else 'experimental']
|
||||
|
||||
|
||||
def test_factory_lkas_traffic_reaches_existing_controller_owner(native_session):
|
||||
_, _, ui, _, _ = native_session
|
||||
ui.CP = factory()
|
||||
event = emit(ui.params, ui.CP, 6)
|
||||
runtime = CoastRuntime()
|
||||
assert runtime.receive(event, ui.params, ui.CP, ui.sm, now_ns=NOW)
|
||||
assert runtime.traffic_toggles == 1
|
||||
assert not runtime.receive(event, ui.params, ui.CP, ui.sm, now_ns=NOW)
|
||||
assert runtime.traffic_toggles == 1
|
||||
|
||||
|
||||
def test_factory_lkas_cem_assignment_uses_existing_card_arbitration(tmp_path):
|
||||
params, cp = Params(str(tmp_path)), factory()
|
||||
params.put(KEY, 5, block=True)
|
||||
params.put('ConditionalModeConfig', json.loads(encode_preferences(SavedPreferences(mode=ModeChoice.CEM))), block=True)
|
||||
settings, tracker, parser = ConditionalSettingsOwner(params), ButtonTracker(), ToyotaButtons(cp)
|
||||
sm = services(cp)
|
||||
conditional_manual_candidate(parser.sample(), tracker, params, settings, cp, sm, now_ns=NOW - 10_000_000)
|
||||
physical = parser.sample(1)
|
||||
receipt = conditional_manual_candidate(physical, tracker, params, settings, cp, sm, now_ns=NOW)
|
||||
assert receipt is not None and receipt[3] == ModeChoice.CEM
|
||||
source = WheelPublisher(params)
|
||||
source.observe(params, cp, state(), now_ns=NOW - 10_000_000, drive_id=DRIVE)
|
||||
assert not any(action for _, action in source.observe(params, cp, physical, now_ns=NOW,
|
||||
drive_id=DRIVE, blocked_keys={KEY}))
|
||||
|
||||
|
||||
def test_factory_lkas_personality_reaches_native_saved_personality_callback(native_session, monkeypatch):
|
||||
runtime, session, ui, _, _ = native_session
|
||||
ui.CP = factory()
|
||||
ui.has_longitudinal_control = ui.CP.openpilotLongitudinalControl
|
||||
ui.params.put('LongitudinalPersonality', 1, block=True)
|
||||
def save(value):
|
||||
ui.params.put('LongitudinalPersonality', value, block=True)
|
||||
return True
|
||||
session._native_favorite_actions = lambda: session.native_favorite_actions(lambda: True, save, lambda: True)
|
||||
queue = [emit(ui.params, ui.CP, 1)]
|
||||
monkeypatch.setattr(runtime.messaging, 'recv_one_or_none', lambda _: queue.pop(0) if queue else None)
|
||||
session._poll_wheel(NOW)
|
||||
assert ui.params.get('LongitudinalPersonality') == 0
|
||||
session._poll_wheel(NOW)
|
||||
assert ui.params.get('LongitudinalPersonality') == 0
|
||||
|
||||
|
||||
@pytest.mark.parametrize('case', ('invalid_can', 'stale_source', 'remapped', 'wrong_vehicle'))
|
||||
def test_factory_lkas_consumer_rechecks_current_source_and_assignment(tmp_path, case):
|
||||
params, cp = Params(str(tmp_path)), factory()
|
||||
event = emit(params, cp, 8)
|
||||
sm = services(cp)
|
||||
if case == 'invalid_can':
|
||||
sm['carState'].canValid = False
|
||||
elif case == 'stale_source':
|
||||
sm.recv_time['carState'] = (NOW - 300_000_000) / 1e9
|
||||
elif case == 'remapped':
|
||||
params.put(KEY, 5, block=True)
|
||||
else:
|
||||
cp = CarInterface.get_params(CAR.TOYOTA_COROLLA_TSS2, {0: {}, 1: {}, 2: {}}, [], False, False, False)
|
||||
assert WheelConsumer().accept(event, params, cp, sm, now_ns=NOW) is None
|
||||
|
||||
|
||||
class FactoryToyotaInterface:
|
||||
"""Real CarInterface transport validity with complete factory parser messages."""
|
||||
def __init__(self, cp, *, complete):
|
||||
self.interface = CarInterface(cp)
|
||||
self.interface.update([])
|
||||
self.packer = CANPacker(DBC[cp.carFingerprint][Bus.pt])
|
||||
self.tracked = {bus: tuple(parser.vl) for bus, parser in self.interface.can_parsers.items()}
|
||||
self.complete = complete
|
||||
|
||||
def sample(self, now_ns, lda=0):
|
||||
values = {'PCM_CRUISE': {'GAS_RELEASED': 1}, 'PCM_CRUISE_2': {'MAIN_ON': 1},
|
||||
'GEAR_PACKET': {'GEAR': 0}, 'EPS_STATUS': {'LKA_STATE': 1},
|
||||
'BODY_CONTROL_STATE': {}, 'LKAS_HUD': {'LDA_ON_MESSAGE': lda}}
|
||||
frames = [self.packer.make_can_msg(name, 0 if bus == Bus.pt else 2, values.get(name, {}))
|
||||
for bus, names in self.tracked.items() for name in names
|
||||
if self.complete or bus == Bus.cam and name == 'LKAS_HUD']
|
||||
return self.interface.update([(now_ns, frames)])
|
||||
|
||||
|
||||
@pytest.mark.parametrize('complete', (True, False))
|
||||
def test_actual_factory_interface_can_validity_gates_lkas_delivery(tmp_path, complete):
|
||||
params, cp = Params(str(tmp_path)), factory()
|
||||
params.put(KEY, 8, block=True)
|
||||
bench = FactoryToyotaInterface(cp, complete=complete)
|
||||
for tick in range(8):
|
||||
physical = bench.sample(NOW - 90_000_000 + tick * 10_000_000)
|
||||
assert physical.canValid == complete
|
||||
if complete:
|
||||
assert not physical.canTimeout
|
||||
assert physical.cruiseState.available
|
||||
assert physical.gearShifter == structs.CarState.GearShifter.drive
|
||||
assert not (physical.doorOpen or physical.seatbeltUnlatched or physical.steerFaultPermanent)
|
||||
source = WheelPublisher(params)
|
||||
neutral = bench.sample(NOW - 10_000_000)
|
||||
source.observe(params, cp, neutral, now_ns=NOW - 10_000_000, drive_id=DRIVE)
|
||||
physical = bench.sample(NOW, lda=1)
|
||||
assert physical.canValid == complete # Never override parser or interface health.
|
||||
assert [(e.type, e.pressed) for e in physical.buttonEvents] == [(TYPE.lkas, True), (TYPE.lkas, False)]
|
||||
commands = source.observe(params, cp, physical, now_ns=NOW, drive_id=DRIVE)
|
||||
assert commands == (((KEY, 8),) if complete else ())
|
||||
sender = Publisher()
|
||||
# For missing CAN, even a forged publication cannot pass current source health.
|
||||
source.publish(commands if complete else ((KEY, 8),), cp, sender, now_ns=NOW, drive_id=DRIVE,
|
||||
source_car_ns=NOW, source_control_ns=NOW)
|
||||
sm = services(cp)
|
||||
sm['carState'] = physical
|
||||
accepted = WheelConsumer().accept(messaging.log_from_bytes(sender.events[-1][1]), params, cp, sm, now_ns=NOW)
|
||||
if complete:
|
||||
assert (accepted.key, accepted.action) == (KEY, 8)
|
||||
else:
|
||||
assert accepted is None
|
||||
|
||||
|
||||
@pytest.mark.parametrize('vehicle', (CAR.TOYOTA_SIENNA_4TH_GEN, CAR.TOYOTA_RAV4_TSS2_2023))
|
||||
def test_parser_supported_toyota_auxiliary_lkas_does_not_require_aol_admission(tmp_path, vehicle):
|
||||
params, cp = Params(str(tmp_path)), factory(vehicle)
|
||||
baseline = cp.to_dict()
|
||||
policy = policy_for(cp)
|
||||
assert not (policy.intent_supported or policy.settings_supported or policy.runtime_supported or
|
||||
policy.normal_runtime_supported or policy.full_axis_runtime_required)
|
||||
assert policy.alternative_experience_addition == policy.safety_param_addition == 0
|
||||
assert policy.auxiliary_lkas_supported
|
||||
wheel = feature(params, cp)
|
||||
rows = {row.key: row for row in wheel.rows()}
|
||||
row = rows['wheel:' + KEY]
|
||||
assert 'Bookmark' in row.choices
|
||||
assert not {'Pause steering', 'Pause longitudinal', 'Toggle AOL'} & set(row.choices)
|
||||
assert not any('MainCruise' in key or 'Cancel' in key for key in rows)
|
||||
assert wheel.apply(FeatureSettingsRequest(row.key, row.source, 'Bookmark', capability=row.capability))
|
||||
assert not wheel.apply(FeatureSettingsRequest(row.key, b'8', 'Toggle AOL', capability=row.capability))
|
||||
# This bench uses full factory parser transport, without forcing CAN validity.
|
||||
bench = FactoryToyotaInterface(cp, complete=True)
|
||||
for tick in range(8):
|
||||
physical = bench.sample(NOW - 90_000_000 + tick * 10_000_000)
|
||||
assert physical.canValid and not physical.canTimeout
|
||||
source = WheelPublisher(params)
|
||||
source.observe(params, cp, bench.sample(NOW - 10_000_000), now_ns=NOW - 10_000_000, drive_id=DRIVE)
|
||||
physical = bench.sample(NOW, lda=1)
|
||||
assert physical.canValid and not physical.canTimeout
|
||||
assert [(e.type, e.pressed) for e in physical.buttonEvents] == [(TYPE.lkas, True), (TYPE.lkas, False)]
|
||||
commands = source.observe(params, cp, physical, now_ns=NOW, drive_id=DRIVE)
|
||||
assert commands == ((KEY, 8),)
|
||||
sender = Publisher()
|
||||
source.publish(commands, cp, sender, now_ns=NOW, drive_id=DRIVE, source_car_ns=NOW, source_control_ns=NOW)
|
||||
sm = services(cp)
|
||||
sm['carState'] = physical
|
||||
accepted = WheelConsumer().accept(messaging.log_from_bytes(sender.events[-1][1]), params, cp, sm, now_ns=NOW)
|
||||
assert (accepted.key, accepted.action) == (KEY, 8)
|
||||
assert cp.to_dict() == baseline # Auxiliary settings grant no CP/native authority.
|
||||
|
||||
|
||||
@pytest.mark.parametrize('vehicle', (CAR.TOYOTA_SIENNA_4TH_GEN, CAR.TOYOTA_RAV4_TSS2_2023))
|
||||
def test_aol_unsupported_toyota_bookmark_reaches_real_native_callback(native_session, monkeypatch, vehicle):
|
||||
runtime, session, ui, _, _ = native_session
|
||||
ui.CP = factory(vehicle)
|
||||
ui.has_longitudinal_control = ui.CP.openpilotLongitudinalControl
|
||||
assert not policy_for(ui.CP).intent_supported
|
||||
calls = []
|
||||
session._native_favorite_actions = lambda: session.native_favorite_actions(
|
||||
lambda: calls.append('bookmark') or True, lambda _: True, lambda: True)
|
||||
queue = [emit(ui.params, ui.CP, 8)]
|
||||
monkeypatch.setattr(runtime.messaging, 'recv_one_or_none', lambda _: queue.pop(0) if queue else None)
|
||||
session._poll_wheel(NOW)
|
||||
session._poll_wheel(NOW)
|
||||
assert calls == ['bookmark']
|
||||
|
||||
|
||||
@pytest.mark.parametrize('field', ('notCar', 'passive', 'dashcamOnly'))
|
||||
def test_toyota_auxiliary_lkas_keeps_existing_eligibility_limits(tmp_path, field):
|
||||
cp = factory(CAR.TOYOTA_SIENNA_4TH_GEN)
|
||||
setattr(cp, field, True)
|
||||
assert not policy_for(cp).auxiliary_lkas_supported
|
||||
assert feature(Params(str(tmp_path)), cp).rows() == []
|
||||
|
||||
|
||||
def test_toyota_without_parser_lkas_source_has_no_auxiliary_row(tmp_path):
|
||||
cp = factory(CAR.TOYOTA_SIENNA)
|
||||
assert not policy_for(cp).auxiliary_lkas_supported
|
||||
assert feature(Params(str(tmp_path)), cp).choices(KEY) == ()
|
||||
@@ -35,6 +35,19 @@ def eligible(cp):
|
||||
return False
|
||||
|
||||
|
||||
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]:
|
||||
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 True
|
||||
|
||||
|
||||
def cp_fingerprint(cp):
|
||||
reader = cp.as_reader() if hasattr(cp, "as_reader") else cp
|
||||
return hashlib.sha256(reader.as_builder().to_bytes()).hexdigest()
|
||||
@@ -154,7 +167,7 @@ class WheelPublisher:
|
||||
if index < len(keys):
|
||||
gestures.append((keys[index], gesture))
|
||||
commands = [(key, mapping[key]) for key, _ in gestures if key is not None and key not in blocked_keys and
|
||||
key not in self.axis_keys and mapping[key] in DELIVERED_ACTIONS and
|
||||
key not in self.axis_keys and button_reported(cp, key) and mapping[key] in DELIVERED_ACTIONS and
|
||||
(key != "MainCruiseButtonControl" or mapping[key] == 10)]
|
||||
if commands and not audit:
|
||||
current = capture(params)
|
||||
@@ -231,7 +244,7 @@ class WheelConsumer:
|
||||
if action == 0:
|
||||
if key:
|
||||
return None
|
||||
elif key not in KEYS or action not in DELIVERED_ACTIONS:
|
||||
elif key not in KEYS or action not in DELIVERED_ACTIONS or not button_reported(cp, key):
|
||||
return None
|
||||
else:
|
||||
mapping = capture(params)
|
||||
|
||||
@@ -46,7 +46,7 @@ def frame_from_submaster(sm, CP, now_ns: int, *, drive_id: int = 0) -> Frame:
|
||||
car_state, car_control = sm['carState'], sm['carControl']
|
||||
controls, selfdrive = sm['controlsState'], sm['selfdriveState']
|
||||
cruise_mps = float(car_state.vCruise) / 3.6
|
||||
system_longitudinal = bool(CP.openpilotLongitudinalControl) and not bool(CP.pcmCruise)
|
||||
system_longitudinal = bool(CP.openpilotLongitudinalControl) and not any(getattr(CP, flag, False) for flag in ('passive', 'dashcamOnly', 'notCar'))
|
||||
return Frame(now_ns=now_ns, model_ns=model_ns, inputs_fresh=current and bool(car_state.canValid) and not bool(car_state.canTimeout),
|
||||
profile=profile, system_longitudinal=system_longitudinal, long_active=bool(car_control.longActive),
|
||||
enabled=bool(selfdrive.enabled), force_decel=bool(controls.forceDecel), cruise_mps=cruise_mps,
|
||||
|
||||
@@ -107,7 +107,10 @@ class TestCurveStatus(unittest.TestCase):
|
||||
with self.subTest(slc=slc_on, curve=curve_on), tempfile.TemporaryDirectory() as directory, ExitStack() as stack:
|
||||
params = Params(directory)
|
||||
params.put('CarParams', cp_bytes, block=True)
|
||||
params.put_bool(MASTER_KEY, True, block=True)
|
||||
# Saved requests, not a replay flag set to zero, choose the hosts.
|
||||
params.put_bool(MASTER_KEY, curve_on, block=True)
|
||||
params.put_bool('SpeedLimitController', slc_on, block=True)
|
||||
params.put_bool('ShowSpeedLimits', slc_on, block=True)
|
||||
sm = bus(NOW)
|
||||
for name in ('deviceState', 'slcDashboardObservation'):
|
||||
sm.messages[name] = getattr(messaging.new_message(name), name)
|
||||
@@ -161,7 +164,7 @@ class TestCurveStatus(unittest.TestCase):
|
||||
plannerd.main()
|
||||
counts = Counter(name for name, _raw in sent)
|
||||
self.assertEqual(counts['longitudinalPlan'], 2)
|
||||
self.assertEqual(counts['slcState'], 2) # Display-only signs still publish with both controllers off.
|
||||
self.assertEqual(counts['slcState'], 2) # Shared optional status still publishes with both controllers off.
|
||||
self.assertEqual(sum(name == 'slcCruiseEvent' for name, _kw in subscriptions), 1)
|
||||
self.assertTrue(all(not kw['conflate'] for _name, kw in subscriptions))
|
||||
for name, raw in sent:
|
||||
|
||||
@@ -0,0 +1,16 @@
|
||||
"""Generic optional-feature admission from finalized native CarParams.
|
||||
|
||||
Admission creates a host; the host still checks live axis and source authority.
|
||||
PCM cruise describes engagement/set-speed integration, not who controls acceleration.
|
||||
"""
|
||||
|
||||
|
||||
def display_supported(cp) -> bool:
|
||||
return bool(cp is not None and getattr(cp, 'carFingerprint', '') and
|
||||
getattr(cp, 'passive', None) is False and
|
||||
getattr(cp, 'dashcamOnly', None) is False and
|
||||
getattr(cp, 'notCar', None) is False)
|
||||
|
||||
|
||||
def longitudinal_supported(cp) -> bool:
|
||||
return display_supported(cp) and getattr(cp, 'openpilotLongitudinalControl', None) is True
|
||||
@@ -7,12 +7,12 @@ each owner still verifies current vehicle, source, clock, and axis authority.
|
||||
from collections.abc import Mapping
|
||||
from dataclasses import replace
|
||||
|
||||
from opendbc.car.gm.feature_capabilities import longitudinal_supported as gm_long_supported, display_supported as gm_display_supported
|
||||
from opendbc.car.gm.feature_capabilities import longitudinal_supported as gm_long_supported
|
||||
|
||||
from openpilot.starpilot.conditional_mode.preferences import PreferenceError, decode_preferences
|
||||
from openpilot.starpilot.aol.intent import independent_axis_requested, read_settings as read_aol_settings
|
||||
from openpilot.starpilot.aol.vehicle import policy_for as aol_policy_for
|
||||
from openpilot.starpilot.longitudinal.ioniq6_start import eligible as ioniq6_long_eligible
|
||||
from openpilot.starpilot.feature_capabilities import display_supported, longitudinal_supported
|
||||
from openpilot.starpilot.longitudinal.profile_runtime import read_settings as read_profile_settings, selected_profiles_requested, read_global_powertrain_preset
|
||||
from openpilot.starpilot.longitudinal.profile_preferences import read_document_value
|
||||
from openpilot.starpilot.saved_source import read_saved
|
||||
@@ -56,7 +56,7 @@ def requested(params, feature: str, *, include_auxiliary: bool = False, saved_sl
|
||||
if feature in ('slc', 'vision'):
|
||||
settings = read_slc_settings(params) if saved_slc is None else saved_slc
|
||||
return (settings.enabled or settings.display) if feature == 'slc' else (
|
||||
settings.display and Source.VISION in settings.selection.slots)
|
||||
(settings.enabled or settings.display) and Source.VISION in settings.selection.slots)
|
||||
raise ValueError('unknown optional feature')
|
||||
|
||||
|
||||
@@ -78,9 +78,7 @@ def enabled(params, cp, feature: str, environment: Mapping[str, str], *, saved_s
|
||||
return True
|
||||
if feature != 'vision' and environment.get(flags[feature]) == '1':
|
||||
return True
|
||||
capable = cp is not None and (ioniq6_long_eligible(cp) or
|
||||
(feature in ('conditional', 'curve', 'slc', 'profile') and gm_long_supported(cp)) or
|
||||
(feature in ('slc', 'vision') and gm_display_supported(cp)))
|
||||
capable = display_supported(cp) if feature in ('slc', 'vision') else longitudinal_supported(cp)
|
||||
powertrain_requested = (feature == 'profile' and cp is not None and gm_long_supported(cp) and
|
||||
read_document_value(params).valid and read_global_powertrain_preset(params, cp) is not None)
|
||||
return bool(capable and (requested(params, feature, saved_slc=saved_slc) or powertrain_requested))
|
||||
@@ -89,13 +87,13 @@ def enabled(params, cp, feature: str, environment: Mapping[str, str], *, saved_s
|
||||
def slc_transport_capable(cp, environment: Mapping[str, str]) -> bool:
|
||||
"""Reserve optional transport without admitting a saved request or control."""
|
||||
return bool(environment.get('SLC_REPLAY_RUNTIME') == '1' or
|
||||
(cp is not None and (ioniq6_long_eligible(cp) or gm_long_supported(cp) or gm_display_supported(cp))))
|
||||
display_supported(cp))
|
||||
|
||||
|
||||
def slc_runtime_settings(params, cp, environment: Mapping[str, str], *, saved_slc: SlcSettings | None = None):
|
||||
"""Stock display startup cannot inherit a saved speed-control request."""
|
||||
settings = read_slc_settings(params) if saved_slc is None else saved_slc
|
||||
if environment.get('SLC_REPLAY_RUNTIME') != '1' and gm_display_supported(cp) and not gm_long_supported(cp):
|
||||
if environment.get('SLC_REPLAY_RUNTIME') != '1' and not longitudinal_supported(cp):
|
||||
return replace(settings, enabled=False, display=settings.display or settings.enabled, acceptance=replace(settings.acceptance, display_only=True))
|
||||
return settings
|
||||
|
||||
@@ -103,5 +101,5 @@ def slc_runtime_settings(params, cp, environment: Mapping[str, str], *, saved_sl
|
||||
def vision_control_enabled(params, cp, *, saved_slc: SlcSettings | None = None) -> bool:
|
||||
"""Source selection never turns stock-cruise or display-only sessions into control."""
|
||||
settings = read_slc_settings(params) if saved_slc is None else saved_slc
|
||||
return bool(cp is not None and (ioniq6_long_eligible(cp) or gm_long_supported(cp)) and
|
||||
return bool(longitudinal_supported(cp) and
|
||||
settings.enabled and Source.VISION in settings.selection.slots)
|
||||
|
||||
@@ -105,7 +105,7 @@ def _qualified(ctx: AuthorityContext, group: str) -> bool:
|
||||
return bool((cp.openpilotLongitudinalControl or ctx.configuration_longitudinal) and
|
||||
not cp.notCar and not cp.dashcamOnly and not cp.passive)
|
||||
if group == "long":
|
||||
return bool(((cp.openpilotLongitudinalControl and not cp.pcmCruise) or ctx.configuration_longitudinal) and
|
||||
return bool((cp.openpilotLongitudinalControl or ctx.configuration_longitudinal) and
|
||||
not cp.notCar and not cp.dashcamOnly and not cp.passive)
|
||||
if group in ("aol", "aol_wheel") and ctx.configuration_vehicle and ctx.parked:
|
||||
from openpilot.starpilot.aol.vehicle import configuration_settings_policy
|
||||
|
||||
@@ -432,7 +432,7 @@ class SettingsGatewayTest(unittest.TestCase):
|
||||
|
||||
vehicle = HondaInterface.get_params(HondaCAR.HONDA_CIVIC_BOSCH, gen_empty_fingerprint(), [], True, False, False)
|
||||
self.assertTrue(vehicle.openpilotLongitudinalControl and not vehicle.pcmCruise)
|
||||
self.assertFalse(diagnostic_choice_enabled({}, vehicle))
|
||||
self.assertTrue(diagnostic_choice_enabled({}, vehicle))
|
||||
self.context.value = AuthorityContext(True, vehicle, vehicle.to_bytes())
|
||||
self.params.put('SLCPriority1', 'Dashboard', block=True)
|
||||
self.params.put('SLCPriority2', 'Dashboard', block=True)
|
||||
@@ -444,22 +444,56 @@ class SettingsGatewayTest(unittest.TestCase):
|
||||
rows = self.page("slc")["rows"]
|
||||
primary = next(row for row in rows if row["label"] == "Primary source")
|
||||
secondary = next(row for row in rows if row["label"] == "Secondary source")
|
||||
self.assertEqual("Vision" in primary["choices"], available)
|
||||
self.assertEqual("Vision" in secondary["choices"], available)
|
||||
self.assertIn("Vision", primary["choices"])
|
||||
self.assertIn("Vision", secondary["choices"])
|
||||
self.assertNotIn("require your confirmation", primary["reason"])
|
||||
confirmation = next(row for row in rows if row["label"] == "Require confirmation")
|
||||
self.assertEqual(confirmation["value"], "Off")
|
||||
self.assertIn("first speed limit", confirmation["reason"])
|
||||
self.assertIn("any source", confirmation["reason"])
|
||||
|
||||
with mock.patch.dict(os.environ, {"SLC_REPLAY_RUNTIME": "1", "SLC_VISION_DEVELOPMENT": "1"}):
|
||||
with mock.patch.dict(os.environ, {"SLC_REPLAY_RUNTIME": "0", "SLC_VISION_DEVELOPMENT": "0"}):
|
||||
# A normal finalized car offers Vision without development/replay flags.
|
||||
page = self.page("slc")
|
||||
index = next(i for i, row in enumerate(page["rows"]) if row["label"] == "Primary source")
|
||||
intent = self.gateway.preview(page["view"], index, 1, self.session, self.generation)
|
||||
self.assertEqual(intent["proposed"], "Vision")
|
||||
with mock.patch.dict(os.environ, {"SLC_VISION_DEVELOPMENT": "0"}):
|
||||
self.assertFalse(self.gateway.confirm(intent["intent"], self.session, self.generation))
|
||||
self.assertEqual(read_saved(self.params, "SLCPriority1", 128)[0], b"Dashboard")
|
||||
self.assertTrue(self.gateway.confirm(intent["intent"], self.session, self.generation))
|
||||
self.assertEqual(read_saved(self.params, "SLCPriority1", 128)[0], b"Vision")
|
||||
self.params.put('SLCPriority1', 'Dashboard', block=True)
|
||||
original_context = self.context.value
|
||||
for withdrawal in ('passive', 'unavailable'):
|
||||
with self.subTest(withdrawal=withdrawal):
|
||||
self.context.value = original_context
|
||||
page = self.page("slc")
|
||||
index = next(i for i, row in enumerate(page["rows"]) if row["label"] == "Primary source")
|
||||
intent = self.gateway.preview(page["view"], index, 1, self.session, self.generation)
|
||||
self.assertEqual(intent["proposed"], "Vision")
|
||||
if withdrawal == 'passive':
|
||||
withdrawn = vehicle.as_reader().as_builder()
|
||||
withdrawn.passive = True
|
||||
self.context.value = AuthorityContext(True, withdrawn, withdrawn.to_bytes())
|
||||
else:
|
||||
self.context.value = AuthorityContext(True, None, None)
|
||||
with self.assertRaises(SettingsChanged):
|
||||
self.gateway.confirm(intent["intent"], self.session, self.generation)
|
||||
self.assertEqual(read_saved(self.params, "SLCPriority1", 128)[0], b"Dashboard")
|
||||
|
||||
# Source choice is display access on a native stock-ACC car, never a
|
||||
# grant of longitudinal control from a remembered controller switch.
|
||||
from openpilot.starpilot.feature_runtime import slc_runtime_settings, vision_control_enabled
|
||||
stock = HondaInterface.get_params(HondaCAR.HONDA_CIVIC_BOSCH, gen_empty_fingerprint(), [], False, False, False)
|
||||
self.assertFalse(stock.openpilotLongitudinalControl)
|
||||
self.context.value = AuthorityContext(True, stock, stock.to_bytes())
|
||||
self.params.put_bool('SpeedLimitController', True, block=True)
|
||||
page = self.page("slc")
|
||||
index = next(i for i, row in enumerate(page["rows"]) if row["label"] == "Primary source")
|
||||
intent = self.gateway.preview(page["view"], index, 1, self.session, self.generation)
|
||||
self.assertEqual(intent["proposed"], "Vision")
|
||||
self.assertTrue(self.gateway.confirm(intent["intent"], self.session, self.generation))
|
||||
settings = slc_runtime_settings(self.params, stock, {})
|
||||
self.assertTrue(settings.display and settings.acceptance.display_only)
|
||||
self.assertFalse(settings.enabled or vision_control_enabled(self.params, stock))
|
||||
|
||||
def test_pip_saved_choices_no_vehicle_cp_and_one_use_confirmation(self):
|
||||
from openpilot.starpilot.ui.pip_preferences import BLINKER, MASK, read_pip
|
||||
|
||||
@@ -488,19 +488,22 @@ class ProfileRuntimeTests(unittest.TestCase):
|
||||
tuple(message.longitudinalPlan.speeds)))
|
||||
|
||||
self.params.put('CarParams', self.cp.to_bytes(), block=True)
|
||||
self.params.put_bool('CustomPersonalities', True, block=True)
|
||||
self.params.put('StandardFollow', 2.5, block=True)
|
||||
self.params.put('StandardFollowHigh', 2.5, block=True)
|
||||
results = []
|
||||
for enabled in (False, True):
|
||||
self.params.put_bool('CustomPersonalities', enabled, block=True)
|
||||
# Explicit native response removes the separate default Comfort request;
|
||||
# this comparison isolates the saved following/jerk master switch.
|
||||
self.params.put('LongitudinalPersonalityProfiles', current_profile_document(
|
||||
default_personality_profiles(False), enabled=enabled,
|
||||
selected_deceleration_profile='dom_default'), block=True)
|
||||
data, _ = messages(lead=True)
|
||||
data['carState'].canValid = True
|
||||
data['carControl'].longActive = True
|
||||
sm = LoopSM(data)
|
||||
publisher = Publisher()
|
||||
environment = {}
|
||||
if enabled:
|
||||
environment['LONG_PLANNER_REPLAY_RUNTIME'] = '1'
|
||||
environment = {} # Normal admission follows the actual saved request.
|
||||
with (mock.patch.object(plannerd, 'Params', return_value=self.params),
|
||||
mock.patch.object(plannerd, 'config_realtime_process'),
|
||||
mock.patch.object(plannerd, 'LeadApproachPreferences', return_value=None),
|
||||
|
||||
@@ -227,7 +227,12 @@ def test_cached_provider_verification_invalidates_on_replacement_and_source_addi
|
||||
os.utime(binary, ns=(prior.st_atime_ns, prior.st_mtime_ns + 1_000_000_000))
|
||||
assert binary.stat().st_mtime_ns != prior.st_mtime_ns
|
||||
assert ready() and len(calls) == 3
|
||||
prior_source = source.stat()
|
||||
(source / 'new.go').write_bytes(b'package main')
|
||||
# Directory additions can share a metadata timestamp tick too. The cache
|
||||
# watches directory identity before enumerating newly added source files.
|
||||
os.utime(source, ns=(prior_source.st_atime_ns, prior_source.st_mtime_ns + 1_000_000_000))
|
||||
assert source.stat().st_mtime_ns != prior_source.st_mtime_ns
|
||||
assert not ready() and len(calls) == 4
|
||||
|
||||
|
||||
|
||||
@@ -1,11 +1,10 @@
|
||||
"""Availability of the optional camera speed-limit source."""
|
||||
|
||||
from collections.abc import Mapping
|
||||
from openpilot.starpilot.car.hyundai.aol import ioniq6_settings_capable
|
||||
from opendbc.car.gm.feature_capabilities import display_supported as gm_display_supported
|
||||
from openpilot.starpilot.feature_capabilities import display_supported
|
||||
|
||||
|
||||
def diagnostic_choice_enabled(environment: Mapping[str, str], cp=None) -> bool:
|
||||
# A saved source choice does not itself authorize Vision speed control.
|
||||
return ((environment.get("SLC_REPLAY_RUNTIME") == "1" and environment.get("SLC_VISION_DEVELOPMENT") == "1") or
|
||||
(cp is not None and (ioniq6_settings_capable(cp) or gm_display_supported(cp))))
|
||||
display_supported(cp))
|
||||
|
||||
@@ -1,9 +1,8 @@
|
||||
"""Saved Ioniq runtime startup does not replace live driving authority."""
|
||||
"""Saved generic runtime startup does not replace live driving authority."""
|
||||
|
||||
from pathlib import Path
|
||||
import json
|
||||
from tempfile import TemporaryDirectory
|
||||
from unittest.mock import patch
|
||||
|
||||
from openpilot.starpilot.conditional_mode.policy import ModeChoice
|
||||
from openpilot.starpilot.conditional_mode.preferences import SavedPreferences, encode_preferences
|
||||
@@ -12,6 +11,22 @@ from openpilot.starpilot.longitudinal.profile_document import default_personalit
|
||||
from openpilot.starpilot.speed_limits.vision_gate import diagnostic_choice_enabled
|
||||
|
||||
|
||||
def long_cp():
|
||||
from opendbc.car.toyota.interface import CarInterface
|
||||
from opendbc.car.toyota.values import CAR
|
||||
cp = CarInterface.get_non_essential_params(CAR.TOYOTA_RAV4_TSS2)
|
||||
assert cp.openpilotLongitudinalControl and cp.pcmCruise
|
||||
return cp
|
||||
|
||||
|
||||
def stock_cp():
|
||||
from opendbc.car.honda.interface import CarInterface
|
||||
from opendbc.car.honda.values import CAR
|
||||
cp = CarInterface.get_non_essential_params(CAR.HONDA_CIVIC_BOSCH)
|
||||
assert not cp.openpilotLongitudinalControl
|
||||
return cp
|
||||
|
||||
|
||||
class Files:
|
||||
def __init__(self, root: Path):
|
||||
self.root = root
|
||||
@@ -24,83 +39,79 @@ class Files:
|
||||
return b'0' if key == 'CustomPersonalities' else {}
|
||||
|
||||
|
||||
def test_readable_absence_starts_factory_cem_only_on_exact_eligible_cp():
|
||||
def test_readable_absence_starts_factory_cem_only_on_native_long_cp():
|
||||
with TemporaryDirectory() as temporary:
|
||||
params = Files(Path(temporary))
|
||||
with patch('openpilot.starpilot.feature_runtime.ioniq6_long_eligible', side_effect=lambda cp: cp == 'tagged'):
|
||||
assert requested(params, 'conditional')
|
||||
assert enabled(params, 'tagged', 'conditional', {})
|
||||
assert not enabled(params, 'other', 'conditional', {})
|
||||
for name in ('curve', 'slc', 'vision', 'profile'):
|
||||
assert not enabled(params, object(), name, {})
|
||||
(params.root / 'SafeMode').write_bytes(b'1')
|
||||
assert not requested(params, 'conditional')
|
||||
(params.root / 'SafeMode').write_bytes(b'0')
|
||||
(params.root / 'ConditionalModeConfig').write_bytes(b'{broken')
|
||||
(params.root / 'CurveSpeedController').write_bytes(b'true')
|
||||
assert not requested(params, 'conditional')
|
||||
assert not requested(params, 'curve')
|
||||
(params.root / 'ConditionalModeConfig').unlink()
|
||||
(params.root / 'ConditionalModeConfig').mkdir()
|
||||
assert not requested(params, 'conditional') # Read error is never factory absence.
|
||||
assert requested(params, 'conditional')
|
||||
assert enabled(params, long_cp(), 'conditional', {})
|
||||
assert not enabled(params, stock_cp(), 'conditional', {})
|
||||
for name in ('curve', 'slc', 'vision', 'profile'):
|
||||
assert not enabled(params, object(), name, {})
|
||||
(params.root / 'SafeMode').write_bytes(b'1')
|
||||
assert not requested(params, 'conditional')
|
||||
(params.root / 'SafeMode').write_bytes(b'0')
|
||||
(params.root / 'ConditionalModeConfig').write_bytes(b'{broken')
|
||||
(params.root / 'CurveSpeedController').write_bytes(b'true')
|
||||
assert not requested(params, 'conditional')
|
||||
assert not requested(params, 'curve')
|
||||
(params.root / 'ConditionalModeConfig').unlink()
|
||||
(params.root / 'ConditionalModeConfig').mkdir()
|
||||
assert not requested(params, 'conditional') # Read error is never factory absence.
|
||||
|
||||
|
||||
def test_exact_tagged_cp_and_saved_document_start_owner_without_granting_other_cars():
|
||||
def test_native_long_cp_and_saved_document_start_owner_without_granting_stock_control():
|
||||
with TemporaryDirectory() as temporary:
|
||||
params = Files(Path(temporary))
|
||||
(params.root / 'ConditionalModeConfig').write_bytes(encode_preferences(SavedPreferences(mode=ModeChoice.STOCK)))
|
||||
(params.root / 'CurveSpeedController').write_bytes(b'1')
|
||||
with patch('openpilot.starpilot.feature_runtime.ioniq6_long_eligible', side_effect=lambda cp: cp == 'tagged'):
|
||||
for name in ('conditional', 'curve'):
|
||||
assert enabled(params, 'tagged', name, {})
|
||||
assert not enabled(params, 'stock', name, {})
|
||||
(params.root / 'SafeMode').write_bytes(b'1')
|
||||
assert not enabled(params, 'tagged', 'conditional', {})
|
||||
for name in ('conditional', 'curve'):
|
||||
assert enabled(params, long_cp(), name, {})
|
||||
assert not enabled(params, stock_cp(), name, {})
|
||||
(params.root / 'SafeMode').write_bytes(b'1')
|
||||
assert not enabled(params, long_cp(), 'conditional', {})
|
||||
|
||||
|
||||
def test_global_braking_choice_starts_only_qualified_profile_owner_without_custom_master():
|
||||
with TemporaryDirectory() as temporary:
|
||||
params = Files(Path(temporary))
|
||||
source = params.root / 'LongitudinalPersonalityProfiles'
|
||||
with patch('openpilot.starpilot.feature_runtime.ioniq6_long_eligible', side_effect=lambda cp: cp == 'tagged'):
|
||||
assert requested(params, 'profile')
|
||||
assert enabled(params, 'tagged', 'profile', {})
|
||||
assert not enabled(params, 'stock', 'profile', {})
|
||||
for response in ('eco', 'sport'):
|
||||
source.write_text(json.dumps(profile_document(default_personality_profiles(False), enabled=False,
|
||||
global_braking_response=response)))
|
||||
assert requested(params, 'profile')
|
||||
assert enabled(params, 'tagged', 'profile', {})
|
||||
assert not enabled(params, 'stock', 'profile', {})
|
||||
assert requested(params, 'profile')
|
||||
assert enabled(params, long_cp(), 'profile', {})
|
||||
assert not enabled(params, stock_cp(), 'profile', {})
|
||||
for response in ('eco', 'sport'):
|
||||
source.write_text(json.dumps(profile_document(default_personality_profiles(False), enabled=False,
|
||||
selected_deceleration_profile='dom_default')))
|
||||
assert not requested(params, 'profile')
|
||||
bad = profile_document(default_personality_profiles(False), enabled=False)
|
||||
bad['globalBrakingResponse'] = 'unsupported'
|
||||
source.write_text(json.dumps(bad))
|
||||
assert not requested(params, 'profile')
|
||||
source.write_text('{invalid')
|
||||
assert not requested(params, 'profile')
|
||||
(params.root / 'SafeMode').write_bytes(b'corrupt')
|
||||
assert not enabled(params, 'tagged', 'conditional', {})
|
||||
global_braking_response=response)))
|
||||
assert requested(params, 'profile')
|
||||
assert enabled(params, long_cp(), 'profile', {})
|
||||
assert not enabled(params, stock_cp(), 'profile', {})
|
||||
source.write_text(json.dumps(profile_document(default_personality_profiles(False), enabled=False,
|
||||
selected_deceleration_profile='dom_default')))
|
||||
assert not requested(params, 'profile')
|
||||
bad = profile_document(default_personality_profiles(False), enabled=False)
|
||||
bad['globalBrakingResponse'] = 'unsupported'
|
||||
source.write_text(json.dumps(bad))
|
||||
assert not requested(params, 'profile')
|
||||
source.write_text('{invalid')
|
||||
assert not requested(params, 'profile')
|
||||
(params.root / 'SafeMode').write_bytes(b'corrupt')
|
||||
assert not enabled(params, long_cp(), 'conditional', {})
|
||||
|
||||
|
||||
def test_vision_control_needs_saved_master_and_vision_source_on_exact_long_cp():
|
||||
def test_vision_control_needs_saved_master_and_vision_source_on_native_long_cp():
|
||||
with TemporaryDirectory() as temporary:
|
||||
params = Files(Path(temporary))
|
||||
(params.root / 'SpeedLimitController').write_bytes(b'1')
|
||||
(params.root / 'SLCPriority1').write_bytes(b'Dashboard')
|
||||
(params.root / 'SLCPriority2').write_bytes(b'Vision')
|
||||
with patch('openpilot.starpilot.feature_runtime.ioniq6_long_eligible', side_effect=lambda cp: cp == 'tagged'):
|
||||
from openpilot.starpilot.feature_runtime import vision_control_enabled
|
||||
assert requested(params, 'vision')
|
||||
assert enabled(params, 'tagged', 'vision', {})
|
||||
assert not enabled(params, 'stock', 'vision', {})
|
||||
assert vision_control_enabled(params, 'tagged')
|
||||
assert not vision_control_enabled(params, 'stock')
|
||||
(params.root / 'SpeedLimitController').write_bytes(b'0')
|
||||
assert enabled(params, 'tagged', 'vision', {}) # Default-on display still needs its selected Vision producer.
|
||||
assert not vision_control_enabled(params, 'tagged')
|
||||
from openpilot.starpilot.feature_runtime import vision_control_enabled
|
||||
assert requested(params, 'vision')
|
||||
assert enabled(params, long_cp(), 'vision', {})
|
||||
assert enabled(params, stock_cp(), 'vision', {}) # Stock owner may display selected Vision.
|
||||
assert vision_control_enabled(params, long_cp())
|
||||
assert not vision_control_enabled(params, stock_cp())
|
||||
(params.root / 'SpeedLimitController').write_bytes(b'0')
|
||||
assert enabled(params, long_cp(), 'vision', {}) # Default-on display still needs its selected Vision producer.
|
||||
assert not vision_control_enabled(params, long_cp())
|
||||
|
||||
|
||||
def test_explicit_replay_flags_remain_available_for_existing_replay_fixtures():
|
||||
@@ -113,13 +124,11 @@ def test_explicit_replay_flags_remain_available_for_existing_replay_fixtures():
|
||||
assert not enabled(params, None, 'vision', {'SLC_VISION_DEVELOPMENT': '1'})
|
||||
|
||||
|
||||
def test_saved_vision_choice_is_offered_only_for_supported_vehicle_or_explicit_replay():
|
||||
with patch('openpilot.starpilot.speed_limits.vision_gate.ioniq6_settings_capable',
|
||||
side_effect=lambda cp: cp == 'reviewed_ioniq'):
|
||||
assert diagnostic_choice_enabled({}, 'reviewed_ioniq')
|
||||
assert not diagnostic_choice_enabled({}, 'stock_or_other')
|
||||
assert not diagnostic_choice_enabled({})
|
||||
assert diagnostic_choice_enabled({'SLC_REPLAY_RUNTIME': '1', 'SLC_VISION_DEVELOPMENT': '1'})
|
||||
def test_saved_vision_choice_is_offered_for_native_display_vehicle_or_explicit_replay():
|
||||
assert diagnostic_choice_enabled({}, long_cp())
|
||||
assert diagnostic_choice_enabled({}, stock_cp()) # Source visibility grants no speed authority.
|
||||
assert not diagnostic_choice_enabled({})
|
||||
assert diagnostic_choice_enabled({'SLC_REPLAY_RUNTIME': '1', 'SLC_VISION_DEVELOPMENT': '1'})
|
||||
|
||||
|
||||
def test_gm_saved_profiles_require_existing_longitudinal_owner():
|
||||
|
||||
@@ -0,0 +1,226 @@
|
||||
"""Generic admission reaches real Toyota hosts and native planner application."""
|
||||
from types import SimpleNamespace as NS
|
||||
from unittest.mock import Mock, patch
|
||||
|
||||
import pytest
|
||||
|
||||
from opendbc.car.toyota.interface import CarInterface
|
||||
from opendbc.car.toyota.values import CAR
|
||||
from openpilot.common.params import Params
|
||||
from openpilot.selfdrive.controls import plannerd
|
||||
from openpilot.selfdrive.controls.lib.longitudinal_planner import LongitudinalPlanner
|
||||
from openpilot.selfdrive.controls.tests.test_curve_runtime_loop import bus, NOW
|
||||
from openpilot.starpilot.curve_speed.host import CurveHost
|
||||
from openpilot.starpilot.feature_runtime import enabled, slc_runtime_settings, vision_control_enabled
|
||||
from openpilot.starpilot.longitudinal.cruise_ceiling import CruiseCeiling
|
||||
from openpilot.starpilot.longitudinal.profile_document import default_personality_profiles, profile_document
|
||||
from openpilot.starpilot.longitudinal.profile_runtime import ProfileHost
|
||||
from openpilot.starpilot.speed_limits.acceptance import Authority, LongitudinalOwner, Mode
|
||||
from openpilot.starpilot.speed_limits.selection import Source
|
||||
from openpilot.starpilot.speed_limits.vision_gate import diagnostic_choice_enabled
|
||||
from openpilot.starpilot.ui.feature_settings_owner import FeatureSettingsOwner
|
||||
|
||||
|
||||
def toyota():
|
||||
cp = CarInterface.get_non_essential_params(CAR.TOYOTA_RAV4_TSS2)
|
||||
assert cp.openpilotLongitudinalControl and cp.pcmCruise
|
||||
return cp
|
||||
|
||||
|
||||
@pytest.mark.parametrize('brand', ['toyota', 'honda', 'hyundai', 'subaru', 'ford', 'volkswagen', 'mazda', 'nissan', 'tesla', 'rivian', 'future'])
|
||||
def test_generic_admission_is_brand_independent_and_stock_is_display_only(tmp_path, brand):
|
||||
params = Params(str(tmp_path))
|
||||
params.put_bool('CurveSpeedController', True, block=True)
|
||||
params.put_bool('SpeedLimitController', True, block=True)
|
||||
cp = NS(brand=brand, carFingerprint=brand + '_native', openpilotLongitudinalControl=True,
|
||||
pcmCruise=True, passive=False, dashcamOnly=False, notCar=False)
|
||||
for feature in ('conditional', 'curve', 'profile', 'slc', 'vision'):
|
||||
assert enabled(params, cp, feature, {})
|
||||
assert diagnostic_choice_enabled({}, cp)
|
||||
assert vision_control_enabled(params, cp)
|
||||
cp.openpilotLongitudinalControl = False
|
||||
for feature in ('conditional', 'curve', 'profile'):
|
||||
assert not enabled(params, cp, feature, {})
|
||||
settings = slc_runtime_settings(params, cp, {})
|
||||
assert settings.display and not settings.enabled and settings.acceptance.display_only
|
||||
assert enabled(params, cp, 'slc', {}) and enabled(params, cp, 'vision', {})
|
||||
assert not vision_control_enabled(params, cp)
|
||||
for field in ('passive', 'dashcamOnly', 'notCar'):
|
||||
setattr(cp, field, True)
|
||||
for feature in ('conditional', 'curve', 'profile', 'slc', 'vision'):
|
||||
assert not enabled(params, cp, feature, {})
|
||||
setattr(cp, field, False)
|
||||
|
||||
|
||||
def test_factory_defaults_and_hidden_vision_control_producer(tmp_path):
|
||||
params, cp = Params(str(tmp_path)), toyota()
|
||||
settings = slc_runtime_settings(params, cp, {})
|
||||
assert settings.selection.slots[:2] == (Source.VISION, Source.DASHBOARD)
|
||||
assert not settings.enabled and settings.display
|
||||
assert enabled(params, cp, 'conditional', {})
|
||||
assert enabled(params, cp, 'slc', {}) and enabled(params, cp, 'vision', {})
|
||||
assert not enabled(params, cp, 'curve', {})
|
||||
params.put_bool('ShowSpeedLimits', False, block=True)
|
||||
params.put_bool('SpeedLimitController', True, block=True)
|
||||
assert enabled(params, cp, 'vision', {}) and vision_control_enabled(params, cp)
|
||||
|
||||
|
||||
def test_toyota_actual_default_plannerd_constructs_requested_hosts(tmp_path):
|
||||
params, cp = Params(str(tmp_path)), toyota()
|
||||
params.put('CarParams', cp.to_bytes(), block=True)
|
||||
params.put_bool('CurveSpeedController', True, block=True)
|
||||
class EndLoop(Exception):
|
||||
pass
|
||||
sm = Mock()
|
||||
sm.update.side_effect = EndLoop
|
||||
with (patch.object(plannerd, 'Params', return_value=params),
|
||||
patch.object(plannerd, 'config_realtime_process'), patch.object(plannerd, 'set_core_affinity'),
|
||||
patch.object(plannerd.messaging, 'sub_sock'),
|
||||
patch.object(plannerd.messaging, 'SubMaster', return_value=sm) as subscriber,
|
||||
patch.object(plannerd.messaging, 'PubMaster') as publisher,
|
||||
patch.object(plannerd, 'ConditionalPlannerHost', wraps=plannerd.ConditionalPlannerHost) as conditional,
|
||||
patch.object(plannerd, 'ProfileHost', wraps=ProfileHost) as profiles,
|
||||
patch.object(plannerd, 'CurveHost', wraps=CurveHost),
|
||||
patch.dict('os.environ', {}, clear=True)):
|
||||
with pytest.raises(EndLoop):
|
||||
plannerd.main()
|
||||
conditional.assert_called_once()
|
||||
profiles.assert_called_once()
|
||||
assert 'slcState' in publisher.call_args.args[0]
|
||||
assert 'slcCruiseCommand' in publisher.call_args.args[0]
|
||||
for source in ('slcDashboardObservation', 'slcVisionObservation', 'mapdOut'):
|
||||
assert source in subscriber.call_args.args[0]
|
||||
assert plannerd.SavedSlcLifecycle(params, cp, {}).refresh(NOW)[0] is not None
|
||||
|
||||
|
||||
def test_toyota_curve_host_cap_reaches_actual_native_mpc_and_inactive_axis_denies_it():
|
||||
cp, host = toyota(), CurveHost(enabled=True, replay=True)
|
||||
planner = LongitudinalPlanner(cp, init_v=25.0)
|
||||
applied = []
|
||||
for tick in range(1, 31):
|
||||
stamp = NOW + tick * 50_000_000
|
||||
sm = bus(stamp)
|
||||
result = plannerd.update_curve_frame(planner, sm, cp, stamp, host=host)
|
||||
assert result.ceiling_mps is not None
|
||||
applied.append(planner.last_curve_ceiling_applied)
|
||||
assert any(applied)
|
||||
sm.messages['carControl'] = sm.messages['carControl'].as_builder()
|
||||
sm.messages['carControl'].longActive = False
|
||||
result = plannerd.update_curve_frame(planner, sm, cp, stamp, host=host)
|
||||
assert result.ceiling_mps is None and not planner.last_curve_ceiling_applied
|
||||
|
||||
|
||||
def test_toyota_profile_host_and_slc_ceiling_reach_actual_native_planner(tmp_path):
|
||||
params, cp = Params(str(tmp_path)), toyota()
|
||||
params.put_bool('CustomPersonalities', True, block=True)
|
||||
params.put('LongitudinalPersonalityProfiles', profile_document(default_personality_profiles(True), enabled=True), block=True)
|
||||
host = ProfileHost(params)
|
||||
planner = LongitudinalPlanner(cp, init_v=25.0)
|
||||
ceiling = CruiseCeiling(10.0, Authority(Mode.LONGITUDINAL_ONLY, LongitudinalOwner.SYSTEM, False, True, False, False))
|
||||
sm = bus(NOW)
|
||||
target = plannerd.profile_for_frame(host, sm, cp, NOW)
|
||||
assert target is not None
|
||||
planner.update(sm, profile_tuning=target, cruise_ceiling=ceiling, now_ns=NOW)
|
||||
assert planner.last_profile is not None
|
||||
assert planner.last_cruise_ceiling_status == 'applied'
|
||||
sm.messages['carControl'] = sm.messages['carControl'].as_builder()
|
||||
sm.messages['carControl'].longActive = False
|
||||
planner.update(sm, profile_tuning=target, now_ns=NOW)
|
||||
assert planner.last_profile is None
|
||||
owner = FeatureSettingsOwner(params, lambda group: True, vehicle_fingerprint=lambda: cp.carFingerprint, vehicle_params=lambda: cp)
|
||||
assert owner.long_profiles.capability() is not None
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mode_transport():
|
||||
from openpilot.starpilot.conditional_mode.tests import test_status_consumer as fixture
|
||||
with patch.object(fixture, 'CP', toyota()):
|
||||
case = fixture.TestModeTransport('test_actual_selfdrived_owner_preserves_stock_and_separates_axes')
|
||||
case.setUp()
|
||||
try:
|
||||
yield case
|
||||
finally:
|
||||
case.doCleanups()
|
||||
|
||||
|
||||
def test_toyota_actual_selfdrived_acceptance_keeps_current_axis_and_freshness(mode_transport):
|
||||
from openpilot.cereal import messaging
|
||||
from openpilot.selfdrive.selfdrived.selfdrived import SelfdriveD
|
||||
from openpilot.starpilot.conditional_mode.policy import ManualIntent
|
||||
from openpilot.starpilot.conditional_mode.tests.test_projection import MONO, BOOT
|
||||
case = mode_transport
|
||||
drive = SelfdriveD.__new__(SelfdriveD)
|
||||
drive.conditional_replay = False
|
||||
drive.requested_experimental_mode = True
|
||||
drive.update_conditional_mode(None)
|
||||
case.assertTrue(drive.experimental_mode)
|
||||
case.params.put_bool('ExperimentalMode', True, block=True)
|
||||
case.owner.refresh(MONO, force=True)
|
||||
drive.conditional_replay = True
|
||||
drive.conditional_settings = case.owner
|
||||
drive.conditional_consumer = case.consumer
|
||||
drive.conditional_car_state_valid = True
|
||||
drive.initialized = True
|
||||
drive.enabled = True
|
||||
drive.aol_replay = True
|
||||
axes = NS(longitudinal_active=True, lateral_active=False)
|
||||
case.enterContext(patch.object(drive, "aol_axis_decision", axes, create=True))
|
||||
drive.CP = toyota()
|
||||
for index, intent in enumerate((ManualIntent.FORCE_CHILL, ManualIntent.FORCE_EXPERIMENTAL), start=1):
|
||||
now = MONO + index * 50_000_000
|
||||
event, _, sm = case.produce(now, intent=intent)
|
||||
device = messaging.new_message('deviceState', valid=True)
|
||||
device.deviceState.started = True
|
||||
device.deviceState.startedMonoTime = case.drive_id
|
||||
sm.payloads['deviceState'] = messaging.log_from_bytes(device.to_bytes()).deviceState
|
||||
sm.payloads['slcState'] = messaging.log_from_bytes(event.as_reader().as_builder().to_bytes()).slcState
|
||||
for service in ('deviceState', 'slcState'):
|
||||
sm.logMonoTime[service] = now
|
||||
sm.recv_time[service] = now / 1e9
|
||||
sm.seen[service] = sm.alive[service] = sm.valid[service] = True
|
||||
sm.valid['slcState'] = False # Unavailable SLC does not invalidate this nested proposal.
|
||||
drive.sm = sm
|
||||
drive.aol_car_state_log_ns = now - 2_000_000
|
||||
with patch('openpilot.selfdrive.selfdrived.selfdrived.paired_clocks_ns', return_value=(now, BOOT + now - MONO, 1000)):
|
||||
drive.update_conditional_mode(sm['carState'])
|
||||
case.assertEqual(drive.experimental_mode, intent is ManualIntent.FORCE_EXPERIMENTAL)
|
||||
axes.longitudinal_active = False
|
||||
drive.update_conditional_mode(sm['carState'])
|
||||
case.assertTrue(drive.experimental_mode) # Exact existing stock request; no conditional override.
|
||||
axes.longitudinal_active = True
|
||||
|
||||
|
||||
def test_toyota_actual_conditional_planner_host_uses_native_mpc_and_adjacent_radar():
|
||||
# Reuse the serialized fresh-source integration trace with Toyota's unchanged
|
||||
# native CarParams, including pcmCruise=True, rather than a modified CP stub.
|
||||
from openpilot.selfdrive.controls.tests import test_conditional_runtime_loop as fixture
|
||||
case = fixture.TestConditionalRuntimeLoop('test_automatic_chill_joins_native_mpc_radar_and_disabled_slc')
|
||||
with patch.object(fixture, 'CarInterface', NS(get_non_essential_params=lambda candidate: toyota())):
|
||||
case.test_automatic_chill_joins_native_mpc_radar_and_disabled_slc()
|
||||
|
||||
|
||||
def test_toyota_galaxy_long_authority_and_native_curve_rows(tmp_path):
|
||||
from openpilot.starpilot.galaxy.settings import AuthorityContext, _qualified
|
||||
cp = toyota()
|
||||
assert _qualified(AuthorityContext(parked=True, cp=cp, cp_raw=cp.to_bytes()), 'long')
|
||||
params = Params(str(tmp_path))
|
||||
owner = FeatureSettingsOwner(params, lambda group: True, vehicle_fingerprint=lambda: cp.carFingerprint, vehicle_params=lambda: cp)
|
||||
snapshot = owner.snapshot('curve', parked=True, system_long=True, lateral_context=True, metric=True)
|
||||
master = next(row for row in snapshot.rows if row.key == 'CurveSpeedController')
|
||||
assert master.available and 'not supported' not in master.reason
|
||||
|
||||
|
||||
def test_invalid_saved_sources_and_explicit_replay_optins(tmp_path):
|
||||
from pathlib import Path
|
||||
params, cp = Params(str(tmp_path)), toyota()
|
||||
for key, value, feature in (('ConditionalModeConfig', b'{bad', 'conditional'),
|
||||
('CurveSpeedController', b'true', 'curve'),
|
||||
('LongitudinalPersonalityProfiles', b'{bad', 'profile')):
|
||||
Path(params.get_param_path(key)).write_bytes(value)
|
||||
assert not enabled(params, cp, feature, {})
|
||||
for feature, flag in (('conditional', 'CONDITIONAL_MODE_REPLAY_RUNTIME'),
|
||||
('curve', 'CURVE_REPLAY_RUNTIME'), ('profile', 'LONG_PLANNER_REPLAY_RUNTIME'),
|
||||
('slc', 'SLC_REPLAY_RUNTIME')):
|
||||
assert enabled(params, None, feature, {flag: '1'})
|
||||
assert enabled(params, None, 'vision', {'SLC_REPLAY_RUNTIME': '1', 'SLC_VISION_DEVELOPMENT': '1'})
|
||||
assert not enabled(params, None, 'vision', {'SLC_VISION_DEVELOPMENT': '1'})
|
||||
@@ -37,8 +37,7 @@ from openpilot.starpilot.ui.feature_settings_state import (
|
||||
)
|
||||
from openpilot.starpilot.speed_limits.runtime_settings import MPH_TO_MPS
|
||||
from openpilot.starpilot.aol.vehicle import policy_for as aol_policy_for
|
||||
from openpilot.starpilot.longitudinal.ioniq6_start import eligible as ioniq6_long_eligible
|
||||
from opendbc.car.gm.feature_capabilities import longitudinal_supported as gm_longitudinal_supported
|
||||
from openpilot.starpilot.feature_capabilities import longitudinal_supported as feature_longitudinal_supported
|
||||
from openpilot.starpilot.lateral.torque_settings import DOCUMENT_KEY, LEGACY_KEYS
|
||||
from openpilot.starpilot.ui.torque_feature import TorqueFeature
|
||||
from openpilot.starpilot.ui.controller_feature import ControllerFeature, SETUP_ACTION
|
||||
@@ -812,7 +811,7 @@ class FeatureSettingsOwner:
|
||||
title = "Curve Speed Controller"
|
||||
allowed = configurable and system_long and self.authority("long")
|
||||
vehicle_cp = self.vehicle_params()
|
||||
curve_supported = vehicle_cp is not None and (ioniq6_long_eligible(vehicle_cp) or gm_longitudinal_supported(vehicle_cp))
|
||||
curve_supported = feature_longitudinal_supported(vehicle_cp)
|
||||
learning = learning_snapshot(self.params)
|
||||
document_raw = self._raw(CURVE_DOCUMENT_KEY)
|
||||
dependencies = ((CURVE_DOCUMENT_KEY, document_raw),)
|
||||
|
||||
@@ -64,7 +64,7 @@ class LongProfileFeature:
|
||||
def capability(self) -> tuple | None:
|
||||
cp = self.owner.vehicle_params()
|
||||
try:
|
||||
if (cp is None or not cp.carFingerprint or not self.owner.longitudinal_available() or (cp.pcmCruise and not self.owner.configuration_longitudinal()) or
|
||||
if (cp is None or not cp.carFingerprint or not self.owner.longitudinal_available() or
|
||||
cp.passive or cp.notCar or cp.dashcamOnly):
|
||||
return None
|
||||
return (str(cp.carFingerprint), self.owner.longitudinal_available(), bool(cp.pcmCruise and not self.owner.configuration_longitudinal()),
|
||||
|
||||
@@ -668,7 +668,7 @@ class StarShellSession:
|
||||
return bool(cp.openpilotLongitudinalControl and not cp.notCar and not cp.passive and not cp.dashcamOnly)
|
||||
if group in ("aol_wheel", "lane", "torque", "aol"):
|
||||
return not cp.notCar and not cp.passive and not cp.dashcamOnly
|
||||
return bool(cp.openpilotLongitudinalControl and not cp.pcmCruise and not cp.notCar and
|
||||
return bool(cp.openpilotLongitudinalControl and not cp.notCar and
|
||||
not cp.passive and not cp.dashcamOnly)
|
||||
return False
|
||||
|
||||
|
||||
@@ -562,7 +562,7 @@ class RuntimeSnapshotAdapter:
|
||||
cruise = None
|
||||
stock_cruise = bool(display_car_valid and
|
||||
bool(getattr(getattr(display_car, "cruiseState", None), "enabled", False)) and
|
||||
cp is not None and (not cp.openpilotLongitudinalControl or cp.pcmCruise))
|
||||
cp is not None and not cp.openpilotLongitudinalControl)
|
||||
observation = speed_limit_from_message(slc) if slc is not None and (slc.enabled or slc.displayOnly) else SpeedLimitObservation()
|
||||
if slc is None and started:
|
||||
observation = SpeedLimitObservation(kind=ObservationKind.STALE)
|
||||
|
||||
@@ -314,11 +314,20 @@ class ConditionalFeatureTests(unittest.TestCase):
|
||||
with patch.object(shell, "confirmed_offroad", return_value=True), \
|
||||
patch.object(runtime_app.ui_state, "CP", self.cp):
|
||||
self.assertTrue(shell._feature_authority("conditional"))
|
||||
self.assertFalse(shell._feature_authority("long"))
|
||||
self.assertTrue(shell._feature_authority("long"))
|
||||
assert self.cp is not None
|
||||
self.cp.openpilotLongitudinalControl = False
|
||||
self.assertFalse(shell._feature_authority("conditional"))
|
||||
self.assertFalse(shell._feature_authority("long"))
|
||||
self.assertTrue(shell._feature_authority("preferences"))
|
||||
self.cp.openpilotLongitudinalControl = True
|
||||
for field in ('passive', 'dashcamOnly', 'notCar'):
|
||||
with self.subTest(field=field):
|
||||
setattr(self.cp, field, True)
|
||||
self.assertFalse(shell._feature_authority("conditional"))
|
||||
self.assertFalse(shell._feature_authority("long"))
|
||||
self.assertTrue(shell._feature_authority("preferences"))
|
||||
setattr(self.cp, field, False)
|
||||
|
||||
def test_native_saved_actions_use_settings_mode_and_supported_current_car(self):
|
||||
from openpilot.starpilot.ui import runtime_app
|
||||
|
||||
@@ -123,7 +123,7 @@ class FeatureNavigationTests(unittest.TestCase):
|
||||
session.feature_owner = FeatureSettingsOwner(params, session._feature_authority,
|
||||
vehicle_fingerprint=lambda: vehicle.carFingerprint, vehicle_params=lambda: vehicle)
|
||||
with patch.object(runtime_app, 'ui_state', NS(CP=vehicle)):
|
||||
self.assertFalse(session._feature_authority('long'))
|
||||
self.assertTrue(session._feature_authority('long'))
|
||||
self.assertTrue(session._feature_authority('toyota_cruise'))
|
||||
for expected in (True, False):
|
||||
state = session.feature_owner.snapshot('profiles', parked=True, system_long=False,
|
||||
@@ -144,7 +144,7 @@ class FeatureNavigationTests(unittest.TestCase):
|
||||
other = HondaCarInterface.get_non_essential_params(HONDA_CAR.HONDA_CIVIC)
|
||||
with patch.object(runtime_app, 'ui_state', NS(CP=other)):
|
||||
self.assertFalse(session._feature_authority('toyota_cruise'))
|
||||
self.assertFalse(session._feature_authority('long'))
|
||||
self.assertTrue(session._feature_authority('long'))
|
||||
|
||||
def test_native_saved_preferences_need_settings_or_favorite_context_not_cp(self):
|
||||
from openpilot.starpilot.ui import runtime_app
|
||||
@@ -154,9 +154,14 @@ class FeatureNavigationTests(unittest.TestCase):
|
||||
session = runtime_app.StarShellSession.__new__(runtime_app.StarShellSession)
|
||||
session._mode = runtime_app.ShellMode.SETTINGS
|
||||
with patch.object(session, "confirmed_offroad", return_value=True), patch.object(runtime_app, "ui_state", NS(CP=cp)):
|
||||
self.assertTrue(session._feature_authority("long"))
|
||||
self.assertTrue(session._feature_authority("preferences"))
|
||||
cp.openpilotLongitudinalControl = False
|
||||
self.assertFalse(session._feature_authority("long"))
|
||||
self.assertTrue(session._feature_authority("preferences"))
|
||||
cp.openpilotLongitudinalControl = True
|
||||
cp.passive = True
|
||||
self.assertFalse(session._feature_authority("long"))
|
||||
self.assertTrue(session._feature_authority("preferences"))
|
||||
with patch.object(session, "confirmed_offroad", return_value=True), patch.object(runtime_app, "ui_state", NS(CP=None)):
|
||||
self.assertTrue(session._feature_authority("preferences"))
|
||||
|
||||
@@ -2,7 +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.controllers.wheel_actions import ACTIONS, DEFAULTS, KEYS, eligible
|
||||
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
|
||||
from openpilot.starpilot.ui.feature_settings_state import FeatureRow
|
||||
@@ -38,9 +38,12 @@ class WheelFeature:
|
||||
from openpilot.starpilot.car.hyundai.settings import configuration_wheel_policy, configuration_media_supported
|
||||
policy = configuration_wheel_policy(cp) or policy
|
||||
media = configuration_media_supported(cp) or media
|
||||
if key == "LKASButtonControl" and not policy.lkas_button_supported:
|
||||
if not button_reported(cp, key):
|
||||
return ()
|
||||
if key in KEYS[:2] and policy.fixed_cruise_buttons:
|
||||
if key == "LKASButtonControl" and not (policy.lkas_button_supported or policy.auxiliary_lkas_supported):
|
||||
return ()
|
||||
if key in KEYS[:2] and policy.fixed_cruise_buttons and not (
|
||||
key == "LKASButtonControl" and policy.auxiliary_lkas_supported):
|
||||
return ()
|
||||
if key in KEYS[5:8] and cp.brand == 'gm':
|
||||
from opendbc.car.gm.cancel_remap import supported
|
||||
@@ -48,7 +51,8 @@ class WheelFeature:
|
||||
return ()
|
||||
if key in KEYS[8:] and not media:
|
||||
return ()
|
||||
if key in KEYS[:5] and not policy.settings_supported:
|
||||
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():
|
||||
|
||||
Reference in New Issue
Block a user