From bea3b340c984e8177794ca01e4fca20c285b68a9 Mon Sep 17 00:00:00 2001 From: firestar5683 <168790843+firestar5683@users.noreply.github.com> Date: Mon, 19 Oct 2026 16:16:22 -0500 Subject: [PATCH] Global Feature Availability --- openpilot/selfdrive/controls/plannerd.py | 9 +- openpilot/selfdrive/selfdrived/selfdrived.py | 4 +- .../conditional_mode/tests/test_traffic.py | 7 +- .../starpilot/controllers/mode_actions.py | 8 +- .../tests/test_global_mode_availability.py | 119 +++++++++ .../controllers/tests/test_mode_actions.py | 21 +- .../starpilot/controllers/wheel_actions.py | 40 ++- openpilot/starpilot/feature_capabilities.py | 27 ++ openpilot/starpilot/galaxy/settings.py | 4 +- openpilot/starpilot/lateral/lane_runtime.py | 9 +- .../lateral/tests/test_global_lane_runtime.py | 154 ++++++++++++ .../lateral/tests/test_lane_runtime.py | 16 +- .../starpilot/ui/feature_settings_owner.py | 6 +- openpilot/starpilot/ui/runtime_app.py | 4 +- openpilot/starpilot/ui/runtime_snapshot.py | 12 +- .../ui/tests/test_feature_settings_owner.py | 15 ++ .../tests/test_global_feature_presentation.py | 238 ++++++++++++++++++ openpilot/starpilot/ui/wheel_feature.py | 6 +- 18 files changed, 647 insertions(+), 52 deletions(-) create mode 100644 openpilot/starpilot/controllers/tests/test_global_mode_availability.py create mode 100644 openpilot/starpilot/lateral/tests/test_global_lane_runtime.py create mode 100644 openpilot/starpilot/ui/tests/test_global_feature_presentation.py diff --git a/openpilot/selfdrive/controls/plannerd.py b/openpilot/selfdrive/controls/plannerd.py index 8d1737893a..a12dbfc3e7 100755 --- a/openpilot/selfdrive/controls/plannerd.py +++ b/openpilot/selfdrive/controls/plannerd.py @@ -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, diff --git a/openpilot/selfdrive/selfdrived/selfdrived.py b/openpilot/selfdrive/selfdrived/selfdrived.py index a22474c1d5..8f37ebf14b 100755 --- a/openpilot/selfdrive/selfdrived/selfdrived.py +++ b/openpilot/selfdrive/selfdrived/selfdrived.py @@ -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() diff --git a/openpilot/starpilot/conditional_mode/tests/test_traffic.py b/openpilot/starpilot/conditional_mode/tests/test_traffic.py index 16bafe4dc7..b9641fc74b 100644 --- a/openpilot/starpilot/conditional_mode/tests/test_traffic.py +++ b/openpilot/starpilot/conditional_mode/tests/test_traffic.py @@ -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() diff --git a/openpilot/starpilot/controllers/mode_actions.py b/openpilot/starpilot/controllers/mode_actions.py index e2d3d7b2bb..7a4bfc672f 100644 --- a/openpilot/starpilot/controllers/mode_actions.py +++ b/openpilot/starpilot/controllers/mode_actions.py @@ -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) diff --git a/openpilot/starpilot/controllers/tests/test_global_mode_availability.py b/openpilot/starpilot/controllers/tests/test_global_mode_availability.py new file mode 100644 index 0000000000..178b272283 --- /dev/null +++ b/openpilot/starpilot/controllers/tests/test_global_mode_availability.py @@ -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() diff --git a/openpilot/starpilot/controllers/tests/test_mode_actions.py b/openpilot/starpilot/controllers/tests/test_mode_actions.py index cbb6dfdda3..58805f581f 100644 --- a/openpilot/starpilot/controllers/tests/test_mode_actions.py +++ b/openpilot/starpilot/controllers/tests/test_mode_actions.py @@ -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, diff --git a/openpilot/starpilot/controllers/wheel_actions.py b/openpilot/starpilot/controllers/wheel_actions.py index 8c3e90f12b..e173ba7017 100644 --- a/openpilot/starpilot/controllers/wheel_actions.py +++ b/openpilot/starpilot/controllers/wheel_actions.py @@ -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 diff --git a/openpilot/starpilot/feature_capabilities.py b/openpilot/starpilot/feature_capabilities.py index 89dcbbcdb3..cca845d08e 100644 --- a/openpilot/starpilot/feature_capabilities.py +++ b/openpilot/starpilot/feature_capabilities.py @@ -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 diff --git a/openpilot/starpilot/galaxy/settings.py b/openpilot/starpilot/galaxy/settings.py index bca2a74e5d..28d083a651 100644 --- a/openpilot/starpilot/galaxy/settings.py +++ b/openpilot/starpilot/galaxy/settings.py @@ -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": diff --git a/openpilot/starpilot/lateral/lane_runtime.py b/openpilot/starpilot/lateral/lane_runtime.py index 776b199e08..3a98f978ee 100644 --- a/openpilot/starpilot/lateral/lane_runtime.py +++ b/openpilot/starpilot/lateral/lane_runtime.py @@ -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: diff --git a/openpilot/starpilot/lateral/tests/test_global_lane_runtime.py b/openpilot/starpilot/lateral/tests/test_global_lane_runtime.py new file mode 100644 index 0000000000..17d8588e6c --- /dev/null +++ b/openpilot/starpilot/lateral/tests/test_global_lane_runtime.py @@ -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 diff --git a/openpilot/starpilot/lateral/tests/test_lane_runtime.py b/openpilot/starpilot/lateral/tests/test_lane_runtime.py index ee4c4412a5..e875016fda 100644 --- a/openpilot/starpilot/lateral/tests/test_lane_runtime.py +++ b/openpilot/starpilot/lateral/tests/test_lane_runtime.py @@ -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 diff --git a/openpilot/starpilot/ui/feature_settings_owner.py b/openpilot/starpilot/ui/feature_settings_owner.py index 005605683c..3e9915f753 100644 --- a/openpilot/starpilot/ui/feature_settings_owner.py +++ b/openpilot/starpilot/ui/feature_settings_owner.py @@ -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 diff --git a/openpilot/starpilot/ui/runtime_app.py b/openpilot/starpilot/ui/runtime_app.py index 3479f8a55c..2c40e38342 100644 --- a/openpilot/starpilot/ui/runtime_app.py +++ b/openpilot/starpilot/ui/runtime_app.py @@ -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"): diff --git a/openpilot/starpilot/ui/runtime_snapshot.py b/openpilot/starpilot/ui/runtime_snapshot.py index fc0649fe06..505135cab6 100644 --- a/openpilot/starpilot/ui/runtime_snapshot.py +++ b/openpilot/starpilot/ui/runtime_snapshot.py @@ -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'), diff --git a/openpilot/starpilot/ui/tests/test_feature_settings_owner.py b/openpilot/starpilot/ui/tests/test_feature_settings_owner.py index 16bb33f8d2..ad08173353 100644 --- a/openpilot/starpilot/ui/tests/test_feature_settings_owner.py +++ b/openpilot/starpilot/ui/tests/test_feature_settings_owner.py @@ -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))) diff --git a/openpilot/starpilot/ui/tests/test_global_feature_presentation.py b/openpilot/starpilot/ui/tests/test_global_feature_presentation.py new file mode 100644 index 0000000000..5d636976ab --- /dev/null +++ b/openpilot/starpilot/ui/tests/test_global_feature_presentation.py @@ -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() diff --git a/openpilot/starpilot/ui/wheel_feature.py b/openpilot/starpilot/ui/wheel_feature.py index 745f2a3a32..7a8ba9bc36 100644 --- a/openpilot/starpilot/ui/wheel_feature.py +++ b/openpilot/starpilot/ui/wheel_feature.py @@ -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: