Restore HKG AOL with auxiliary Pandas and default buttons

opendbc_repo/opendbc/car/hyundai/canfd_owner.py

@@ -1,4 +1,4 @@
-"""Selected safety slot for the exact first-generation captured CANFD owner."""
+"""Selected CANFD safety slot, retaining the exact captured-startup topology."""
 from opendbc.car.structs import CarParams
 from opendbc.car.hyundai.values import CAR, HyundaiFlags
 
@@ -23,6 +23,20 @@ def topology_index(cp):
 
 def config_index(cp):
   owner = topology_index(cp)
+  if owner is None and len(cp.safetyConfigs) == 2:
+    auxiliary, selected = cp.safetyConfigs
+    try:
+      factory = CAR(cp.carFingerprint)
+    except ValueError:
+      return None
+    if (cp.brand == 'hyundai' and factory.config.flags & HyundaiFlags.CANFD and
+        not factory.config.flags & HyundaiFlags.CANFD_ANGLE_STEERING and
+        cp.flags & HyundaiFlags.CANFD and not cp.flags & HyundaiFlags.CANFD_ANGLE_STEERING and
+        not cp.passive and not cp.dashcamOnly and not cp.notCar and
+        cp.steerControlType == CarParams.SteerControlType.torque and
+        auxiliary.safetyModel == CarParams.SafetyModel.noOutput and auxiliary.safetyParam == 0 and
+        selected.safetyModel == CarParams.SafetyModel.hyundaiCanfd):
+      return 1
   if owner != 1:
     return owner
   raw = int(cp.safetyConfigs[owner].safetyParam)

openpilot/starpilot/aol/intent.py

@@ -332,7 +332,7 @@ class AolCardIntent:
       if cancel_pressed:
         # An unassigned cruise CANCEL cannot toggle, arm or renew independent
         # lateral. Keep the existing withdrawal for configured CANCEL actions.
-        if not independent_buttons or any(self.settings.cancel_actions):
+        if not self.independent_cruise_buttons or any(self.settings.cancel_actions):
           self._record_latch_clear('cancel_press')
           self.allowed_latch = False
       elif not self._gesture_neutral_seen:

openpilot/starpilot/aol/tests/test_canfd_auxiliary_owner.py

@@ -0,0 +1,172 @@
+from dataclasses import replace
+from types import SimpleNamespace
+
+import pytest
+
+from opendbc.car import gen_empty_fingerprint, structs
+from opendbc.car.hyundai.canfd_owner import config_index
+from opendbc.car.hyundai.canfd_stock_aol import qualified, qualified_long
+from opendbc.car.hyundai.interface import CarInterface
+from opendbc.car.hyundai.values import CAR, HyundaiFlags
+from openpilot.starpilot.aol.intent import AolSettings
+from openpilot.starpilot.aol.runtime import current_native
+from openpilot.starpilot.aol.tests.test_panda_inventory import NOW, bound_state, sm_for
+from openpilot.starpilot.car.hyundai.aol import create_intent, native_accepts_cp, policy_for
+from openpilot.starpilot.ui.wheel_feature import WheelFeature
+
+
+def factory(car=CAR.KIA_SPORTAGE_5TH_GEN, *, offset=4, alpha=True, lka=0, alternate=True, hybrid=False):
+  fp = gen_empty_fingerprint()
+  bus = offset + int(bool(lka))
+  fp[bus].update({0x1aa if alternate else 0x1cf: 16 if alternate else 8,
+                  0x40: 32, 0x35: 32, 0x175: 24, 0xA0: 24, 0xEA: 24})
+  if lka:
+    fp[offset + 2].update({0x110: 32, 0x362: 32} if lka == 2 else {0x50: 16, 0x2a4: 24})
+  if hybrid:
+    fp[bus][0xfa] = 8
+  fw = [structs.CarParams.CarFw(ecu='adas')] if lka else []
+  return CarInterface.get_params(car, fp, fw, alpha, False, False)
+
+
+def mark(cp):
+  policy = policy_for(cp)
+  assert policy.intent_supported and policy.runtime_supported and policy.settings_supported
+  auxiliary = cp.safetyConfigs[0].to_dict() if len(cp.safetyConfigs) == 2 else None
+  cp.safetyConfigs[-1].safetyParam |= policy.safety_param_addition
+  cp.alternativeExperience |= policy.alternative_experience_addition
+  if auxiliary is not None:
+    assert cp.safetyConfigs[0].to_dict() == auxiliary
+  assert native_accepts_cp(cp, int(cp.safetyConfigs[-1].safetyModel.raw), cp.safetyConfigs[-1].safetyParam)
+
+
+def test_recorded_sportage_layout_arms_from_lkas_without_changing_saved_action():
+  cp = factory()
+  assert cp.flags == 8206 and cp.safetyConfigs[-1].safetyParam == 0x2c
+  assert cp.safetyConfigs[0].safetyModel == 'noOutput' and len(cp.safetyConfigs) == 2
+  assert qualified_long(cp)
+  owner = SimpleNamespace(vehicle_params=lambda: cp, configuration_vehicle=lambda: False,
+                          configuration_longitudinal=lambda: False, longitudinal_available=lambda: True)
+  assert 'Toggle AOL' in WheelFeature(owner).choices('LKASButtonControl')
+  mark(cp)
+  assert config_index(cp) == 1 and cp.safetyConfigs[-1].safetyParam == 0x82c
+  assert cp.alternativeExperience == 32
+  intent = create_intent(cp, AolSettings(True, 0., 0, 0, (0, 0, 0), (0, 0, 0)))
+  cs = structs.CarState(canValid=True, gearShifter='drive', vEgo=10.)
+  cs.cruiseState.available = True
+  intent.update(cs, now_ns=1, fault_active=False)
+  cs.buttonEvents = [structs.CarState.ButtonEvent(type='lkas', pressed=True)]
+  intent.update(cs, now_ns=2, fault_active=False)
+  assert intent.allowed_latch
+  for event in ('cancel', None):
+    cs.buttonEvents = [structs.CarState.ButtonEvent(type=event, pressed=True)] if event else []
+    cs.brakePressed = event is None
+    intent.update(cs, now_ns=3 if event else 4, fault_active=False)
+    assert intent.allowed_latch
+  state = bound_state(cp)
+  assert current_native(sm_for(cp, state), cp, now_ns=NOW) == state
+
+
+@pytest.mark.parametrize('alpha', (False, True))
+@pytest.mark.parametrize('hybrid', (False, True))
+@pytest.mark.parametrize('lka,alternate', ((0, False), (0, True), (1, False), (2, False)))
+def test_sportage_existing_profile_is_independent_of_auxiliary_slot(alpha, hybrid, lka, alternate):
+  for offset in (0, 4):
+    cp = factory(offset=offset, alpha=alpha, hybrid=hybrid, lka=lka, alternate=alternate)
+    assert (qualified_long(cp) if alpha else qualified(cp))
+    mark(cp)
+    assert config_index(cp) == offset // 4
+    assert (qualified_long(cp, marked_only=True) if alpha else qualified(cp, marked_only=True))
+    state = bound_state(cp)
+    assert current_native(sm_for(cp, state), cp, now_ns=NOW) == state
+
+
+@pytest.mark.parametrize('lka', (0, 1, 2))
+def test_existing_stock_canfd_profiles_keep_exact_word_with_auxiliary(lka):
+  checked = []
+  for car in CAR:
+    if not car.config.flags & HyundaiFlags.CANFD or car.config.flags & HyundaiFlags.CANFD_ANGLE_STEERING:
+      continue
+    single = factory(car, offset=0, alpha=False, lka=lka, alternate=False)
+    if not qualified(single):
+      continue
+    cp = factory(car, alpha=False, lka=lka, alternate=False)
+    assert qualified(cp), str(car)
+    assert cp.safetyConfigs[-1].safetyParam == single.safetyConfigs[0].safetyParam
+    mark(cp)
+    state = bound_state(cp)
+    assert current_native(sm_for(cp, state), cp, now_ns=NOW) == state
+    checked.append(car)
+  assert len(checked) > 20
+
+
+@pytest.mark.parametrize('change', ('aux_model', 'aux_word', 'owner_model', 'owner_word', 'order', 'third_slot',
+                                  'unknown', 'brand', 'passive', 'dashcamOnly', 'notCar', 'angle', 'experience'))
+def test_auxiliary_slot_never_bypasses_profile_admission(change):
+  cp = factory()
+  mark(cp)
+  if change == 'aux_model':
+    cp.safetyConfigs[0].safetyModel = 'allOutput'
+  elif change == 'aux_word':
+    cp.safetyConfigs[0].safetyParam = 1
+  elif change == 'owner_model':
+    cp.safetyConfigs[-1].safetyModel = 'hyundai'
+  elif change == 'owner_word':
+    cp.safetyConfigs[-1].safetyParam ^= 1
+  elif change == 'order':
+    cp.safetyConfigs = list(reversed(cp.safetyConfigs))
+  elif change == 'third_slot':
+    cp.safetyConfigs = [*cp.safetyConfigs, structs.CarParams.SafetyConfig(safetyModel='noOutput')]
+  elif change == 'unknown':
+    cp.carFingerprint = 'UNKNOWN_HYUNDAI'
+  elif change == 'brand':
+    cp.brand = 'ford'
+  elif change == 'angle':
+    cp.steerControlType = 'angle'
+  elif change == 'experience':
+    cp.alternativeExperience = 0
+  else:
+    setattr(cp, change, True)
+  assert not qualified_long(cp, marked_only=True)
+  assert not native_accepts_cp(cp, int(structs.CarParams.SafetyModel.hyundaiCanfd), 0x82c)
+
+
+@pytest.mark.parametrize('slot', (0, 1))
+@pytest.mark.parametrize('fault', ('safetyRxChecksInvalid', 'heartbeatLost', 'faults'))
+def test_both_native_inventory_slots_keep_all_live_health_checks(slot, fault):
+  cp = factory()
+  mark(cp)
+  state = bound_state(cp)
+  inventory = list(state.pandaInventory)
+  inventory[slot] = replace(inventory[slot], **{fault: 1 if fault == 'faults' else True})
+  state = replace(state, pandaInventory=tuple(inventory))
+  assert current_native(sm_for(cp, state), cp, now_ns=NOW) is None
+
+
+def test_actual_card_marks_recorded_slot_and_preserves_saved_quick_select():
+  import os
+  from unittest.mock import patch
+  from opendbc.car.hyundai.radar_interface import RadarInterface
+  from openpilot.common.params import Params
+  from openpilot.common.prefix import OpenpilotPrefix
+  from openpilot.selfdrive.car.card import Car
+
+  with OpenpilotPrefix(), patch.dict(os.environ, {'SIMULATION': '0', 'AOL_REPLAY_RUNTIME': '1'}):
+    saved = Params()
+    saved.put_bool('OpenpilotEnabledToggle', True, block=True)
+    saved.put_bool('AlwaysOnLateral', True, block=True)
+    saved.put_bool('AlphaLongitudinalEnabled', True, block=True)
+    saved.put('LKASButtonControl', 11, block=True)
+    saved.put('MainCruiseButtonControl', 0, block=True)
+    cp = factory()
+    auxiliary = cp.safetyConfigs[0].to_dict()
+    card = Car(CI=CarInterface(cp), RI=RadarInterface(cp))
+    try:
+      assert card.aol_qualified and card.aol_card_intent is not None
+      assert card.CP.safetyConfigs[0].to_dict() == auxiliary
+      assert card.CP.safetyConfigs[1].safetyParam == 0x82c and card.CP.alternativeExperience == 32
+      assert card.aol_card_intent.independent_cruise_buttons
+      assert saved.get('LKASButtonControl') == 11
+      assert saved.get('MainCruiseButtonControl') == 0
+      assert saved.get_bool('AlwaysOnLateral') and saved.get_bool('AlphaLongitudinalEnabled')
+    finally:
+      card.vehicle_startup.close()

openpilot/starpilot/aol/tests/test_hkg_cancel_joined.py

@@ -78,7 +78,7 @@ def _invalid_controls_emits_no_commands(received):
   return envelope.which() == 'sendcan' and not envelope.valid and len(envelope.sendcan) == 0
 
 
-def _exercise_join(identity, alternate, incidental, withdrawal, monkeypatch, *, shared_long=False, two_panda=False):
+def _exercise_join(identity, alternate, incidental, withdrawal, monkeypatch, *, shared_long=False, two_panda=False, unmapped=False):
   assert not two_panda or (shared_long and not alternate and identity == CAR.GENESIS_GV70_ELECTRIFIED_1ST_GEN)
   offset = 4 if two_panda else 0
   bootstrap_refusals = []
@@ -90,7 +90,7 @@ def _exercise_join(identity, alternate, incidental, withdrawal, monkeypatch, *,
     saved = Params()
     saved.put_bool('OpenpilotEnabledToggle', True, block=True)
     saved.put_bool('AlwaysOnLateral', True, block=True)
-    saved.put('LKASButtonControl', AOL_TOGGLE, block=True)
+    saved.put('LKASButtonControl', 0 if unmapped else AOL_TOGGLE, block=True)
     saved.put('MainCruiseButtonControl', 0, block=True)
     now = [10_000_000_000]
     holder = None
@@ -157,7 +157,7 @@ def _exercise_join(identity, alternate, incidental, withdrawal, monkeypatch, *,
       sd, controls, drive = case._authority_callers(False, cp=card.CP.as_reader().as_builder(), genuine_events=True)
       # The reusable helper defaults this to None; exercise actual saved mappings.
       sd.aol_settings = read_settings(saved)
-      assert sd.aol_settings.lkas_action == AOL_TOGGLE and sd.aol_settings.main_action == 0
+      assert sd.aol_settings.lkas_action == (0 if unmapped else AOL_TOGGLE) and sd.aol_settings.main_action == 0
       assert sd.CP.to_dict() == card.CP.to_dict() == controls.CP.to_dict()
       assert len(card.CP.safetyConfigs) == (2 if two_panda else 1)
       assert bool(card.CP.openpilotLongitudinalControl) == shared_long
@@ -614,3 +614,9 @@ def test_gv70_single_panda_long_cancel_neighbor(withdrawal, monkeypatch):
   assert _native_variant() == 'debug'
   _exercise_join(CAR.GENESIS_GV70_ELECTRIFIED_1ST_GEN, False, 'cancel', withdrawal,
                  monkeypatch, shared_long=True)
+
+
+@pytest.mark.parametrize('identity,alternate', [
+  (CAR.GENESIS_GV70_ELECTRIFIED_1ST_GEN, False), (CAR.KIA_EV6, True)])
+def test_default_aol_actual_cancel_join(identity, alternate, monkeypatch):
+  _exercise_join(identity, alternate, 'cancel', 'toggle-off', monkeypatch, unmapped=True)

openpilot/starpilot/aol/tests/test_torque_stock_brake_guards.py

@@ -178,8 +178,8 @@ def test_stock_brake_exception_does_not_expand_configuration_admission(identity,
     configs[1].safetyParam = word
     from opendbc.car.hyundai.canfd_owner import config_index
 
-    assert config_index(cp) == (1 if lka else None)
-    assert qualified(cp, marked_only=True) == lka
+    assert config_index(cp) == 1
+    assert qualified(cp, marked_only=True)
   assert not stock_lateral_off(cp, state, native, allow_independent_brake=boundary != 'no_brake_role')
 
 

openpilot/starpilot/car/hyundai/aol.py

@@ -388,8 +388,7 @@ def physical_cancel_lateral_allowed(CP, CS, intent, native, *, settings, previou
         not any(e.type == car.CarState.ButtonEvent.Type.cancel for e in CS.buttonEvents)):
       return False
     ev6_long = CP.carFingerprint == HyundaiCAR.KIA_EV6 and qualified_canfd_long(CP, marked_only=True)
-    return bool((ev6_long or (settings.lkas_action == 9 or settings.main_action == 9) and
-      independent_cruise_buttons_supported(CP, marked_only=True)) and
+    return bool((ev6_long or independent_cruise_buttons_supported(CP, marked_only=True)) and
       previous is not None and previous.lateral_active and previous.native_acknowledged and
       intent is not None and intent.settingsQualified and intent.allowedLatch and intent.lateralArmed and
       not intent.pauseLateral and intent.producerSessionId and intent.carStateLogMonoTime == source_ns and
@@ -434,8 +433,7 @@ def retained_shared_long_cancel_intent(CP, CS, sm, feedback, owner, *, cancel_so
         not owner.allowed_latch or owner.pause_lateral or owner.pause_longitudinal or owner._fault_inhibit):
       return False
     settings = owner.settings
-    if (not settings.enabled or not (ev6_long or settings.lkas_action == 9 or settings.main_action == 9) or
-        any(settings.cancel_actions) or cp_identity(CP) != feedback.cp or
+    if (not settings.enabled or any(settings.cancel_actions) or cp_identity(CP) != feedback.cp or
         feedback.failed or feedback.paused or feedback.pending_event or feedback.qualified or feedback.awaiting or
         feedback.retention_started_ns or feedback.retention_pending or feedback.retention_qualified or
         feedback.source_retention_started_ns or feedback.temporary_host_qualified or

prebuilt.json

@@ -1 +1 @@
-{"protected_inventory_sha256": "b22ffad92c34c5e211191c3782f4ae02f0e7e2c80a08ee0ee1e0215b68cbbddf", "schema_version": 1}
+{"protected_inventory_sha256": "69bc987a943156453c1211b28177b5d828832341441d643d5d7385cbda9a8e12", "schema_version": 1}