From 9bbf79766a1d679da70b2832ecf9fa85f064e295 Mon Sep 17 00:00:00 2001 From: firestar5683 <168790843+firestar5683@users.noreply.github.com> Date: Sun, 8 Nov 2026 00:08:13 -0600 Subject: [PATCH] Preserve Tesla screen intent and Palisade Cancel behavior --- opendbc_repo/opendbc/car/hyundai/carstate.py | 3 +- opendbc_repo/opendbc/car/tesla/carstate.py | 3 +- .../car/tesla/tests/test_screen_button.py | 33 +++ .../safety/modes/hyundai_blended_stock_aol.h | 25 +- .../opendbc/safety/modes/tesla_screen_aol.h | 9 +- .../tests/test_hyundai_blended_alpha.py | 93 +++++++- .../safety/tests/test_tesla_screen_button.py | 61 ++++- openpilot/selfdrive/car/card.py | 6 +- openpilot/starpilot/aol/intent.py | 7 +- .../aol/tests/test_palisade_cancel.py | 214 ++++++++++++++++++ openpilot/starpilot/aol/transport_pause.py | 3 +- openpilot/starpilot/car/hyundai/aol.py | 8 +- .../car/hyundai/classic_long_intent.py | 4 +- .../starpilot/car/tesla/screen_intent.py | 4 +- .../car/tesla/tests/test_screen_intent.py | 49 ++++ .../tests/test_screen_pedal_continuity.py | 100 ++++++++ 16 files changed, 599 insertions(+), 23 deletions(-) create mode 100644 openpilot/starpilot/aol/tests/test_palisade_cancel.py create mode 100644 openpilot/starpilot/car/tesla/tests/test_screen_pedal_continuity.py diff --git a/opendbc_repo/opendbc/car/hyundai/carstate.py b/opendbc_repo/opendbc/car/hyundai/carstate.py index 0078d9eb88..afcef43400 100644 --- a/opendbc_repo/opendbc/car/hyundai/carstate.py +++ b/opendbc_repo/opendbc/car/hyundai/carstate.py @@ -348,9 +348,10 @@ class CarState(CarStateBase): self.lkas11 = copy.copy(cp_cam.vl["LKAS11"]) self.clu11 = copy.copy(cp.vl["CLU11"]) from opendbc.car.hyundai.classic_long_aol import independent_buttons_qualified + from opendbc.car.hyundai.blended_stock_aol import qualified_alpha self.ev6_aol_samples = () self.ev6_aol_source_healthy = False - if independent_buttons_qualified(self.CP): + if independent_buttons_qualified(self.CP) or qualified_alpha(self.CP, marked_only=True): from opendbc.can.parser import MAX_BAD_COUNTER source = cp.message_states.get(cp.dbc.name_to_msg['CLU11'].address) self.ev6_aol_sample_stamp_ns = int(source.timestamps[-1]) if source is not None and source.timestamps else 0 diff --git a/opendbc_repo/opendbc/car/tesla/carstate.py b/opendbc_repo/opendbc/car/tesla/carstate.py index 3a63988818..d6ba6fd9d5 100644 --- a/opendbc_repo/opendbc/car/tesla/carstate.py +++ b/opendbc_repo/opendbc/car/tesla/carstate.py @@ -9,7 +9,8 @@ from opendbc.car.tesla.values import CAR, DBC, CANBUS, GEAR_MAP, STEER_THRESHOLD class TeslaScreenCANParser(CANParser): def __init__(self): - super().__init__('tesla_model3_vehicle', [('UI_status2', 0)], CANBUS.vehicle) + # Screen gestures are optional; their silence is not a control-bus fault. + super().__init__('tesla_model3_vehicle', [('UI_status2', float('nan'))], CANBUS.vehicle) def update(self, strings, sendcan=False): if strings and not isinstance(strings[0], list | tuple): diff --git a/opendbc_repo/opendbc/car/tesla/tests/test_screen_button.py b/opendbc_repo/opendbc/car/tesla/tests/test_screen_button.py index 465078e2a0..c02d2bdda9 100644 --- a/opendbc_repo/opendbc/car/tesla/tests/test_screen_button.py +++ b/opendbc_repo/opendbc/car/tesla/tests/test_screen_button.py @@ -80,3 +80,36 @@ def test_literal_optional_screen_frames_baseline_edge_and_exact_dlc(): (structs.CarState.ButtonEvent.Type.lkas, True)] parser.update([(1_030_000_000, [(frame[0], frame[1][:-1], 1), (frame[0], frame[1], 0)])]) assert cs.update_screen_button(parser) == [] + + +@pytest.mark.parametrize('screen_seen', [False, True]) +def test_optional_screen_silence_does_not_report_vehicle_bus_loss(screen_seen): + parser = TeslaScreenCANParser() + if screen_seen: + parser.update([(1_000_000_000, [(0x3DF, bytes(8), CANBUS.vehicle)])]) + for stamp in (1_500_000_001, 12_000_000_000): + parser.update([(stamp, [(0x111, bytes(8), CANBUS.vehicle)])]) + assert not parser.bus_timeout + assert parser.can_valid + assert parser.vl_all['UI_status2']['UI_activeTouchPoints'] == [] + + +@pytest.mark.parametrize('missing_bus', [None, CANBUS.party, CANBUS.autopilot_party]) +def test_screen_silence_keeps_required_vehicle_bus_timeouts(missing_bus): + cp = screen_params() + apply_screen_button(cp, True, False) + ci = CarInterface(cp) + packer = CANPacker('tesla_model3_party') + messages = { + CANBUS.party: ['DI_speed', 'DI_systemStatus', 'ESP_B', 'EPAS3S_sysStatus', 'ESP_status', 'UI_warning', 'DI_state'], + CANBUS.autopilot_party: ['SCCM_steeringAngleSensor', 'DAS_status', 'DAS_control', 'DAS_steeringControl', 'DAS_settings'], + } + for index in range(150): + frames = [packer.make_can_msg(name, bus, {}) for bus, names in messages.items() + if index < 50 or bus != missing_bus for name in names] + if index == 0: + frames.append((0x3DF, bytes(8), CANBUS.vehicle)) + result = ci.update([(1_000_000_000 + index * 10_000_000, frames)]) + assert result.canTimeout is (missing_bus is not None) + if missing_bus is None: + assert result.canValid diff --git a/opendbc_repo/opendbc/safety/modes/hyundai_blended_stock_aol.h b/opendbc_repo/opendbc/safety/modes/hyundai_blended_stock_aol.h index df9819946d..44a640bdd2 100644 --- a/opendbc_repo/opendbc/safety/modes/hyundai_blended_stock_aol.h +++ b/opendbc_repo/opendbc/safety/modes/hyundai_blended_stock_aol.h @@ -158,10 +158,29 @@ static void blended_aol_rx(const CANPacket_t *msg) { bool pressed = false; if ((msg->bus == pt) && (msg->addr == 0x4F1U) && (GET_LEN(msg) == 4U)) { if ((msg->data[0] & 7U) == 4U) { - blended_aol_clear(); + // Mixed LONG Pause/Resume still clears ordinary/LONG authority in the + // cruise hook. Retain only an already claimed, current lateral token. + // CANCEL cannot arm, renew a request/gesture lease, or clear an inhibit. + const bool source_current = (blended_aol_token_source < 3U) && + (safety_get_ts_elapsed(now, blended_aol_source_ts[blended_aol_token_source]) <= 300000U); + const bool retain_lateral = blended_aol_long && !GET_BIT(msg, 3U) && + blended_aol_token && blended_aol_claimed && blended_aol_session && heartbeat_engaged && + blended_aol_main_seen && blended_aol_main && + (safety_get_ts_elapsed(now, blended_aol_main_ts) <= 100000U) && source_current && + blended_aol_request_seen && ((aol_host_axis_mask & 1U) != 0U) && + (safety_get_ts_elapsed(now, aol_host_request_ts) <= AOL_HOST_REQUEST_TIMEOUT_US) && + aol_rx_healthy() && !relay_malfunction && !safety_rx_checks_invalid; + if (!retain_lateral) { + blended_aol_clear(); + } + // A simultaneous MAIN is not an arming gesture. The advancing physical + // source remains observable without changing its token or host lease. + blended_aol_pressed[0] = GET_BIT(msg, 3U); + blended_aol_source_ts[0] = now; + } else { + source = 0U; + pressed = GET_BIT(msg, 3U); } - source = 0U; - pressed = GET_BIT(msg, 3U); } else if ((msg->bus == pt) && (msg->addr == 0x391U) && (GET_LEN(msg) == 8U)) { source = 1U; pressed = GET_BIT(msg, 4U); diff --git a/opendbc_repo/opendbc/safety/modes/tesla_screen_aol.h b/opendbc_repo/opendbc/safety/modes/tesla_screen_aol.h index 593c3d4c03..8a40a4d5e2 100644 --- a/opendbc_repo/opendbc/safety/modes/tesla_screen_aol.h +++ b/opendbc_repo/opendbc/safety/modes/tesla_screen_aol.h @@ -13,6 +13,7 @@ static bool tesla_screen_belt_latched = false; static bool tesla_screen_eps_ready = false; static bool tesla_screen_session = false; static bool tesla_screen_autopark = false; +static bool tesla_screen_acc_fault = false; static int tesla_screen_override = 0; static void tesla_screen_clear(void) { @@ -32,6 +33,7 @@ static void tesla_screen_reset(void) { tesla_screen_eps_ready = false; tesla_screen_session = false; tesla_screen_autopark = false; + tesla_screen_acc_fault = false; } static uint8_t tesla_screen_request(void) { @@ -54,7 +56,7 @@ static uint8_t tesla_screen_request(void) { static bool tesla_screen_physical_ready(void) { return tesla_screen_drive && tesla_screen_occupant_ready && tesla_screen_eps_ready && - !steering_disengage && !tesla_screen_autopark; + !steering_disengage && !tesla_screen_autopark && !tesla_screen_acc_fault; } static uint8_t tesla_screen_permission(void) { @@ -108,14 +110,17 @@ static void tesla_screen_rx(const CANPacket_t *msg, bool autopark) { } if (msg_matches(msg, 0x286U, 0U, 8U)) { const unsigned int state = (msg->data[1] >> 4U) & 7U; + tesla_screen_acc_fault = state == 5U; const bool cruise = ((state == 2U) || (state == 3U) || (state == 4U) || (state == 6U) || (state == 7U)) && !tesla_screen_autopark; const bool main_available = ((state == 1U) || cruise) && !tesla_screen_autopark; if (cruise && !tesla_screen_cruise) { tesla_screen_override = 0; } - if (tesla_screen_main && !main_available && !brake_pressed) { + // Stock availability cannot revoke an explicit screen selection. + if (tesla_screen_main && !main_available && !brake_pressed && (tesla_screen_override != 1)) { tesla_screen_override = -1; } + if (tesla_screen_acc_fault) { tesla_screen_clear(); } tesla_screen_main = main_available; tesla_screen_cruise = cruise; } diff --git a/opendbc_repo/opendbc/safety/tests/test_hyundai_blended_alpha.py b/opendbc_repo/opendbc/safety/tests/test_hyundai_blended_alpha.py index 9f2b767caa..9ccd0f98c8 100644 --- a/opendbc_repo/opendbc/safety/tests/test_hyundai_blended_alpha.py +++ b/opendbc_repo/opendbc/safety/tests/test_hyundai_blended_alpha.py @@ -313,8 +313,8 @@ class TestHyundaiBlendedAlphaAol(unittest.TestCase): self.assertEqual(self.request(1), 0) self.assertFalse(self.safety.get_controls_allowed()) - def test_main_off_cancel_and_heartbeat_loss_revoke_physical_token(self): - for cause in ('main_off', 'cancel', 'heartbeat'): + def test_main_off_and_heartbeat_loss_revoke_physical_token(self): + for cause in ('main_off', 'heartbeat'): self.now = 1_000_000 self.counter = 0 self.reset() @@ -325,13 +325,98 @@ class TestHyundaiBlendedAlphaAol(unittest.TestCase): continue if cause == 'main_off': self.feed(available=False) - elif cause == 'cancel': - self.feed(button=4) else: self.safety.set_aol_test_heartbeat(False) self.assertEqual(self.request(1), 0) self.assertFalse(self.safety.get_controls_allowed()) + def test_cancel_retains_only_existing_claimed_lateral_and_stops_long(self): + for word in (0x2004, 0x2014): + for held in (7, 23): # Both reported short and longer Pause/Resume holds. + self.counter, self.now = 0, 1_000_000 + self.reset(word) + self.warm() + self.feed(lda=True) + self.assertEqual(self.request(1), 0 if self.release else 1) + if self.release: + continue + self.feed(button=2) + self.feed() + self.assertTrue(self.safety.get_controls_allowed()) + self.assertEqual(self.request(3), 3) + for _ in range(held): + self.feed(button=4) + self.assertFalse(self.safety.get_controls_allowed()) + self.assertEqual(self.request(1), 1) + self.feed() + self.assertFalse(self.safety.get_controls_allowed()) + self.assertEqual(self.request(1), 1) + if self.bus: + self.primary_counter = 0 + steering = TestHyundaiBlendedAlpha.primary(self, 1, True) + else: + from opendbc.car.hyundai.hyundaican import hyundai_checksum + + def steering_checksum(message): + address, data, bus = message + return address, bytes([hyundai_checksum(data[1:8])]) + data[1:8], bus + steering = self.packer.make_can_msg_safety('LKAS11', 0, + {'CF_Lkas_MsgCount': 0, 'CR_Lkas_StrToqReq': 1, 'CF_Lkas_ActToi': 1}, fix_checksum=steering_checksum) + self.assertTrue(self.safety.safety_tx_hook(steering), 'Only retained LAT can authorize steering') + accel = self.packer.make_can_msg_safety('SCC11', self.bus, {'aReqRaw': .1}) + self.assertFalse(self.safety.safety_tx_hook(accel)) + self.assertEqual(self.request(0), 0) # Configured cancellation remains authoritative. + + def test_cancel_cannot_arm_or_restore_expired_or_withdrawn_authority(self): + for word in (0x2004, 0x2014): + for cause in ('cold', 'unclaimed', 'withdrawn', 'stale_request', 'heartbeat', 'main_off', 'relay'): + with self.subTest(word=word, cause=cause): + self.counter, self.now = 0, 1_000_000 + self.reset(word) + self.warm() + if cause != 'cold': + self.feed(lda=True) + if cause not in ('cold', 'unclaimed'): + self.assertEqual(self.request(1), 0 if self.release else 1) + if self.release: + continue + if cause == 'withdrawn': + self.request(0) + elif cause == 'stale_request': + for _ in range(32): + self.feed() # Fresh main and token source, without a host renewal. + elif cause == 'heartbeat': + self.safety.set_aol_test_heartbeat(False) + elif cause == 'relay': + self.safety.set_relay_malfunction(True) + self.feed(button=4, available=cause != 'main_off') + self.assertEqual(self.request(1), 0, cause) + + def test_cancel_cannot_restore_expired_source_simultaneous_main_or_bad_rx(self): + for word in (0x2004, 0x2014): + for cause in ('source', 'simultaneous_main', 'bad_rx'): + with self.subTest(word=word, cause=cause): + self.counter, self.now = 0, 1_000_000 + self.reset(word) + self.warm() + self.feed(lda=True) + self.assertEqual(self.request(1), 0 if self.release else 1) + if self.release: + continue + if cause == 'source': + for _ in range(32): + self.feed(omit='BCM_PO_11') + self.safety.aol_set_host_request(1) # Fresh request, expired physical LDA source. + elif cause == 'bad_rx': + invalid = self.packer.make_can_msg_safety('TCS13', self.bus, {'AliveCounterTCS': self.counter % 8}, fix_checksum=checksum) + invalid[0].data[6] ^= 1 + self.assertFalse(self.safety.safety_rx_hook(invalid)) + packet = self.packer.make_can_msg_safety('CLU11', self.bus, + {'CF_Clu_CruiseSwState': 4, 'CF_Clu_CruiseSwMain': int(cause == 'simultaneous_main'), + 'CF_Clu_AliveCnt1': self.counter % 16}) + self.assertTrue(self.safety.safety_rx_hook(packet)) + self.assertEqual(self.request(1), 0, cause) + def test_cluster_lda_source_needs_its_own_neutral_edge(self): self.reset() self.warm() diff --git a/opendbc_repo/opendbc/safety/tests/test_tesla_screen_button.py b/opendbc_repo/opendbc/safety/tests/test_tesla_screen_button.py index d0a56c931b..84fa1e01fb 100644 --- a/opendbc_repo/opendbc/safety/tests/test_tesla_screen_button.py +++ b/opendbc_repo/opendbc/safety/tests/test_tesla_screen_button.py @@ -33,7 +33,8 @@ class TestTeslaScreenButton(unittest.TestCase): ('DAS_steeringControl', 2, {'DAS_steeringControlType': stock_type}), ('DI_speed', 0, {'DI_vehicleSpeed': 0}), ('ESP_B', 0, {'ESP_vehicleSpeed': 0, 'ESP_wheelSpeedsQF': 1, 'ESP_vehicleStandstillSts': 1}), - ('EPAS3S_sysStatus', 0, {'EPAS3S_internalSAS': 0, 'EPAS3S_eacStatus': eps, 'EPAS3S_handsOnLevel': hands}), + ('EPAS3S_sysStatus', 0, {'EPAS3S_internalSAS': 0, 'EPAS3S_eacStatus': eps, 'EPAS3S_handsOnLevel': hands, + 'EPAS3S_torsionBarTorque': 0}), ('DI_systemStatus', 0, {'DI_gear': gear, 'DI_accelPedalPos': 0}), ('ESP_status', 0, {'ESP_driverBrakeApply': 2 if brake else 1}), ('UI_warning', 0, {'anyDoorOpen': int(door), 'buckleStatus': int(belt)}), @@ -106,6 +107,64 @@ class TestTeslaScreenButton(unittest.TestCase): self.assertEqual(self.permission(), 0) self.arm() + def test_availability_loss_preserves_only_explicit_screen_arm(self): + tx_packer = CANPackerSafety('tesla_model3_party') + for word in (512, 768, 1536, 1792): + with self.subTest(word=word): + self.start(word) + self.arm() + self.touch(0) + for state in (1, 0, 1, 0, 1): + self.feed(state=state) + self.assertEqual(self.permission(3), 1) + positive = tx_packer.make_can_msg_safety('DAS_control', 0, {'DAS_accelMin': 1, 'DAS_accelMax': 1}) + self.assertFalse(self.lib.safety_tx_hook(positive)) + self.touch(3) + self.assertEqual(self.permission(), 0) + for state in (0, 1, 0, 1): + self.feed(state=state) + self.assertEqual(self.permission(), 0) + + self.start(word) + self.feed(state=1) + self.assertEqual(self.permission(), 1) + self.feed(state=0) + self.assertEqual(self.permission(), 0) + self.feed(state=1) + self.assertEqual(self.permission(), 0) + + def test_cruise_fault_clears_screen_arm_and_blocks_gestures(self): + for word in (512, 768, 1536, 1792): + with self.subTest(word=word): + self.start(word) + self.arm() + self.feed(state=1) + self.feed(state=5) + self.assertEqual(self.permission(), 0) + self.touch(0) + self.touch(3) + self.assertEqual(self.permission(), 0) + for state in (0, 1): + self.feed(state=state) + self.assertEqual(self.permission(), 0) + self.arm() + + def test_availability_loss_cannot_hide_physical_fault_or_brake(self): + for word in (512, 768, 1536, 1792): + faults = [{'door': True}, {'eps': 3}, {'hands': 3}, {'gear': 0}, {'autopark': 3}] + if word & 1024: + faults.append({'brake': True}) + for fault in faults: + with self.subTest(word=word, fault=fault): + self.start(word) + self.arm() + self.feed(state=1) + self.feed(state=0, **fault) + self.assertEqual(self.permission(), 0) + for state in (0, 1, 0, 1): + self.feed(state=state) + self.assertEqual(self.permission(), 0) + def test_brake_disengagement_cannot_be_undone_by_availability(self): self.start(1536) self.arm() diff --git a/openpilot/selfdrive/car/card.py b/openpilot/selfdrive/car/card.py index 10d7ccccfe..f85578e8b1 100644 --- a/openpilot/selfdrive/car/card.py +++ b/openpilot/selfdrive/car/card.py @@ -60,6 +60,7 @@ from opendbc.car.ford.aol import ( temporary_restriction as ford_temporary_restriction, ) from opendbc.car.hyundai.canfd_angle_aol import native_observation, temporary_restriction, qualified as qualified_angle_aol +from opendbc.car.hyundai.blended_stock_aol import qualified_alpha as qualified_blended_alpha_aol from opendbc.car.hyundai.canfd_owner import config_index as hyundai_canfd_config_index from opendbc.car.hyundai.captured_adrv_startup import CapturedADRVStartup from openpilot.starpilot.car.nissan.aol import ( @@ -615,9 +616,10 @@ class Car: self.aol_card_intent, cancel_source_ns=cancel_source_ns, now_ns=now_ns, producer=getattr(self, 'slc_producer_session', '')) cancel_feedback = self.aol_transport_feedback - if (hkg_cancel_retained and self.CP.carFingerprint in (HYUNDAI_CAR.KIA_EV9, HYUNDAI_CAR.HYUNDAI_IONIQ_5_PE) and + if (hkg_cancel_retained and (self.CP.carFingerprint in (HYUNDAI_CAR.KIA_EV9, HYUNDAI_CAR.HYUNDAI_IONIQ_5_PE) or + qualified_blended_alpha_aol(self.CP, marked_only=True)) and cancel_feedback is not None and not cancel_feedback.longitudinal_cancel_ns): - # The pure guard proved this exact CCNC raw CANCEL/FULL join. Bind + # The pure guard proved this exact raw CANCEL/FULL join. Bind # its first local receipt for the existing bounded withdrawal handshake; # held samples never renew it or any source/native/output deadline. if cancel_source_ns > cancel_feedback.last_cancel_source_ns: diff --git a/openpilot/starpilot/aol/intent.py b/openpilot/starpilot/aol/intent.py index a6e7d33196..4717bb9a03 100644 --- a/openpilot/starpilot/aol/intent.py +++ b/openpilot/starpilot/aol/intent.py @@ -141,10 +141,11 @@ def disarming_fault(events, CS, *, temporary_ui_process_failure: bool = False, class AolCardIntent: def __init__(self, settings: AolSettings, *, explicit_latch: bool = False, lkas_button_supported: bool = True, - independent_cruise_buttons: bool = False): + independent_cruise_buttons: bool = False, preserve_unassigned_cancel: bool = False): self.settings = settings self.explicit_latch = explicit_latch self.independent_cruise_buttons = independent_cruise_buttons + self.preserve_unassigned_cancel = preserve_unassigned_cancel self.lkas_button_supported = lkas_button_supported self.allowed_latch = False # Observational only: records the exact branch that changed True to False. @@ -341,12 +342,12 @@ class AolCardIntent: (not independent_buttons or int(ButtonType.mainCruise) not in consumed_buttons)) or (self._lkas_held and lkas_toggle and (not independent_buttons or int(ButtonType.lkas) not in consumed_buttons))) - if not previously_held and gesture_held and (not independent_buttons or not cancel_pressed): + if not previously_held and gesture_held and (not (independent_buttons or self.preserve_unassigned_cancel) or not cancel_pressed): self._last_latch_edge_ns = now_ns if cancel_pressed: # An unassigned cruise CANCEL cannot toggle, arm or renew independent # lateral. Keep the existing withdrawal for configured CANCEL actions. - if not self.independent_cruise_buttons or any(self.settings.cancel_actions): + if not (self.independent_cruise_buttons or self.preserve_unassigned_cancel) or any(self.settings.cancel_actions): self._record_latch_clear('cancel_press') self.allowed_latch = False elif not self._gesture_neutral_seen: diff --git a/openpilot/starpilot/aol/tests/test_palisade_cancel.py b/openpilot/starpilot/aol/tests/test_palisade_cancel.py new file mode 100644 index 0000000000..9540746bfd --- /dev/null +++ b/openpilot/starpilot/aol/tests/test_palisade_cancel.py @@ -0,0 +1,214 @@ +"""Mixed Palisade Pause/Resume retains only existing, current lateral authority.""" +from dataclasses import replace +from types import SimpleNamespace +from typing import cast +import pytest +from opendbc.car import structs +from openpilot.cereal import log +from openpilot.selfdrive.selfdrived.events import Events +from openpilot.selfdrive.selfdrived.selfdrived import SelfdriveD +from openpilot.starpilot.aol.intent import AolSettings, AOL_TOGGLE +from openpilot.starpilot.aol.runtime import AxisDecision +from openpilot.starpilot.aol.wire import IntentState, SafetyState, PandaSlot +from openpilot.starpilot.aol.tests.test_classic_brake_withdrawal import profile +from openpilot.starpilot.car.hyundai.aol import create_intent + + +class Messages(dict): + ready = True + def all_checks(self, services=None): + return self.ready + + +@pytest.mark.parametrize('identity', ('palisade_hdai', 'palisade_hdaii')) +@pytest.mark.parametrize('case', ['retain', 'prior_off', 'settings_off', 'configured_cancel', 'intent_off', 'stale_source', + 'native_denied', 'native_expired', 'wrong_session', 'wrong_inventory', 'hard_event', + 'model', 'dm', 'reverse', 'steer_fault']) +def test_actual_selfdrived_cancel_keeps_only_current_owned_lateral(identity, case): + now = 4_000_000_000 + cp = profile(identity) + config = cp.safetyConfigs[0] + model, word = int(config.safetyModel.raw), int(config.safetyParam) + intent = IntentState('card', 1, now, now, now + 200_000_000, True, False, False, True, True) + slot = PandaSlot(0, 'panda', model, word, cp.alternativeExperience) + inventory = (slot,) + native = SafetyState(1, True, now, now + 200_000_000, model, word, True, False, True, False, + 'panda', 'axis', now, inventory) + cs = structs.CarState(canValid=True, gearShifter='drive', vEgo=10.) + cs.cruiseState.available = True + cs.buttonEvents = [structs.CarState.ButtonEvent(type='cancel', pressed=True)] + sm = Messages(driverMonitoringState=SimpleNamespace(alertLevel=0, lockout=False, alwaysOnLockout=False), + extrinsicsCalibration=SimpleNamespace(calStatus=log.ExtrinsicsCalibration.Status.calibrated)) + sd = SimpleNamespace(CP=cp, active=False, enabled=False, initialized=True, sm=sm, + aol_settings=AolSettings(True, 0., AOL_TOGGLE, 0, (0,0,0), (0,0,0)), + aol_axis_decision=AxisDecision(lateral_active=True, native_acknowledged=True), + aol_car_state_log_ns=now, aol_session_id='axis', aol_dm_lateral_inhibit=False, + nostalgia_paddle_cancel=False, events=Events()) + sd.events.add(log.OnroadEvent.EventName.buttonCancel) + if case == 'prior_off': + sd.aol_axis_decision = AxisDecision() + elif case == 'settings_off': + sd.aol_settings = replace(sd.aol_settings, enabled=False) + elif case == 'configured_cancel': + sd.aol_settings = replace(sd.aol_settings, cancel_actions=(3,0,0)) + elif case == 'intent_off': + intent = replace(intent, allowedLatch=False) + elif case == 'stale_source': + sd.aol_car_state_log_ns = now - 30_000_001 + intent = replace(intent, carStateLogMonoTime=sd.aol_car_state_log_ns) + elif case == 'native_denied': + native = replace(native, lateralAllowed=False) + elif case == 'native_expired': + native = replace(native, validUntilMonoTime=now-1) + elif case == 'wrong_session': + native = replace(native, axisSessionId='other') + elif case == 'wrong_inventory': + native = replace(native, pandaInventory=(*inventory[:-1], replace(slot, safetyParam=word ^ 0x80))) + elif case == 'hard_event': + sd.events.add(log.OnroadEvent.EventName.controlsMismatch) + elif case == 'model': + sm.ready = False + elif case == 'dm': + sm['driverMonitoringState'].lockout = True + elif case == 'reverse': + cs.gearShifter = 'reverse' + elif case == 'steer_fault': + cs.steerFaultPermanent = True + result = SelfdriveD._decide_aol_axes(cast(SelfdriveD, sd), cs, intent, native, now_ns=now) + assert result.lateral_active == (case == 'retain') + assert not result.longitudinal_active + assert log.OnroadEvent.EventName.buttonCancel in sd.events.names + + +@pytest.mark.parametrize('identity', ('palisade_hdai', 'palisade_hdaii')) +@pytest.mark.parametrize('duration', (61_047_724, 221_004_624)) +@pytest.mark.parametrize('cancel_actions', ((0, 0, 0), (3, 0, 0), (0, 3, 0), (0, 0, 3))) +def test_host_cancel_hold_keeps_only_unassigned_preexisting_latch(identity, duration, cancel_actions): + cp = profile(identity) + settings = AolSettings(True, 0., AOL_TOGGLE, 0, (0, 0, 0), cancel_actions) + owner = create_intent(cp, settings) + cs = structs.CarState(canValid=True, gearShifter='drive') + cs.cruiseState.available = True + owner.update(cs, now_ns=1_000_000_000) + cs.buttonEvents = [structs.CarState.ButtonEvent(type='lkas', pressed=True)] + owner.update(cs, now_ns=1_010_000_000) + assert owner.allowed_latch + edge, arm = owner._last_latch_edge_ns, owner.physical_arm_ns + cs.buttonEvents = [structs.CarState.ButtonEvent(type='lkas', pressed=False)] + owner.update(cs, now_ns=1_020_000_000) + for stamp, events in ((1_030_000_000, [('cancel', True)]), + (1_040_000_000, []), (1_030_000_000 + duration, [('cancel', False)])): + cs.buttonEvents = [structs.CarState.ButtonEvent(type=kind, pressed=pressed) for kind, pressed in events] + owner.update(cs, now_ns=stamp) + assert owner.allowed_latch == (not any(cancel_actions)) + assert (owner._last_latch_edge_ns, owner.physical_arm_ns) == (edge, arm) + assert not owner.independent_cruise_buttons # MAIN/LKAS semantics are not changed. + + +@pytest.mark.parametrize('identity', ('palisade_hdai', 'palisade_hdaii')) +@pytest.mark.parametrize('simultaneous', (None, 'lkas', 'mainCruise')) +def test_cold_cancel_never_arms_or_renews_gesture(identity, simultaneous): + owner = create_intent(profile(identity), AolSettings(True, 0., AOL_TOGGLE, 0, (0,0,0), (0,0,0))) + cs = structs.CarState(canValid=True, gearShifter='drive') + cs.cruiseState.available = True + owner.update(cs, now_ns=1_000_000_000) + buttons = [structs.CarState.ButtonEvent(type='cancel', pressed=True)] + if simultaneous: + buttons.append(structs.CarState.ButtonEvent(type=simultaneous, pressed=True)) + cs.buttonEvents = buttons + owner.update(cs, now_ns=1_010_000_000) + assert not owner.allowed_latch + assert owner._last_latch_edge_ns == owner.physical_arm_ns == 0 + + +@pytest.mark.parametrize('identity', ('palisade_hdai', 'palisade_hdaii')) +def test_card_retains_original_cancel_receipt_until_lateral_only_ack(identity): + """Real Card caller, synthetic transport; parser receipt is tested separately.""" + from unittest.mock import Mock, patch + from opendbc.car.hyundai.blended_longitudinal import BlendedStartup, candidate_from_stock + from opendbc.car.hyundai.tests.test_palisade_2023 import params as mixed_params + from openpilot.common.params import Params + from openpilot.common.prefix import OpenpilotPrefix + from openpilot.starpilot.aol.tests.test_ioniq_availability import IoniqAvailabilityTests, car_state + + case = IoniqAvailabilityTests(methodName='runTest') + with OpenpilotPrefix(): + try: + stock = mixed_params('hdaii' if identity == 'palisade_hdaii' else 'hdai') + cp = candidate_from_stock(stock, alpha_requested=True, native_qualified=True) + startup = BlendedStartup(stock, cp.as_reader().as_builder(), (Mock(), Mock())) + startup.prepare(admission=lambda: True) + case.addCleanup(startup.close) + Params().put('LKASButtonControl', AOL_TOGGLE, block=True) + sd, _, drive, card, receive = case._ioniq_availability_callers(True, cp=cp, startup_owner=startup) + sd.aol_settings = card.aol_settings + source = 10_020_000_000 + cs = car_state() + cs.cruiseState.available = True + cs.buttonEvents = [structs.CarState.ButtonEvent(type='cancel', pressed=True)] + # Native withdrawal arrives before the next SelfdriveD publication. + drive(2, native_request_lat=True, native_request_long=True, long_allowed=False, + panda_controls_allowed=False) + old_axis = sd.pm.messages['aolAxisState'].as_builder() + old_axis.logMonoTime = old_axis.aolAxisState.observedMonoTime = source - 1_000_000 + old_axis.aolAxisState.sourceCarStateMonoTime = source - 1_000_000 + old_axis.aolAxisState.validUntilMonoTime = source + 29_000_000 + old_axis.aolAxisState.mode = 'combined' + old_axis.aolAxisState.desiredLongitudinal = old_axis.aolAxisState.longitudinalActive = True + with patch.object(card, 'aol_cancel_source', return_value=source): + assert receive(source, axis=old_axis.as_reader(), state=cs) + assert card.aol_transport_feedback.longitudinal_cancel_ns == source + # No second physical press, no newer axis yet, now the cancellation event. + event = log.Event.new_message(logMonoTime=source) + event.init('onroadEvents', 1) + event.onroadEvents[0].name = 'buttonCancel' + event.onroadEvents[0].userDisable = True + event.onroadEvents[0].noEntry = True + cs.buttonEvents = [] + with patch.object(card, 'aol_cancel_source', return_value=0): + assert receive(source + 10_000_000, event=event.as_reader(), state=cs) + assert card.aol_transport_feedback.longitudinal_cancel_ns == source + assert not card.aol_transport_feedback.failed + # SelfdriveD receives the same physical press; its actual LAT-only output + # and the native LAT-only request close the bounded withdrawal handshake. + cancel = car_state() + cancel.buttonEvents = [structs.CarState.ButtonEvent(type='cancel', pressed=True)] + axis, state, command = drive(4, observed_state=cancel, native_request_lat=True, + native_request_long=False, long_allowed=False, panda_controls_allowed=False) + assert not state.enabled and axis.lateralActive and command.latActive + assert not axis.longitudinalActive and not command.longActive + with patch.object(card, 'aol_cancel_source', return_value=0): + assert receive(source + 20_000_000, axis=sd.pm.messages['aolAxisState'], + event=sd.pm.messages['onroadEvents'], state=cs) + assert card.aol_card_intent.allowed_latch + assert card.aol_transport_feedback.longitudinal_cancel_ns == 0 + finally: + case.doCleanups() + + +@pytest.mark.parametrize('identity', ('palisade_hdai', 'palisade_hdaii')) +@pytest.mark.parametrize('fault', ('none', 'wrong_bus', 'stale', 'unmarked', 'stock')) +def test_cancel_source_is_bound_to_real_mixed_parser_receipt(identity, fault): + from opendbc.can import CANPacker + from opendbc.car import Bus + from opendbc.car.hyundai.interface import CarInterface + from openpilot.selfdrive.car.card import Car + cp = profile(identity) + if fault == 'unmarked': + cp.alternativeExperience = 0 + elif fault == 'stock': + cp.openpilotLongitudinalControl, cp.pcmCruise = False, True + cp.safetyConfigs[0].safetyParam &= ~4 + ci = CarInterface(cp) + parser = ci.can_parsers[Bus.pt] + packer = CANPacker(parser.dbc.name) + for tick in range(15): + now = 2_000_000_000 + tick * 20_000_000 + packet = packer.make_can_msg('CLU11', 2 if fault == 'wrong_bus' else parser.bus, + {'CF_Clu_CruiseSwState': 4 if tick == 14 else 0, 'CF_Clu_AliveCnt1': tick}) + cs = ci.update([(now, [packet])]) + if fault == 'wrong_bus': + cs.buttonEvents = [structs.CarState.ButtonEvent(type='cancel', pressed=True)] + received = now - 30_000_001 if fault == 'stale' else now + caller = SimpleNamespace(CI=ci, _aol_can_observation=(cs, received, now, now)) + assert Car.aol_cancel_source(cast(Car, caller), cs, now) == (now if fault == 'none' else 0) diff --git a/openpilot/starpilot/aol/transport_pause.py b/openpilot/starpilot/aol/transport_pause.py index be92dd56df..ef1c874889 100644 --- a/openpilot/starpilot/aol/transport_pause.py +++ b/openpilot/starpilot/aol/transport_pause.py @@ -805,11 +805,12 @@ class TransportPauseFeedback: def _longitudinal_cancel_current(self, sm, cp, cs, axis, now_ns, *, latched, cancel_source_ns, withdrawal_required=True) -> SafetyState | None: from opendbc.car.hyundai.canfd_stock_aol import qualified_long as qualified from openpilot.starpilot.car.hyundai.aol import qualified_ccnc_long_aol as ev9_qualified + from opendbc.car.hyundai.blended_stock_aol import qualified_alpha as mixed_qualified from opendbc.car.structs import car if (not latched or self.failed or self.paused or self.pending_event or not self.session or str(axis.sessionId) != self.session or - not (qualified(cp, marked_only=True) or ev9_qualified(cp, marked_only=True)) or + not (qualified(cp, marked_only=True) or ev9_qualified(cp, marked_only=True) or mixed_qualified(cp, marked_only=True)) or not self.lateral_baseline_ns or not 0 <= now_ns - self.lateral_baseline_ns <= AXIS_MAX_AGE_NS or str(axis.faultReason) != 'none' or not axis.nativeAcknowledged or not axis.desiredLateral or not axis.lateralActive or cs.accFaulted or diff --git a/openpilot/starpilot/car/hyundai/aol.py b/openpilot/starpilot/car/hyundai/aol.py index a73d137d31..4ecd39443a 100644 --- a/openpilot/starpilot/car/hyundai/aol.py +++ b/openpilot/starpilot/car/hyundai/aol.py @@ -388,7 +388,8 @@ 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 independent_cruise_buttons_supported(CP, marked_only=True)) and + return bool((ev6_long or qualified_blended_alpha(CP, marked_only=True) 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 @@ -427,9 +428,10 @@ def retained_shared_long_cancel_intent(CP, CS, sm, feedback, owner, *, cancel_so from openpilot.starpilot.car.hyundai.ev6_intent import Ev6CardIntent ev6_long = bool(isinstance(owner, Ev6CardIntent) and owner._buttons_current and owner.physical.authorized and owner._last_latch_edge_ns > 0) - if (not (ev6_long or qualified_sonata_independent(CP) or qualified_ccnc_long_aol(CP, marked_only=True) or + mixed_long = qualified_blended_alpha(CP, marked_only=True) + if (not (ev6_long or mixed_long or qualified_sonata_independent(CP) or qualified_ccnc_long_aol(CP, marked_only=True) or CP.carFingerprint in INDEPENDENT_LONG_BUTTON_CARS and qualified_canfd_long(CP, marked_only=True)) or - owner is None or not owner.explicit_latch or not (owner.independent_cruise_buttons or ev6_long) or + owner is None or not owner.explicit_latch or not (owner.independent_cruise_buttons or ev6_long or mixed_long) or not owner.allowed_latch or owner.pause_lateral or owner.pause_longitudinal or owner._fault_inhibit): return False settings = owner.settings diff --git a/openpilot/starpilot/car/hyundai/classic_long_intent.py b/openpilot/starpilot/car/hyundai/classic_long_intent.py index 3fcd1127e3..c062685022 100644 --- a/openpilot/starpilot/car/hyundai/classic_long_intent.py +++ b/openpilot/starpilot/car/hyundai/classic_long_intent.py @@ -1,4 +1,5 @@ from opendbc.car.hyundai.classic_long_aol import independent_buttons_qualified +from opendbc.car.hyundai.blended_stock_aol import qualified_alpha from openpilot.starpilot.aol.intent import AolCardIntent @@ -7,7 +8,8 @@ class ClassicLongCardIntent(AolCardIntent): def __init__(self, cp, settings): super().__init__(settings, explicit_latch=True, independent_cruise_buttons=( - independent_buttons_qualified(cp, marked_only=False))) + independent_buttons_qualified(cp, marked_only=False)), + preserve_unassigned_cancel=qualified_alpha(cp)) self.requires_fault_observation = True self.observe_stock_engagement = True diff --git a/openpilot/starpilot/car/tesla/screen_intent.py b/openpilot/starpilot/car/tesla/screen_intent.py index 521cc3b3f7..697c6bfde1 100644 --- a/openpilot/starpilot/car/tesla/screen_intent.py +++ b/openpilot/starpilot/car/tesla/screen_intent.py @@ -21,7 +21,9 @@ class TeslaScreenCardIntent(AolCardIntent): super().update(filtered.as_reader(), **kwargs) engaged = bool(CS.cruiseState.enabled) available = bool(CS.cruiseState.available) - if (engaged and not self.previous_enabled) or (self.previous_available and not available and not CS.brakePressed): + # Stock cruise availability does not revoke an explicit screen selection. + if (engaged and not self.previous_enabled) or (self.previous_available and not available and not CS.brakePressed and + self.screen_override is not True): self.screen_override = None if engaged and not self.previous_enabled else False self.brake_blocked = False self.pause_lateral = False diff --git a/openpilot/starpilot/car/tesla/tests/test_screen_intent.py b/openpilot/starpilot/car/tesla/tests/test_screen_intent.py index bfdfb944b4..ad167b946e 100644 --- a/openpilot/starpilot/car/tesla/tests/test_screen_intent.py +++ b/openpilot/starpilot/car/tesla/tests/test_screen_intent.py @@ -181,3 +181,52 @@ def test_availability_off_still_allows_a_new_ordinary_cruise_engagement(): assert not owner.allowed_latch owner.update(state(enabled=True).as_reader(), now_ns=50) assert owner.allowed_latch and owner.screen_override is None + + +@pytest.mark.parametrize('brake_disengage', [False, True]) +def test_stock_cruise_availability_cannot_erase_existing_screen_arm(brake_disengage): + owner = intent(brake_disengage) + owner.update(state(available=False, press=False).as_reader(), now_ns=10) + owner.update(state(available=False, press=True).as_reader(), now_ns=20) + owner.update(state(available=True, press=False).as_reader(), now_ns=30) + assert owner.screen_override is True + for index, available in enumerate((False, True, False, True)): + cs = state(available=available) + cs.gasPressed = bool(index % 2) + owner.update(cs.as_reader(), now_ns=40 + index) + assert owner.allowed_latch and owner.screen_override is True + owner.update(state(press=True).as_reader(), now_ns=50) + assert not owner.allowed_latch and owner.pause_lateral + owner.update(state(available=False, press=False).as_reader(), now_ns=60) + owner.update(state(available=True).as_reader(), now_ns=70) + assert not owner.allowed_latch + + +def test_cruise_baseline_availability_loss_does_not_create_screen_arm(): + owner = intent() + owner.update(state().as_reader(), now_ns=10) + assert owner.allowed_latch and owner.screen_override is None + owner.update(state(available=False).as_reader(), now_ns=20) + assert not owner.allowed_latch and owner.screen_override is False + owner.update(state().as_reader(), now_ns=30) + assert not owner.allowed_latch + + +@pytest.mark.parametrize('fault', ['canTimeout', 'doorOpen', 'steerFaultPermanent', 'steeringDisengage', 'accFaulted', 'cancel', 'brake']) +def test_availability_loss_cannot_hide_fault_cancel_or_configured_brake(fault): + owner = intent(brake=fault == 'brake') + owner.update(state(available=False, press=False).as_reader(), now_ns=10) + owner.update(state(available=False, press=True).as_reader(), now_ns=20) + owner.update(state(press=False).as_reader(), now_ns=30) + cs = state(available=False) + if fault == 'cancel': + cs.buttonEvents = [structs.CarState.ButtonEvent(type=ButtonType.cancel, pressed=True)] + elif fault == 'brake': + cs.brakePressed = True + else: + setattr(cs, fault, True) + owner.update(cs.as_reader(), now_ns=40) + assert not owner.allowed_latch + for index, available in enumerate((False, True, False, True)): + owner.update(state(available=available).as_reader(), now_ns=50 + index) + assert not owner.allowed_latch diff --git a/openpilot/starpilot/car/tesla/tests/test_screen_pedal_continuity.py b/openpilot/starpilot/car/tesla/tests/test_screen_pedal_continuity.py new file mode 100644 index 0000000000..2102f9226a --- /dev/null +++ b/openpilot/starpilot/car/tesla/tests/test_screen_pedal_continuity.py @@ -0,0 +1,100 @@ +"""Independent screen arming through real parser, intent, native and axis callers.""" +from types import SimpleNamespace + +import pytest + +from opendbc.can import CANPacker +from opendbc.car.tesla.interface import CarInterface +from opendbc.car.tesla.screen_button import apply_screen_button +from opendbc.car.tesla.tests.test_screen_button import screen_params +from opendbc.safety.tests.libsafety import libsafety_py +from openpilot.cereal import log +from openpilot.selfdrive.selfdrived.events import Events +from openpilot.selfdrive.selfdrived.selfdrived import SelfdriveD +from openpilot.starpilot.aol.intent import AolSettings +from openpilot.starpilot.aol.runtime import AxisDecision +from openpilot.starpilot.aol.wire import IntentState, PandaSlot, SafetyState +from openpilot.starpilot.car.tesla.screen_intent import TeslaScreenCardIntent + + +class HealthyPeers(dict): + def all_checks(self, services): + return all(service in self for service in services) + + +@pytest.mark.parametrize('cooperative', [False, True]) +@pytest.mark.parametrize('brake_disengage', [False, True]) +def test_explicit_screen_axis_survives_sparse_touch_and_stock_availability(cooperative, brake_disengage): + cp = screen_params() + apply_screen_button(cp, True, brake_disengage) + cp.safetyConfigs[0].safetyParam |= 256 * cooperative + cp.alternativeExperience = 32 + ci = CarInterface(cp) + owner = TeslaScreenCardIntent(cp, AolSettings(True, 0., 0, 0, (0, 0, 0), (0, 0, 0))) + packer = CANPacker('tesla_model3_party') + peers = HealthyPeers(modelV2=SimpleNamespace(), + extrinsicsCalibration=SimpleNamespace(calStatus=log.ExtrinsicsCalibration.Status.calibrated), + driverMonitoringState=SimpleNamespace(lockout=False, alwaysOnLockout=False, alertLevel=0)) + sd = SimpleNamespace(CP=cp, active=False, enabled=False, initialized=True, sm=peers, aol_settings=owner.settings, + aol_axis_decision=AxisDecision(), aol_car_state_log_ns=0, aol_session_id='screen-test', + aol_dm_lateral_inhibit=False, nostalgia_paddle_cancel=False, aol_cruise_main_required=False, + events=Events()) + native = libsafety_py.libsafety + native.init_tests() + native.set_alternative_experience(32) + native.set_safety_hooks(10, cp.safetyConfigs[0].safetyParam) + native.set_aol_test_heartbeat(True) + try: + for tick in range(250): + now = 1_000_000_000 + tick * 10_000_000 + native.set_timer(now // 1000) + stock_state = 1 if 40 <= tick < 80 or 140 <= tick < 180 else 0 + messages = ( + ('DAS_control', 2, {}), ('DAS_steeringControl', 2, {}), + ('SCCM_steeringAngleSensor', 2, {}), ('DAS_status', 2, {}), ('DAS_settings', 2, {}), + ('DI_speed', 0, {'DI_vehicleSpeed': 0}), + ('ESP_B', 0, {'ESP_vehicleSpeed': 0, 'ESP_wheelSpeedsQF': 1, 'ESP_vehicleStandstillSts': 1}), + ('EPAS3S_sysStatus', 0, {'EPAS3S_internalSAS': 0, 'EPAS3S_eacStatus': 1, + 'EPAS3S_handsOnLevel': 3 if tick >= 240 else 0, 'EPAS3S_torsionBarTorque': 0}), + ('DI_systemStatus', 0, {'DI_gear': 4, 'DI_accelPedalPos': 20 if tick % 30 < 15 else 0}), + ('ESP_status', 0, {'ESP_driverBrakeApply': 2 if not brake_disengage and 100 <= tick < 120 else 1}), + ('UI_warning', 0, {'buckleStatus': 1}), + ('DI_state', 0, {'DI_cruiseState': stock_state}), + ) + frames = [packer.make_can_msg(name, bus, values) for name, bus, values in messages] + if tick in (20, 21): + touch = bytearray(8) + touch[3] = 3 if tick == 21 else 0 + frames.append((0x3DF, bytes(touch), 1)) + for address, data, bus in frames: + assert native.safety_rx_hook(libsafety_py.make_CANPacket(address, bus, data)) + native.safety_tick() + cs = ci.update([(now, frames)]).as_reader() + if tick > 1: + assert cs.canValid and not cs.canTimeout + owner.update(cs, now_ns=now) + allowed, pause, long_pause = owner.output(cs) + intention = IntentState('screen-card', tick + 1, now, now, now + 30_000_000, allowed, pause, long_pause, True, allowed) + permission = native.aol_get_permission_mask() + request = native.aol_get_request_mask() + safety = SafetyState(1, True, now, now + 200_000_000, 10, cp.safetyConfigs[0].safetyParam, + bool(permission & 1), bool(permission & 2), bool(request & 1), bool(request & 2), + 'test-panda', 'screen-test', now, + (PandaSlot(0, 'test-panda', 10, cp.safetyConfigs[0].safetyParam, 32, bool(native.get_controls_allowed())),)) + sd.aol_car_state_log_ns = now + sd.events = Events() + sd.events.add(log.OnroadEvent.EventName.pcmDisable) + if not cs.cruiseState.available: + sd.events.add(log.OnroadEvent.EventName.wrongCarMode) + decision = SelfdriveD._decide_aol_axes(sd, cs, intention, safety, now_ns=now) + sd.aol_axis_decision = decision + native.aol_set_host_request(int(decision.desired_lateral)) + assert not decision.longitudinal_active + if 22 <= tick < 240: + assert decision.lateral_active + elif tick >= 240: + assert not decision.lateral_active + finally: + native.init_tests() + native.set_alternative_experience(0) + native.set_safety_hooks(0, 0)