Build Bolt pedal AOL recovery firmware

This commit is contained in:
firestar5683
2026-10-20 11:04:30 -05:00
46 changed files with 342 additions and 25 deletions
+3 -1
View File
@@ -236,7 +236,9 @@ static bool gm_cruise_status_owner_ready(void) {
ready = gm_cruise_status_main && gm_cruise_status_drive && gm_cruise_status_eps &&
gm_pedal_sensor_current();
}
return ready && !brake_pressed && !gas_pressed && !regen_braking;
// Bolt's 0x3D1 replacement is dashboard status only. Accelerator override
// still withdraws pedal/paddle actuation in their independent TX gates.
return ready && !brake_pressed && (!gas_pressed || gm_cruise_status_bolt) && !regen_braking;
}
static void gm_rx_hook(const CANPacket_t *msg) {
@@ -1008,6 +1008,29 @@ class TestGmBoltPedalSafety(unittest.TestCase):
self.assertEqual(self.safety.safety_fwd_hook(0, 0x3D1), 2)
self.assertFalse(self.safety.safety_tx_hook(self.packet((0x3D1, status[1], 2))))
def test_accelerator_override_keeps_only_dashboard_replacement(self):
from opendbc.car.gm import gmcan
for word in (0xBD, 0x9D, 0x19D, 0xE700, 0xE701, 0xE702):
with self.subTest(word=word):
self.safety.set_safety_hooks(CarParams.SafetyModel.gm, word)
self.safety.init_tests()
self.safety.reset_recorded_can()
self.status_sources(1_000_000)
raw = self.stock('ECMCruiseControl', {})
self.safety.safety_rx_hook(raw)
self.assertTrue(self.safety.safety_rx_hook(self.sensor(1, gas=30.)))
self.assertTrue(self.safety.get_gas_pressed_prev())
status = gmcan.create_ecm_cruise_control_command(self.packer, 0, True, 80.)
self.assertFalse(self.safety.safety_tx_hook(self.packet(status)))
self.assertEqual(self.safety.safety_fwd_hook(0, 0x3D1), -1)
self.safety.safety_rx_hook(raw)
self.assertEqual(self.recorded(), [(0x3D1, 0, status[1])])
self.safety.set_controls_allowed(True)
self.assertFalse(self.safety.safety_tx_hook(self.packet(create_pedal_command(self.packer, .2, 0))))
self.safety.safety_tx_hook(self.packet(create_bolt_regen_paddle(self.packer, True)))
self.safety.safety_rx_hook(self.stock("EBCMRegenPaddle", {}))
self.assertEqual(self.recorded(), [(0x3D1, 0, status[1])])
def test_status_physical_jitter_does_not_halve_output_cadence(self):
from opendbc.car.gm import gmcan
for word in (0xBD, 0x9D, 0x19D, 0xE700, 0xE701, 0xE702):
+2 -1
View File
@@ -85,7 +85,8 @@ class VCruiseHelper:
if not self.CP.pcmCruise:
for button in CS.buttonEvents:
self.pending_resume_ns = 0
if (not enabled and CS.cruiseState.available and not button.pressed and
resume_edge = not button.pressed or self.CP.brand == 'gm' and CS.buttonEnable
if (not enabled and CS.cruiseState.available and resume_edge and
button.type in (ButtonType.accelCruise, ButtonType.resumeCruise) and self.v_cruise_initialized):
self.pending_resume_ns = now_ns
+22 -1
View File
@@ -1208,6 +1208,10 @@ class SelfdriveD:
previous = self.aol_axis_decision
was_enabled = self.enabled
self.aol_axis_decision = AxisDecision()
if healthy_expiry and not was_enabled and self.cold_gm_main_readiness(CS, intent, now_ns):
self.aol_fault_reason = 'availabilityPause'
self.aol_publication_retained_intent = intent
return
pause = getattr(self, 'aol_availability_pause', None)
if (healthy_expiry and pause is not None and intent is not None and
getattr(self, 'aol_fault_reason', 'none') in ('none', 'availabilityPause') and
@@ -1359,6 +1363,8 @@ class SelfdriveD:
if source_clock_scoped:
self._expire_axis_source(CS, now_ns)
self._expire_axis_continuity(CS, now_ns)
if self.aol_axis_decision.lateral_active or self.aol_axis_decision.longitudinal_active:
self.aol_ever_active = True
reason = getattr(self, 'aol_fault_reason', 'none')
event_now_ns = now_ns
ce_send = None
@@ -1600,6 +1606,18 @@ class SelfdriveD:
sample.vEgo < (self.aol_settings.pause_brake_mps if self.aol_settings is not None else 0.0)
for sample, _ in getattr(self, 'car_state_samples', ())))
def cold_gm_main_readiness(self, CS, intent, now_ns):
from openpilot.starpilot.car.gm.aol import cold_main_readiness
from openpilot.starpilot.aol.transport_pause import _healthy_transport_native
if (self.CP.brand != 'gm' or self.enabled or self.active or getattr(self, 'aol_ever_active', False) or self.aol_authority_lost or
not self.events.contains(ET.SOFT_DISABLE) or self.events.contains(ET.IMMEDIATE_DISABLE) or
intent is None or not intent.settingsQualified or not intent.allowedLatch or not intent.lateralArmed or
not 0 < intent.carStateLogMonoTime <= intent.observedMonoTime <= now_ns <= intent.validUntilMonoTime or
now_ns - intent.carStateLogMonoTime > SAFETY_MAX_AGE_NS):
return False
native = _healthy_transport_native(self.sm, self.CP, CS, self.aol_session_id, now_ns)
return cold_main_readiness(self.CP, CS, native, self.events.to_msg())
def ordinary_native_long_withdrawal(self, CS, native, previous, requested, intent, *, source_current, now_ns):
"""Keep the existing ordinary Panda mismatch owner for a LONG-only drop.
@@ -1870,7 +1888,8 @@ class SelfdriveD:
source_ns=self.aol_car_state_log_ns, now_ns=now_ns, axis_session_id=self.aol_session_id,
pending_lateral_request=not previous.lateral_active and not self.aol_authority_lost))
for event in self.events.to_msg())
self.aol_fault_reason = ('availabilityPause' if ordinary_availability else
cold_main_wait = self.cold_gm_main_readiness(CS, intent, now_ns)
self.aol_fault_reason = ('availabilityPause' if ordinary_availability or cold_main_wait else
'critical' if self.aol_authority_lost or self.events.contains(ET.IMMEDIATE_DISABLE) or
aol_hard_soft_fault and not availability else
'availabilityPause' if availability else 'transportPause' if pause else 'none')
@@ -1909,6 +1928,8 @@ class SelfdriveD:
# Preserve the exact current SOFT event and normal countdown. A live
# standard-axis proof describes healthy current authority, not critical.
self.aol_fault_reason = 'none'
if cold_main_wait:
self.aol_axis_decision = AxisDecision()
if self.aol_ordinary_source_pause:
self.aol_axis_decision = self.ordinary_source_withdrawal()
elif availability:
@@ -106,6 +106,23 @@ def baseline():
return card, feedback, native
@pytest.mark.parametrize('health_allowed', [False, True])
def test_native_long_withdrawal_before_parser_keeps_existing_card_intent(health_allowed):
with OpenpilotPrefix():
card, feedback, prior = baseline()
now = NOW + 10_000_000
cs = parse(card, now, counter=1)
native = replace(prior, observedMonoTime=now, validUntilMonoTime=now + 200_000_000,
longitudinalAllowed=False, pandaInventory=(replace(prior.pandaInventory[0], controlsAllowed=health_allowed),))
sm, _ = peers(card.CP, card.slc_producer_session, now, controls=health_allowed, native=native, sequence=2)
# The host's last full-axis result precedes the native paddle/brake edge.
reset = feedback.observe(sm, card.CP, cs, now, latched=True,
intent_producer_session=card.slc_producer_session)
card.aol_card_intent.update(cs, fault_active=reset, now_ns=now)
assert not reset and not feedback.failed and card.aol_card_intent.allowed_latch
assert not native.longitudinalAllowed
@pytest.mark.parametrize('gesture', ['brake', 'paddle', 'cancel'])
def test_real_physical_withdrawal_retains_latch_with_held_long_wire(gesture):
with OpenpilotPrefix():
@@ -0,0 +1,169 @@
"""Route 18e regressions through actual parser, Card, SelfdriveD and Controls."""
from contextlib import ExitStack
from unittest.mock import patch
import time
import pytest
from opendbc.can import CANPacker
from opendbc.car import Bus
from opendbc.car.structs import car
from opendbc.car.gm.bolt_cc import button_bytes
from opendbc.car.gm.tests.test_bolt_pedal import params
from opendbc.car.gm.values import CAR, CruiseButtons, DBC
from openpilot.cereal import messaging
from openpilot.common.params import Params
from openpilot.common.prefix import OpenpilotPrefix
from openpilot.selfdrive.car.cruise import VCruiseHelper
from openpilot.starpilot.aol.tests import test_ioniq_availability as fixture
from openpilot.starpilot.aol.tests import test_gm as gm_fixtures
from openpilot.starpilot.aol.tests.test_gm_held_native import NOW, parse as physical_packets
from openpilot.starpilot.aol.tests.test_seatbelt_cold_start import deliver_card as deliver
from openpilot.selfdrive.selfdrived.events import EventName
def deliver_card(card, sd, controls, cs, tick):
# Both clocks are explicit fixture inputs, including on macOS hosts.
with patch.object(time, 'CLOCK_BOOTTIME', getattr(time, 'CLOCK_BOOTTIME', time.CLOCK_MONOTONIC), create=True), \
patch.object(time, 'clock_gettime_ns', return_value=NOW + tick * 10_000_000):
deliver(card, sd, controls, cs, tick)
def parse(card, now, counter=0, *, packets_only=False):
rows = physical_packets(card, now, counter=counter, packets_only=True)
packer = CANPacker(DBC[card.CP.carFingerprint][Bus.pt])
rows.append(packer.make_can_msg('ASCMActiveCruiseControlStatus', 2, {'ACCCruiseState': 0}))
if packets_only:
return rows
if now == NOW:
card.CI.update([(now - 1_000_000, rows)])
cs = card.CI.update([(now, rows)])
assert cs.canValid and not cs.canTimeout
return cs
@pytest.mark.parametrize('event', [EventName.commIssue, EventName.commIssueAvgFreq, EventName.locationdTemporaryError,
EventName.controlsMismatch, EventName.processNotRunning])
def test_cold_main_waits_for_host_readiness_but_keeps_real_faults(event):
with OpenpilotPrefix(), ExitStack() as cleanup:
saved = Params()
saved.put_bool('AlwaysOnLateral', True, block=True)
card = gm_fixtures.TestGmAol.card(params(CAR.CHEVROLET_BOLT_CC_2018_2021, pedal=True, camera=True), saved)
cleanup.callback(card.vehicle_startup.close)
case = fixture.IoniqAvailabilityTests(methodName='runTest')
cleanup.callback(case.doCleanups)
sd, controls, drive = case._authority_callers(False, cp=card.CP.as_reader().as_builder(), genuine_events=True)
card.slc_producer_session = 'authority-card'
cs = parse(card, NOW)
card.aol_card_intent.update(cs, fault_active=False, now_ns=NOW)
original_events = sd.update_events
axis, state, command = drive(0, observed_state=cs, native_request_lat=False, native_request_long=False,
allowed=False, long_allowed=False, critical_event=event)
temporary = event in (EventName.commIssue, EventName.commIssueAvgFreq, EventName.locationdTemporaryError)
assert str(axis.faultReason) == ('availabilityPause' if temporary else 'critical')
assert not command.latActive and not command.longActive and not state.enabled
deliver_card(card, sd, controls, cs, 0)
assert card.aol_card_intent.allowed_latch == temporary
if temporary:
# The source's original 30ms output deadline still expires during startup.
sd._expire_axis_source(cs, NOW + 31_000_000)
assert sd.aol_car_state_log_ns == 0 and str(sd.aol_fault_reason) == 'availabilityPause'
assert EventName.controlsMismatch not in sd.events.names
sd.update_events = original_events
axis, _, command = drive(4, native_request_long=False, native_request_lat=False, allowed=False)
assert axis.desiredLateral and not axis.lateralActive and not command.latActive and not command.longActive
deliver_card(card, sd, controls, parse(card, NOW + 40_000_000, counter=1), 4)
assert card.aol_card_intent.allowed_latch
axis, _, command = drive(5, native_request_long=False)
assert axis.lateralActive and command.latActive and not command.longActive
deliver_card(card, sd, controls, parse(card, NOW + 50_000_000, counter=2), 5)
assert card.aol_card_intent.allowed_latch
@pytest.mark.parametrize('candidate', [CAR.CHEVROLET_BOLT_CC_2017, CAR.CHEVROLET_BOLT_CC_2018_2021,
CAR.CHEVROLET_BOLT_CC_2022_2023, CAR.CHEVROLET_BOLT_ACC_2022_2023_PEDAL])
def test_gm_resume_press_survives_neutral_samples_before_control_ack(candidate):
with OpenpilotPrefix(), ExitStack() as cleanup:
card = gm_fixtures.TestGmAol.card(params(candidate, pedal=True, camera=True), Params())
cleanup.callback(card.vehicle_startup.close)
card.v_cruise_helper = VCruiseHelper(card.CP)
card.v_cruise_helper.v_cruise_kph = 92.8
card.CC_prev = car.CarControl(enabled=False)
card.CS_prev = parse(card, NOW)
for tick in range(3):
now = NOW + (tick + 1) * 10_000_000
packets = parse(card, now, counter=tick % 4, packets_only=True)
packets = [packet for packet in packets if packet[0] != 0x1E1]
packets.append((0x1E1, button_bytes(CruiseButtons.RES_ACCEL, tick % 4), 0))
cs = card.CI.update([(now, packets)])
assert cs.buttonEnable == (tick == 0)
control = messaging.new_message('carControl', valid=True, logMonoTime=now)
control.carControl.enabled = tick == 2
card.sm.update_msgs(now / 1e9, [control.as_reader()])
with patch('openpilot.selfdrive.car.cruise.monotonic_ns', return_value=now):
card.update_cruise_speed(cs)
card.CS_prev = cs
assert card.v_cruise_helper.v_cruise_kph == 92.8
@pytest.mark.parametrize('bad', ['native_denied', 'expired_native', 'rx_fault', 'producer', 'epoch',
'source_zero', 'expired_wait', 'immediate', 'process', 'main_off', 'door'])
def test_cold_intent_retention_never_bypasses_provenance_or_faults(bad):
from dataclasses import replace
from openpilot.starpilot.aol.tests.test_gm_held_native import baseline, peers, Peers
from openpilot.starpilot.aol.transport_pause import TransportPauseFeedback
from openpilot.starpilot.aol.wire import encode_safety
from openpilot.selfdrive.selfdrived.events import Events
with OpenpilotPrefix(), ExitStack() as cleanup:
card, _, prior = baseline()
cleanup.callback(card.vehicle_startup.close)
feedback = TransportPauseFeedback(card.CP)
def sample(now, *, recovery=False, mutate=False):
cs = parse(card, now, counter=1 if recovery else 0)
native = replace(prior, observedMonoTime=now, validUntilMonoTime=now + 200_000_000,
requestedLateral=False, lateralAllowed=False, requestedLongitudinal=False, longitudinalAllowed=False,
pandaInventory=(replace(prior.pandaInventory[0], controlsAllowed=False),))
incoming, _ = peers(card.CP, card.slc_producer_session, now, controls=False, native=native,
sequence=2 if recovery else 1, long_active=False)
axis = incoming.messages['aolAxisState'].as_builder()
a = axis.aolAxisState
a.lateralActive = a.longitudinalActive = a.desiredLongitudinal = False
a.desiredLateral = recovery
a.nativeAcknowledged = recovery
a.faultReason = 'none' if recovery else 'availabilityPause'
a.faultEventMonoTime = now
events = Events()
if not recovery:
events.add(EventName.commIssue)
if mutate:
if bad == 'native_denied':
native = replace(native, requestedLateral=True)
elif bad == 'expired_native':
native = replace(native, observedMonoTime=now - 200_000_001, validUntilMonoTime=now - 1)
elif bad == 'rx_fault':
native = replace(native, pandaInventory=(replace(native.pandaInventory[0], safetyRxChecksInvalid=True),))
elif bad == 'producer':
a.intentProducerSessionId = 'foreign'
elif bad == 'epoch':
a.intentIntegrityEpoch += 1
elif bad == 'source_zero':
a.sourceCarStateMonoTime = 0
elif bad == 'immediate':
events.add(EventName.controlsMismatch)
elif bad == 'process':
events.add(EventName.processNotRunning)
elif bad == 'main_off':
cs.cruiseState.available = False
elif bad == 'door':
cs.doorOpen = True
event = messaging.new_message('onroadEvents', len(events), valid=True, logMonoTime=now)
event.onroadEvents = events.to_msg()
wire = messaging.new_message('aolSafetyWire', 0, valid=True, logMonoTime=native.observedMonoTime)
wire.aolSafetyWire = encode_safety(native)
sm = Peers([axis.as_reader(), event.as_reader(), wire.as_reader(), incoming.messages['pandaStates']])
return feedback.observe(sm, card.CP, cs, now, latched=True, intent_producer_session=card.slc_producer_session)
assert not sample(NOW)
assert feedback.cold_main_wait_ns == NOW
assert sample(NOW + (210_000_000 if bad == 'expired_wait' else 10_000_000), recovery=True, mutate=True)
assert feedback.failed
@@ -9,25 +9,26 @@ from opendbc.can import CANPacker
from opendbc.car import Bus
from opendbc.car.gm.tests.test_volt_sdgm_control import sdgm_params
from opendbc.car.gm.bolt_cc import button_bytes
from opendbc.car.gm.values import DBC, CruiseButtons
from opendbc.car.gm.values import DBC, CruiseButtons, CAR, is_bolt_pedal_profile
from opendbc.car.gm.tests.test_bolt_pedal import params as bolt_params
from openpilot.cereal import messaging
from openpilot.common.params import Params
from openpilot.common.prefix import OpenpilotPrefix
from openpilot.starpilot.aol.tests import test_ioniq_availability as fixture
from openpilot.starpilot.aol.tests import test_gm as gm_fixture
from openpilot.starpilot.aol.tests.test_gm_startup_main import parsed_packets
from openpilot.starpilot.aol.tests.test_seatbelt_cold_start import deliver_card
from openpilot.starpilot.aol.tests.test_gm_route18e_recovery import deliver_card
from openpilot.selfdrive.selfdrived.events import Events, EventName
from openpilot.starpilot.aol.wire import decode_safety, encode_safety
BASE = 10_000_000_000
def callers(cleanup):
def callers(cleanup, cp=None):
params = Params()
cleanup.callback(params._finalizer)
params.put_bool('AlwaysOnLateral', True, block=True)
card = gm_fixture.TestGmAol.card(sdgm_params(), params)
card = gm_fixture.TestGmAol.card(sdgm_params() if cp is None else cp, params)
cleanup.callback(card.vehicle_startup.close)
case = fixture.IoniqAvailabilityTests(methodName='runTest')
cleanup.callback(case.doCleanups)
@@ -37,6 +38,17 @@ def callers(cleanup):
def state(tick, pedal='none', main=True):
now = BASE + tick * 10_000_000
if is_bolt_pedal_profile(card.CP):
from openpilot.starpilot.aol.tests.test_gm_held_native import parse
packets = parse(card, now, gesture={'regen': 'paddle'}.get(pedal, pedal), counter=tick % 4, packets_only=True)
packets = [packet for packet in packets if packet[0] != 0xC9]
packets.append(packer.make_can_msg('ASCMActiveCruiseControlStatus', 2, {'ACCCruiseState': 0}))
packets.append(packer.make_can_msg('ECMEngineStatus', 0, {'CruiseMainOn': int(main), 'BrakePressed': int(pedal == 'brake')}))
if tick == 0:
card.CI.update([(now - 1_000_000, packets)])
cs = card.CI.update([(now, packets)])
assert cs.canValid and not cs.canTimeout
return cs
packets = parsed_packets(card.CP, packer, now, tick, 20., main)
addresses = {packer.make_can_msg(name, 0, {})[0] for name in ('ECMAcceleratorPos', 'EBCMRegenPaddle')}
packets = [p for p in packets if p[0] not in addresses]
@@ -68,11 +80,15 @@ def callers(cleanup):
return card, sd, controls, state, drive
@pytest.mark.parametrize('profile', [None, *[(candidate, removed) for candidate in
(CAR.CHEVROLET_BOLT_CC_2017, CAR.CHEVROLET_BOLT_CC_2018_2021, CAR.CHEVROLET_BOLT_CC_2022_2023,
CAR.CHEVROLET_BOLT_ACC_2022_2023_PEDAL) for removed in (False, True)]])
@pytest.mark.parametrize('pedal', ['brake', 'regen'])
@pytest.mark.parametrize('order', ['together', 'axis_first', 'event_first'])
def test_parsed_pedal_lag_keeps_intent_without_renewing_authority(pedal, order):
def test_parsed_pedal_lag_keeps_intent_without_renewing_authority(pedal, order, profile):
with OpenpilotPrefix(), ExitStack() as cleanup:
card, sd, controls, parse, drive = callers(cleanup)
cp = None if profile is None else bolt_params(profile[0], pedal=True, camera=not profile[1], removed=profile[1])
card, sd, controls, parse, drive = callers(cleanup, cp)
feedback = card.aol_transport_feedback
baseline = (feedback.lateral_baseline_ns, feedback.lateral_source_ns, feedback.lateral_native_ns,
feedback.healthy_feedback_ns, feedback.healthy_native_ns)
@@ -99,9 +115,11 @@ def test_parsed_pedal_lag_keeps_intent_without_renewing_authority(pedal, order):
assert baseline == (feedback.lateral_baseline_ns, feedback.lateral_source_ns, feedback.lateral_native_ns,
feedback.healthy_feedback_ns, feedback.healthy_native_ns)
cs = parse(3)
with mock.patch.object(sd.rk.avg_dt, 'get_average', return_value=.02):
axis, _, command = drive(3, cs, allowed=False, native_request_lat=False)
axis, _, command = drive(3, cs, allowed=False, native_request_lat=False)
assert not axis.lateralActive and not command.latActive and not command.longActive
assert axis.desiredLateral and str(axis.faultReason) == 'none'
deliver_card(card, sd, controls, cs, 3)
assert card.aol_card_intent.allowed_latch and not feedback.failed
cs = parse(4)
axis, _, command = drive(4, cs)
assert axis.lateralActive and command.latActive and not command.longActive
+59 -11
View File
@@ -257,6 +257,9 @@ class TransportPauseFeedback:
self.awaiting = False
self.failed = False
self.lateral_baseline_ns = 0
self.ever_lateral_active = False
self.lag_withdrawal_ns = 0
self.cold_main_wait_ns = 0
self.lateral_only_baseline = False
self.lateral_source_ns = 0
self.lateral_inventory = None
@@ -284,14 +287,16 @@ class TransportPauseFeedback:
method neither grants outputs nor exempts any native or control fault.
"""
from openpilot.starpilot.nostalgia import physical_cancel
from openpilot.starpilot.car.gm.aol import native_accepts_cp
gm = native_accepts_cp(cp, int(cp.safetyConfigs[0].safetyModel.raw), int(cp.safetyConfigs[0].safetyParam))
if (not latched or self.failed or self.paused or self.pending_event or self.retention_started_ns or
not self.session or str(axis.sessionId) != self.session or
not producer or producer != self.availability_producer or
str(getattr(axis, 'intentProducerSessionId', '')) != producer or
int(getattr(axis, 'intentIntegrityEpoch', 0)) != self.availability_epoch or
not self.lateral_inventory or not self.lateral_baseline_ns or
str(axis.faultReason) != 'none' or str(axis.mode) != 'lateralOnly' or not axis.nativeAcknowledged or
not axis.desiredLateral or not axis.lateralActive or axis.longitudinalActive or
str(axis.faultReason) != 'none' or not gm and str(axis.mode) != 'lateralOnly' or not axis.nativeAcknowledged or
not axis.desiredLateral or not axis.lateralActive or axis.longitudinalActive and not gm or
not 0 < self.lateral_source_ns <= axis.sourceCarStateMonoTime <= axis.observedMonoTime <= now_ns or
now_ns - axis.sourceCarStateMonoTime > AXIS_MAX_AGE_NS or
not cs.canValid or cs.canTimeout or cs.steerFaultTemporary or cs.steerFaultPermanent or cs.accFaulted or
@@ -308,8 +313,8 @@ class TransportPauseFeedback:
return None
owners = [index for index, config in enumerate(cp.safetyConfigs)
if str(config.safetyModel) not in ('silent', 'noOutput')]
return native if (len(owners) == 1 and not native.pandaInventory[owners[0]].controlsAllowed and
not sm['pandaStates'][owners[0]].controlsAllowed) else None
return native if (len(owners) == 1 and (gm or not native.pandaInventory[owners[0]].controlsAllowed and
not sm['pandaStates'][owners[0]].controlsAllowed)) else None
def _temporary_host_intent_current(self, sm, cp, cs, axis, stamp, now_ns, *, latched, producer):
"""A fresh inactive proof may retain intent, never native/output authority."""
@@ -388,7 +393,11 @@ class TransportPauseFeedback:
faults[0].softDisable and not faults[0].immediateDisable)
# Healthy cycles never reach policy/native/provenance work. A critical
# proof can precede its event, but cannot yet establish a lag exemption.
if not isolated and (reason != 'critical' or faults):
recovering = bool(cp.brand == 'gm' and not faults and reason == 'none' and
0 < self.lag_withdrawal_ns <= now_ns <= self.lag_withdrawal_ns + SAFETY_MAX_AGE_NS and
axis.desiredLateral and not axis.lateralActive and axis.nativeAcknowledged and
not axis.desiredLongitudinal and not axis.longitudinalActive)
if not recovering and not isolated and (reason != 'critical' or faults):
return False
previous = self.signature
joining = bool(self.pending_event and previous is not None and previous[5] == 'critical' and
@@ -420,7 +429,7 @@ class TransportPauseFeedback:
now_ns - axis.faultEventMonoTime <= AXIS_MAX_AGE_NS and
(joining or known_lag or not faults and previous is not None and previous[5] == 'none' and
0 <= now_ns - self.lateral_baseline_ns <= AXIS_MAX_AGE_NS))
if not (matched or held_original or pending):
if not (matched or held_original or pending or recovering):
return False
from openpilot.starpilot.aol.vehicle import policy_for
if not policy_for(cp).intent_supported:
@@ -433,13 +442,13 @@ class TransportPauseFeedback:
return False
users = [event for event in events if event.userDisable]
if users:
from opendbc.car.gm.values import is_volt_sdgm_accepted_envelope
from opendbc.car.gm.values import is_volt_sdgm_accepted_envelope, is_bolt_pedal_profile
from openpilot.starpilot.nostalgia import physical_cancel
# On the accepted Volt SDGM owner, an already LAT-only physical brake
# On accepted Volt SDGM and Bolt pedal owners, a LAT-only physical brake
# or regen sample also emits ordinary pedalPressed/USER_DISABLE. It
# must not turn an otherwise source-qualified temporary lag withdrawal
# into a permanent driver-intent reset. No axis or lease is retained.
volt_pedal = bool(self.lateral_only_baseline and is_volt_sdgm_accepted_envelope(cp) and
gm_pedal = bool(self.lateral_only_baseline and (is_volt_sdgm_accepted_envelope(cp) or is_bolt_pedal_profile(cp)) and
(cs.brakePressed or cs.regenBraking) and cs.cruiseState.available and
not cs.doorOpen and not cs.seatbeltUnlatched and not cs.steeringDisengage and not physical_cancel(cs, cp) and
not axis.desiredLongitudinal and not axis.longitudinalActive and
@@ -447,7 +456,7 @@ class TransportPauseFeedback:
all(event.name == log.OnroadEvent.EventName.pedalPressed and event.noEntry and
not event.softDisable and not event.immediateDisable for event in users))
stock_user = stock_lateral_user_event(cp)
if not volt_pedal and (not getattr(self, 'lateral_only_baseline', False) or stock_user is None or
if not gm_pedal and (not getattr(self, 'lateral_only_baseline', False) or stock_user is None or
any(event.name != stock_user or event.softDisable or event.immediateDisable for event in users) or
not stock_lateral_off(cp, cs, native)):
return False
@@ -457,6 +466,8 @@ class TransportPauseFeedback:
self.awaiting = True
elif joining:
self.pending_event = 0
if matched and cp.brand == 'gm':
self.lag_withdrawal_ns = event_ns
return True
def _temporary_steering_intent_current(self, sm, cp, cs, axis, stamp, now_ns, *, latched, producer, signature):
@@ -962,6 +973,10 @@ class TransportPauseFeedback:
cp_identity(cp) == self.cp)
continuity = bool(not self.session or axis.sessionId == self.session and
(axis.sequence > self.sequence or axis.sequence == self.sequence and signature == self.signature))
if (self.cold_main_wait_ns and current and continuity and self.signature is not None and
signature[14:16] != self.signature[14:16]):
self.failed = True
return True
if current and continuity and 0 < main_arm_ns <= now_ns <= main_arm_ns + AXIS_MAX_AGE_NS:
from openpilot.starpilot.car.gm.aol import prior_main_off_intent_current
if prior_main_off_intent_current(self, sm, cp, cs, axis, now_ns,
@@ -970,17 +985,47 @@ class TransportPauseFeedback:
# the new request nor renews any command, native or healthy lease.
self.sequence, self.signature = int(axis.sequence), signature
return False
if (cp.brand == 'gm' and current and continuity and latched and not self.failed and
not self.ever_lateral_active and not self.pending_event):
from openpilot.starpilot.car.gm.aol import cold_main_readiness
native = _healthy_transport_native(sm, cp, cs, str(axis.sessionId), now_ns)
event_ns = int(sm.logMonoTime['onroadEvents'])
recovering = bool(reason == 'none' and axis.desiredLateral and not axis.desiredLongitudinal and
not axis.lateralActive and not axis.longitudinalActive and axis.nativeAcknowledged and
0 < self.cold_main_wait_ns <= now_ns <= self.cold_main_wait_ns + SAFETY_MAX_AGE_NS and
0 < axis.sourceCarStateMonoTime <= stamp and now_ns - axis.sourceCarStateMonoTime <= AXIS_MAX_AGE_NS)
if ((reason == 'availabilityPause' and not any(signature[9:13]) or recovering) and
intent_producer_session and signature[14] == intent_producer_session and
(self.signature is None or signature[14:16] == self.signature[14:16]) and
axis.faultEventMonoTime == event_ns and sm.seen['onroadEvents'] and
sm.valid['onroadEvents'] and sm.alive['onroadEvents'] and 0 < event_ns <= stamp and
now_ns - event_ns <= AXIS_MAX_AGE_NS and
cold_main_readiness(cp, cs, native, sm['onroadEvents'], recovering=recovering)):
if not recovering:
self.cold_main_wait_ns = event_ns
self.session = str(axis.sessionId)
self.sequence, self.signature = int(axis.sequence), signature
self.temporary_host_qualified = True
return False
if (self.cold_main_wait_ns and current and continuity and reason == 'none' and
axis.desiredLateral and not axis.lateralActive):
self.failed = True
return True
if current and continuity and self._angle_stop_intent_current(sm, cp, cs, axis, stamp, now_ns,
latched=latched, producer=intent_producer_session):
self.sequence, self.signature = int(axis.sequence), signature
return False
if current and continuity and (reason == 'critical' or
if current and continuity and (reason == 'critical' or self.lag_withdrawal_ns and not axis.lateralActive or
getattr(sm, 'updated', {}).get('onroadEvents', False)) and self._temporary_lag_intent_current(sm, cp, cs, axis, stamp, now_ns,
latched=latched, producer=intent_producer_session, signature=signature):
# Received identity only; every original permission/output deadline and
# healthy baseline timestamp remains unchanged.
self.sequence, self.signature = int(axis.sequence), signature
return False
if (cp.brand == 'gm' and self.lag_withdrawal_ns and current and continuity and reason == 'none' and
axis.desiredLateral and not axis.lateralActive):
self.failed = True
return True
if current and continuity and self._temporary_host_intent_current(sm, cp, cs, axis, stamp, now_ns,
latched=latched, producer=intent_producer_session):
self.sequence, self.signature = int(axis.sequence), signature
@@ -1154,6 +1199,9 @@ class TransportPauseFeedback:
self.lateral_only_baseline = bool(lateral_baseline and self.availability_producer and
not axis.desiredLongitudinal and not axis.longitudinalActive and not native.requestedLongitudinal)
if lateral_baseline and native is not None:
self.ever_lateral_active = True
self.lag_withdrawal_ns = 0
self.cold_main_wait_ns = 0
self.temporary_eps_event_ns = 0
self.lateral_native_ns = int(native.observedMonoTime)
self.lateral_inventory = (native.pandaSerial, tuple(p.hardwareSerial for p in native.pandaInventory))
+18
View File
@@ -122,6 +122,24 @@ def create_intent(cp, settings):
return GmAolCardIntent(settings, cp) if qualified_gm(cp) else None
def cold_main_readiness(cp, cs, native, events, *, recovering=False):
"""A MAIN request may wait for host startup without acquiring either axis."""
from openpilot.cereal import log
names = (log.OnroadEvent.EventName.commIssue, log.OnroadEvent.EventName.commIssueAvgFreq,
log.OnroadEvent.EventName.locationdTemporaryError)
faults = [event for event in events if event.softDisable or event.immediateDisable]
host_wait = not faults if recovering else bool(faults and
all(event.softDisable and not event.immediateDisable and event.name in names for event in faults))
return bool(qualified_gm(cp) and aol_enabled(cp) and host_wait and
not any(event.userDisable for event in events) and
cs.canValid and not cs.canTimeout and cs.cruiseState.available and
not cs.steerFaultTemporary and not cs.steerFaultPermanent and not cs.accFaulted and
not cs.doorOpen and not cs.seatbeltUnlatched and not cs.steeringDisengage and
native is not None and native.pandaInventory and
not native.requestedLateral and not native.lateralAllowed and
not native.requestedLongitudinal and not native.longitudinalAllowed)
def prior_main_off_intent_current(feedback, sm, cp, cs, axis, now_ns, *, latched, main_arm_ns, producer):
"""An older idle response cannot reject a newly armed cruise MAIN request."""
from opendbc.car.structs import car
Binary file not shown.
BIN
View File
Binary file not shown.
BIN
View File
Binary file not shown.
BIN
View File
Binary file not shown.
BIN
View File
Binary file not shown.
Binary file not shown.
BIN
View File
Binary file not shown.
Binary file not shown.
BIN
View File
Binary file not shown.
Binary file not shown.
BIN
View File
Binary file not shown.
Binary file not shown.
BIN
View File
Binary file not shown.
Binary file not shown.
BIN
View File
Binary file not shown.
Binary file not shown.
BIN
View File
Binary file not shown.
Binary file not shown.
+1 -1
View File
@@ -1,4 +1,4 @@
#pragma once
extern const uint8_t gitversion[29];
const uint8_t gitversion[29] = "DEV-9deea7d8-Pa5919516-DEBUG";
const uint8_t gitversion[29] = "DEV-b432b14a-Pa5919516-DEBUG";
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
+1 -1
View File
@@ -1 +1 @@
DEV-9deea7d8-Pa5919516-DEBUG
DEV-b432b14a-Pa5919516-DEBUG
+1 -1
View File
@@ -1 +1 @@
{"protected_inventory_sha256": "52a5eb0ff40bbc58a9a322841444af0c28122286682a8077542d1db218fbbe9b", "schema_version": 1}
{"protected_inventory_sha256": "d79e9d296f38e311b713d5aba3e04e3b023e59ab249a2d897de3b2b417f90246", "schema_version": 1}