From 697c41b33c60e50388faeac9f4f1892944540d10 Mon Sep 17 00:00:00 2001 From: firestar5683 <168790843+firestar5683@users.noreply.github.com> Date: Sun, 18 Oct 2026 10:11:04 -0500 Subject: [PATCH] Restore GM native health compatibility --- openpilot/starpilot/aol/panda_faults.py | 43 ++++ openpilot/starpilot/aol/runtime.py | 3 +- openpilot/starpilot/aol/tests/test_gm.py | 4 +- .../aol/tests/test_gm_diagnostic_faults.py | 205 ++++++++++++++++++ openpilot/starpilot/aol/transport_pause.py | 7 +- 5 files changed, 257 insertions(+), 5 deletions(-) create mode 100644 openpilot/starpilot/aol/panda_faults.py create mode 100644 openpilot/starpilot/aol/tests/test_gm_diagnostic_faults.py diff --git a/openpilot/starpilot/aol/panda_faults.py b/openpilot/starpilot/aol/panda_faults.py new file mode 100644 index 0000000000..13bb9c1a63 --- /dev/null +++ b/openpilot/starpilot/aol/panda_faults.py @@ -0,0 +1,43 @@ +"""Compatibility policy for two recorded GM configurations; native authority stays separate. + +Register divergence remains a protected-register diagnostic, not a benign-fault +claim. Raw fault reports are retained. Every unlisted configuration/bit blocks. +""" +from openpilot.cereal import log + + +def _diagnostic_mask(cp): + try: + from opendbc.car.gm.aol import qualified_gm + from opendbc.car.gm.values import CAR, is_volt_sdgm_profile, is_bolt_present_no_acc_pedal_profile + if cp.brand != 'gm' or not qualified_gm(cp) or int(cp.alternativeExperience) != 32 or len(cp.safetyConfigs) != 1: + return 0 + word = int(cp.safetyConfigs[0].safetyParam) + if (cp.carFingerprint == CAR.CHEVROLET_VOLT_2019 and word == 0x1005 and + is_volt_sdgm_profile(cp, longitudinal=False) and cp.pcmCruise and not cp.openpilotLongitudinalControl): + return 1 << 18 + if (cp.carFingerprint == CAR.CHEVROLET_BOLT_CC_2018_2021 and word == 0x9D and + is_bolt_present_no_acc_pedal_profile(cp) and cp.openpilotLongitudinalControl and not cp.pcmCruise): + return (1 << 2) | (1 << 3) + except (AttributeError, TypeError, ValueError, OverflowError, IndexError): + pass + return 0 + + +def blocking_fault_mask(mask, cp): + try: + value = int(mask) + return value < 0 or value > 0xFFFFFFFF or bool(value & ~_diagnostic_mask(cp)) + except (TypeError, ValueError, OverflowError): + return True + + +def blocking_fault_list(faults, cp): + try: + allowed = _diagnostic_mask(cp) + named = ((log.PandaState.FaultType.interruptRateCan1, 1 << 2), + (log.PandaState.FaultType.interruptRateCan2, 1 << 3), + (log.PandaState.FaultType.registerDivergent, 1 << 18)) + return any(not any(fault == name and allowed & bit for name, bit in named) for fault in faults) + except (TypeError, ValueError, OverflowError): + return True diff --git a/openpilot/starpilot/aol/runtime.py b/openpilot/starpilot/aol/runtime.py index 061cab8ccf..b0f88bbeb3 100644 --- a/openpilot/starpilot/aol/runtime.py +++ b/openpilot/starpilot/aol/runtime.py @@ -9,6 +9,7 @@ from __future__ import annotations from dataclasses import dataclass from opendbc.car.structs import car +from openpilot.starpilot.aol.panda_faults import blocking_fault_mask from openpilot.starpilot.aol.vehicle import native_admission_key, native_matches_cp, native_profile_supported from openpilot.starpilot.aol.wire import INTENT_SERVICE, SAFETY_SERVICE, IntentState, decode_intent, decode_safety @@ -89,7 +90,7 @@ def native_inventory_matches_cp(state, CP) -> bool: return False owners = [] for index, (slot, config) in enumerate(zip(slots, configs, strict=True)): - if slot.safetyRxChecksInvalid or slot.heartbeatLost or slot.faults: + if slot.safetyRxChecksInvalid or slot.heartbeatLost or blocking_fault_mask(slot.faults, CP): return False if (slot.slotIndex != index or slot.safetyModel != int(config.safetyModel.raw) or slot.safetyParam != int(config.safetyParam) or slot.alternativeExperience != int(CP.alternativeExperience)): diff --git a/openpilot/starpilot/aol/tests/test_gm.py b/openpilot/starpilot/aol/tests/test_gm.py index 85953c1bec..1b83c18beb 100644 --- a/openpilot/starpilot/aol/tests/test_gm.py +++ b/openpilot/starpilot/aol/tests/test_gm.py @@ -1106,7 +1106,9 @@ class TestGmAol(unittest.TestCase): policy = policy_for(cp) self.assertTrue(policy.intent_supported, (cp.carFingerprint, cp.safetyConfigs[-1].safetyParam)) selected = create_intent(cp, settings, policy) - reference = AolCardIntent(settings, explicit_latch=policy.explicit_latch) + from openpilot.starpilot.car.hyundai.aol import independent_cruise_buttons_supported + reference = AolCardIntent(settings, explicit_latch=policy.explicit_latch, + independent_cruise_buttons=cp.brand == 'hyundai' and independent_cruise_buttons_supported(cp)) self.assertIs(type(selected), AolCardIntent) for main, permanent, temporary, gear in ((True, False, False, 'drive'), (True, False, True, 'drive'), diff --git a/openpilot/starpilot/aol/tests/test_gm_diagnostic_faults.py b/openpilot/starpilot/aol/tests/test_gm_diagnostic_faults.py new file mode 100644 index 0000000000..58fb6c0103 --- /dev/null +++ b/openpilot/starpilot/aol/tests/test_gm_diagnostic_faults.py @@ -0,0 +1,205 @@ +"""Source-prepared exact GM admission regressions; root executes after candidate review.""" +from dataclasses import replace +from types import SimpleNamespace + +import pytest +from opendbc.car.gm.tests.test_volt_sdgm_control import sdgm_params +from opendbc.car.gm.tests.test_bolt_pedal import params +from opendbc.car.gm.values import CAR +from openpilot.starpilot.aol.runtime import current_native, native_inventory_matches_cp, decide_axes +from openpilot.starpilot.aol.wire import PandaSlot, SafetyState +from openpilot.starpilot.aol.tests.test_panda_inventory import sm_for +from openpilot.starpilot.aol.tests.test_runtime import car_state + +NOW = 10_000_000_000 + + +def owner(label): + cp = sdgm_params(alpha=False, sascm=True, radar=False) if label == 'volt' else params( + CAR.CHEVROLET_BOLT_CC_2018_2021, pedal=True, camera=True) + cp.alternativeExperience = 32 + assert cp.safetyConfigs[0].safetyParam == (0x1005 if label == 'volt' else 0x9D) + assert cp.openpilotLongitudinalControl == (label == 'bolt') + assert cp.pcmCruise == (label == 'volt') + config = cp.safetyConfigs[0] + slot = PandaSlot(0, label, config.safetyModel.raw, config.safetyParam, 32, False) + state = SafetyState(1, True, NOW, NOW + 200_000_000, config.safetyModel.raw, config.safetyParam, + False, False, False, False, label, 'session', NOW, (slot,)) + return cp, state + + +@pytest.mark.parametrize('label', ('volt', 'bolt')) +@pytest.mark.parametrize('mask', (0, 1 << 2, 1 << 3, 1 << 18, 12)) +def test_diagnostic_receipt_does_not_invent_native_authority(label, mask): + cp, state = owner(label) + state = replace(state, pandaInventory=(replace(state.pandaInventory[0], faults=mask),)) + admitted = mask == 0 or (label == 'volt' and mask == 1 << 18) or (label == 'bolt' and mask in (1 << 2, 1 << 3, 12)) + assert native_inventory_matches_cp(state, cp) == admitted + accepted = current_native(sm_for(cp, state), cp, now_ns=NOW, axis_session_id='session') + if not admitted: + assert accepted is None + return + assert accepted == state + intent = SimpleNamespace(allowedLatch=True, pauseLateral=False, pauseLongitudinal=False) + decision = decide_axes(standard_lateral=True, standard_longitudinal=cp.openpilotLongitudinalControl, + intent=intent, native=accepted, car_state=car_state(), initialized=True, + model_ready=True, no_entry=False, immediate_disable=False, dm_lockout=False, pause_brake_mps=0.) + assert not decision.lateral_active and not decision.longitudinal_active + + +@pytest.mark.parametrize('label', ('volt', 'bolt')) +@pytest.mark.parametrize('bit', (0, 1, 4, 5, 7, 8, 15, 20, 21, 22, 23, 25, 26, 27, 31)) +def test_authority_faults_and_unknown_bits_still_block(label, bit): + cp, state = owner(label) + baseline = replace(state, pandaInventory=(replace(state.pandaInventory[0], faults=(1 << 18 if label == 'volt' else 12)),)) + assert current_native(sm_for(cp, baseline), cp, now_ns=NOW, axis_session_id='session') == baseline + state = replace(baseline, pandaInventory=(replace(baseline.pandaInventory[0], faults=baseline.pandaInventory[0].faults | (1 << bit)),)) + assert current_native(sm_for(cp, state), cp, now_ns=NOW, axis_session_id='session') is None + + +@pytest.mark.parametrize('label', ('volt', 'bolt')) +@pytest.mark.parametrize('bad', ('rx', 'heartbeat', 'word', 'ae', 'serial', 'session', 'expired')) +def test_diagnostic_bits_cannot_bypass_other_vetoes(label, bad): + cp, state = owner(label) + slot = replace(state.pandaInventory[0], faults=(1 << 18 if label == 'volt' else 12)) + baseline = replace(state, pandaInventory=(slot,)) + assert current_native(sm_for(cp, baseline), cp, now_ns=NOW, axis_session_id='session') == baseline + mutations = {'rx': {'safetyRxChecksInvalid': True}, 'heartbeat': {'heartbeatLost': True}, + 'word': {'safetyParam': slot.safetyParam ^ 1}, 'ae': {'alternativeExperience': 0}, + 'serial': {'hardwareSerial': 'wrong'}} + state = replace(state, pandaInventory=(replace(slot, **mutations.get(bad, {})),)) + assert current_native(sm_for(cp, state), cp, now_ns=NOW + (200_000_001 if bad == 'expired' else 0), + axis_session_id='wrong' if bad == 'session' else 'session') is None + + +def test_other_brands_keep_existing_all_fault_veto(): + from openpilot.starpilot.aol.panda_faults import blocking_fault_mask, blocking_fault_list + from openpilot.cereal import log + cp = SimpleNamespace(brand='hyundai') + assert blocking_fault_mask(12, cp) + assert blocking_fault_list([log.PandaState.FaultType.interruptRateCan1], cp) + + +@pytest.mark.parametrize('word', (0x1405, 0x5007, 0x5407, 0x5087, 0xD107, 0xE500, 0x205, 0x4007, 0x20)) +def test_other_volt_configs_do_not_borrow_stock_register_policy(word): + from openpilot.starpilot.aol.panda_faults import blocking_fault_mask + cp, _ = owner('volt') + cp.safetyConfigs[0].safetyParam = word + assert blocking_fault_mask(1 << 18, cp) + + +@pytest.mark.parametrize('identity', (CAR.CHEVROLET_BOLT_CC_2017, CAR.CHEVROLET_BOLT_CC_2022_2023, + CAR.CHEVROLET_BOLT_ACC_2022_2023_PEDAL)) +def test_other_bolt_generations_do_not_borrow_irq_policy(identity): + from openpilot.starpilot.aol.panda_faults import blocking_fault_mask + cp = params(identity, pedal=True, camera=True) + cp.alternativeExperience = 32 + assert blocking_fault_mask(12, cp) + + +@pytest.mark.parametrize('label', ('volt', 'bolt')) +def test_latest_raw_faults_use_same_exact_policy_and_retain_native_veto(label): + from openpilot.cereal import messaging + from openpilot.starpilot.aol.transport_pause import healthy_transport + cp, state = owner(label) + allowed = 'registerDivergent' if label == 'volt' else 'interruptRateCan2' + mask = 1 << 18 if label == 'volt' else 1 << 3 + state = replace(state, pandaInventory=(replace(state.pandaInventory[0], faults=mask),)) + sm = sm_for(cp, state) + assert current_native(sm, cp, now_ns=NOW, axis_session_id='session') == state + event = messaging.new_message('pandaStates', 1, valid=True, logMonoTime=NOW) + event.pandaStates[0] = sm['pandaStates'][0] + event.pandaStates[0].faults = [allowed] + sm.update_msgs(NOW / 1e9, [event.as_reader()]) + assert healthy_transport(sm, cp, car_state(), 'session', NOW) + event.pandaStates[0].faults = [allowed, 'relayMalfunction'] + sm.update_msgs(NOW / 1e9, [event.as_reader()]) + assert not healthy_transport(sm, cp, car_state(), 'session', NOW) + event.pandaStates[0].faults = [allowed] + event.pandaStates[0].heartbeatLost = True + sm.update_msgs(NOW / 1e9, [event.as_reader()]) + assert not healthy_transport(sm, cp, car_state(), 'session', NOW) + + +@pytest.mark.parametrize('label', ('volt', 'bolt')) +def test_raw_enum_policy_does_not_borrow_other_config_diagnostics(label): + from openpilot.cereal import log + from openpilot.starpilot.aol.panda_faults import blocking_fault_list + cp, _ = owner(label) + allowed = [log.PandaState.FaultType.registerDivergent] if label == 'volt' else [ + log.PandaState.FaultType.interruptRateCan1, log.PandaState.FaultType.interruptRateCan2] + assert not blocking_fault_list(allowed, cp) + denied = (log.PandaState.FaultType.interruptRateCan1 if label == 'volt' else log.PandaState.FaultType.registerDivergent) + assert blocking_fault_list([*allowed, denied], cp) + assert blocking_fault_list([*allowed, log.PandaState.FaultType.interruptRateCan3], cp) + assert blocking_fault_list([*allowed, log.PandaState.FaultType.relayMalfunction], cp) + assert blocking_fault_list([*allowed, 31], cp) + + +@pytest.mark.parametrize('label', ('volt', 'bolt')) +def test_real_native_permission_and_brake_withdrawal_remain_authoritative(label): + from opendbc.can import CANPacker + from opendbc.car import Bus + from opendbc.car.gm.interface import CarInterface + from opendbc.car.gm.values import DBC, CruiseButtons + from opendbc.car.gm.bolt_cc import button_bytes + from opendbc.car.gm.gmcan import pedal_crc + from opendbc.car.gm.tests.test_bolt_cc import feed + from opendbc.safety.tests.libsafety import libsafety_py + cp, state = owner(label) + ci = CarInterface(cp) + packer = CANPacker(DBC[cp.carFingerprint][Bus.pt]) + safety = libsafety_py.libsafety + safety.init_tests() + safety.set_alternative_experience(32) + assert safety.set_safety_hooks(cp.safetyConfigs[0].safetyModel.raw, cp.safetyConfigs[0].safetyParam) == 0 + for tick in range(32): + now = NOW + tick * 10_000_000 + _, packets = feed(SimpleNamespace(update=lambda _: None), packer, now, counter=tick % 4, + active=label == 'volt', speed=20., camera=True, brake=tick == 31) + packets = [packet for packet in packets if packet[0] != 0x1E1] + button = CruiseButtons.DECEL_SET if tick == 20 else CruiseButtons.UNPRESS + packets.append((0x1E1, button_bytes(button, tick % 4), 0)) + if label == 'volt': + # Stock SDGM requires camera ACC status and its selected PT brake source. + packets = [packet for packet in packets if packet[0] != 0xBE] + packets.append(packer.make_can_msg('ECMAcceleratorPos', 0, {'BrakePedalPos': 12 if tick == 31 else 0})) + packets.append(packer.make_can_msg('ASCMActiveCruiseControlStatus', 2, + {'ACCCruiseState': 3, 'ACCSpeedSetpoint': 60})) + if label == 'bolt': + packets = [packet for packet in packets if packet[0] != 0x1F5] + packets.append(packer.make_can_msg('ECMPRDNL2', 0, {'PRNDL2': 6})) + sensor = bytearray(packer.make_can_msg('GAS_SENSOR', 0, { + 'INTERCEPTOR_GAS': 0., 'INTERCEPTOR_GAS2': 0., 'STATE': 0, 'COUNTER_PEDAL': tick % 16})[1]) + sensor[5] = pedal_crc(sensor) + packets.append((0x201, bytes(sensor), 0)) + parsed = ci.update([(now, packets)]) + safety.set_timer(now // 1000) + for address, data, bus in packets: + assert safety.safety_rx_hook(libsafety_py.make_CANPacket(address, bus, data)) + safety.safety_tick() + # Transport instrumentation supplies heartbeat/request, never a permission or controlsAllowed setter. + safety.set_aol_test_heartbeat(True) + request = 1 if label == 'volt' else 3 + safety.aol_set_host_request(request) + mask = safety.aol_get_permission_mask() + if tick == 30: + assert parsed.canValid + assert mask == request + if tick == 31: + assert parsed.brakePressed + assert not mask & 2 + slot = replace(state.pandaInventory[0], controlsAllowed=bool(safety.get_controls_allowed()), + faults=1 << 18 if label == 'volt' else 12) + receipt = replace(state, observedMonoTime=now, validUntilMonoTime=now + 200_000_000, + lateralAllowed=bool(mask & 1), longitudinalAllowed=bool(mask & 2), + requestedLateral=bool(request & 1), requestedLongitudinal=bool(request & 2), + pandaInventory=(slot,)) + # Admission of a classified diagnostic does not overwrite native LONG withdrawal. + assert native_inventory_matches_cp(receipt, cp) + intent = SimpleNamespace(allowedLatch=True, pauseLateral=False, pauseLongitudinal=False) + decision = decide_axes(standard_lateral=True, standard_longitudinal=cp.openpilotLongitudinalControl, + intent=intent, native=receipt, car_state=parsed, initialized=True, model_ready=True, + no_entry=False, immediate_disable=False, dm_lockout=False, pause_brake_mps=0.) + assert not decision.longitudinal_active + safety.set_alternative_experience(0) diff --git a/openpilot/starpilot/aol/transport_pause.py b/openpilot/starpilot/aol/transport_pause.py index b695361968..11b6c4376e 100644 --- a/openpilot/starpilot/aol/transport_pause.py +++ b/openpilot/starpilot/aol/transport_pause.py @@ -1,5 +1,6 @@ """Reason-qualified intent retention; this module never grants an axis.""" from openpilot.cereal import log +from openpilot.starpilot.aol.panda_faults import blocking_fault_mask, blocking_fault_list from openpilot.starpilot.aol.runtime import AXIS_MAX_AGE_NS, SAFETY_MAX_AGE_NS, current_native, native_inventory_matches_cp from openpilot.starpilot.aol.wire import SafetyState, decode_intent @@ -29,7 +30,7 @@ def _configured_transport_native(sm, cp, session, now_ns, native=None, *, requir # current_native already checked this exact inventory in this call. # Supplied native values still need structural validation here. if coherent and (not validated_here and not native_inventory_matches_cp(native, cp) or - any(p.safetyRxChecksInvalid or p.heartbeatLost or p.faults or + any(p.safetyRxChecksInvalid or p.heartbeatLost or blocking_fault_mask(p.faults, cp) or (require_longitudinal and native.requestedLongitudinal and str(config.safetyModel) not in ('silent', 'noOutput') and not p.controlsAllowed) for p, config in zip(native.pandaInventory, cp.safetyConfigs, strict=True))): @@ -45,7 +46,7 @@ def _configured_transport_native(sm, cp, session, now_ns, native=None, *, requir return None return native if all(int(p.safetyModel.raw) == int(c.safetyModel.raw) and p.safetyParam == c.safetyParam and p.alternativeExperience == cp.alternativeExperience and not p.safetyRxChecksInvalid and - not p.heartbeatLost and not p.faults and + not p.heartbeatLost and not blocking_fault_list(p.faults, cp) and (not require_longitudinal or not native.requestedLongitudinal or str(c.safetyModel) in ('silent', 'noOutput') or p.controlsAllowed) for p, c in zip(pandas, cp.safetyConfigs, strict=True)) else None except (AttributeError, KeyError, IndexError, TypeError, ValueError): @@ -601,7 +602,7 @@ class TransportPauseFeedback: # every latest raw configuration/fault veto and exact producer-read inventory. current_pandas = all(int(p.safetyModel.raw) == int(c.safetyModel.raw) and p.safetyParam == c.safetyParam and p.alternativeExperience == cp.alternativeExperience and not p.safetyRxChecksInvalid and - not p.heartbeatLost and not p.faults and + not p.heartbeatLost and not blocking_fault_list(p.faults, cp) and (str(c.safetyModel) not in ('silent', 'noOutput') or not p.controlsAllowed) for p, c in zip(pandas, cp.safetyConfigs, strict=True)) return native if current_pandas else None