diff --git a/opendbc_repo/opendbc/car/gm/steering_authority.py b/opendbc_repo/opendbc/car/gm/steering_authority.py index 5858fc4dcf..c682aa4b9d 100644 --- a/opendbc_repo/opendbc/car/gm/steering_authority.py +++ b/opendbc_repo/opendbc/car/gm/steering_authority.py @@ -1,5 +1,6 @@ from opendbc.car import structs -from opendbc.car.gm.values import GMFlags, camera_acc_pedal_profile, is_volt_ascm_longitudinal, is_volt_sdgm_accepted_envelope +from opendbc.car.gm.values import (GMFlags, camera_acc_pedal_profile, is_volt_ascm_longitudinal, + is_volt_ascm_stock, is_volt_sdgm_accepted_envelope) # PSCMStatus is expected at 10 Hz. Three periods bound first acquisition and missing feedback. EPS_STATUS_TIMEOUT_NS = 300_000_000 @@ -8,7 +9,7 @@ ButtonType = structs.CarState.ButtonEvent.Type def monitored_profile(CP: structs.CarParams) -> bool: - return ((is_volt_ascm_longitudinal(CP) or is_volt_sdgm_accepted_envelope(CP)) and + return ((is_volt_ascm_longitudinal(CP) or is_volt_ascm_stock(CP) or is_volt_sdgm_accepted_envelope(CP)) and camera_acc_pedal_profile(CP) is None and not CP.flags & GMFlags.PEDAL_LONG) diff --git a/opendbc_repo/opendbc/car/gm/tests/test_ascm_intercept.py b/opendbc_repo/opendbc/car/gm/tests/test_ascm_intercept.py index e9760f0f49..b7d6c6eea2 100644 --- a/opendbc_repo/opendbc/car/gm/tests/test_ascm_intercept.py +++ b/opendbc_repo/opendbc/car/gm/tests/test_ascm_intercept.py @@ -199,8 +199,8 @@ class TestAscmIntercept(unittest.TestCase): cs = self.authority_frame(ci, packer, tick, status=0 if tick < 65 else 1) self.assertFalse(cs.steerFaultTemporary, tick) self.assertTrue(ci.CS.steering_authority.seen_active) - for cp in (params(CAR.CHEVROLET_VOLT_ASCM), params(CAR.CHEVROLET_VOLT), - params(CAR.CHEVROLET_BOLT_EUV)): + self.assertTrue(CarState(params(CAR.CHEVROLET_VOLT_ASCM)).steering_authority.enabled) + for cp in (params(CAR.CHEVROLET_VOLT), params(CAR.CHEVROLET_BOLT_EUV)): self.assertFalse(CarState(cp).steering_authority.enabled) def test_volt_steering_neutral_reacquisition(self): diff --git a/opendbc_repo/opendbc/car/gm/tests/test_stock_ascm_steering_authority.py b/opendbc_repo/opendbc/car/gm/tests/test_stock_ascm_steering_authority.py new file mode 100644 index 0000000000..5266fd27e3 --- /dev/null +++ b/opendbc_repo/opendbc/car/gm/tests/test_stock_ascm_steering_authority.py @@ -0,0 +1,168 @@ +import unittest + +from opendbc.car import structs +from opendbc.car.gm.values import CAR, GMFlags +from opendbc.car.gm.steering_authority import SteeringAuthority, monitored_profile + + +def params(): + # Stock-longitudinal ASCM profile from route5a; no private vehicle data. + cp = structs.CarParams.new_message( + brand='gm', + carFingerprint=str(CAR.CHEVROLET_VOLT_ASCM), + flags=16, + networkLocation='fwdCamera', + pcmCruise=True, + openpilotLongitudinalControl=False, + alphaLongitudinalAvailable=False, + radarUnavailable=True, + minSteerSpeed=3.1292800903320312, + alternativeExperience=32, + ) + cp.safetyConfigs = [{'safetyModel': 'gm', 'safetyParam': 0x605}] + return cp + + +class TestStockAscmSteeringAuthority(unittest.TestCase): + def setUp(self): + self.cp = params() + self.monitor = SteeringAuthority(self.cp) + + def step(self, tick, status=1, active=True, pause=None, missing=False, malformed=False): + ns = 1_000_000_000 + tick * 10_000_000 + cs = structs.CarState.new_message( + canValid=True, + canTimeout=False, + vEgo=17.0, + standstill=False, + gearShifter='low', + steeringPressed=pause == 'driver', + brakePressed=pause == 'brake', + regenBraking=pause == 'paddle', + ) + cs.cruiseState.available = pause != 'main' + if pause == 'stopped': + cs.standstill = True + cs.vEgo = 0.0 + if pause == 'slow': + cs.vEgo = 2.0 + if pause == 'cancel': + cs.buttonEvents = [structs.CarState.ButtonEvent(pressed=True, type='cancel')] + cs.steerFaultTemporary = status == 2 + cs.steerFaultPermanent = status == 3 + packets = [(0xC9, b'\x00' * 8, 0), (0xBE, b'\x00' * 6, 0), (0xBD, b'\x00' * 7, 0), (0x1E1, b'\x00' * 7, 0)] + if tick % 10 == 0 and not missing: + packets.append((0x184, b'\x00' if malformed else bytes([status << 3]) + b'\x00' * 7, 0)) + self.monitor.observe([(ns, packets)]) + self.monitor.update(cs, ns) + self.monitor.command(ns, active) + return cs + + def prime(self): + for tick in range(40): + self.step(tick) + self.assertTrue(self.monitor.seen_active) + + def test_exact_recorded_profile_enabled(self): + self.assertTrue(self.monitor.enabled) + + def test_both_brake_sources_radar_and_bsm_variants(self): + for word in (0x205, 0x605, 0xA05, 0xE05): + for bsm in (0, 16): + with self.subTest(word=word, bsm=bsm): + cp = params() + cp.safetyConfigs[0].safetyParam = word + cp.radarUnavailable = not bool(word & 0x800) + cp.flags = bsm + self.assertTrue(monitored_profile(cp)) + + def test_neighbor_profiles_and_wrong_owners_refused(self): + for field, value in [ + ('brand', 'hyundai'), + ('carFingerprint', str(CAR.CHEVROLET_BOLT_EUV)), + ('networkLocation', 'gateway'), + ('passive', True), + ('dashcamOnly', True), + ('notCar', True), + ('pcmCruise', False), + ('openpilotLongitudinalControl', True), + ('flags', int(GMFlags.PEDAL_LONG)), + ('flags', int(GMFlags.NO_CAMERA)), + ('radarUnavailable', False), + ]: + with self.subTest(field=field): + cp = params() + setattr(cp, field, value) + self.assertFalse(monitored_profile(cp)) + for word in (5, 0x1005, 0xC163, 0xC151, 0x4087, 0x605 ^ 0x10, 0x605 ^ 0x100, 0x605 ^ 0x8000): + with self.subTest(word=word): + cp = params() + cp.safetyConfigs[0].safetyParam = word + self.assertFalse(monitored_profile(cp)) + + def test_unresponsive_eps_latches_and_healthy_feedback_cannot_auto_clear(self): + self.prime() + for tick in range(40, 81): + cs = self.step(tick, status=0) + self.assertTrue(self.monitor.latched) + self.assertTrue(cs.steerFaultTemporary) + for tick in range(81, 120): + cs = self.step(tick, status=1) + self.assertTrue(self.monitor.latched) + self.assertTrue(cs.steerFaultTemporary) + + def test_driver_pauses_and_positive_recovery(self): + for pause in ('brake', 'paddle', 'cancel', 'main', 'driver', 'stopped', 'slow'): + with self.subTest(pause=pause): + self.monitor = SteeringAuthority(self.cp) + self.prime() + for tick in range(40, 80): + self.step(tick, status=0, active=False, pause=pause) + self.assertFalse(self.monitor.latched) + for tick in range(80, 120): + self.step(tick, status=1) + self.assertFalse(self.monitor.latched) + self.assertTrue(self.monitor.seen_active) + + def test_stale_eps_feed_remains_a_fault(self): + self.prime() + for tick in range(40, 90): + cs = self.step(tick, missing=True, active=tick < 70) + self.assertTrue(self.monitor.latched) + self.assertTrue(cs.steerFaultTemporary) + + def test_malformed_eps_remains_a_fault(self): + self.prime() + for tick in range(40, 80): + cs = self.step(tick, malformed=True) + self.assertTrue(self.monitor.latched) + self.assertTrue(cs.steerFaultTemporary) + + def test_factory_fault_flags_preserved(self): + for status in (2, 3): + with self.subTest(status=status): + self.monitor = SteeringAuthority(self.cp) + self.prime() + for tick in range(40, 80): + cs = self.step(tick, status=status, active=False) + self.assertEqual(cs.steerFaultTemporary, status == 2) + self.assertEqual(cs.steerFaultPermanent, status == 3) + + def test_main_cycle_uses_neutral_and_fresh_rearm(self): + self.prime() + for tick in range(40, 81): + self.step(tick, status=0) + self.assertTrue(self.monitor.latched) + for tick in range(81, 86): + self.step(tick, status=0, active=False, pause='main') + self.assertTrue(self.monitor.latched) + self.step(86, status=0, active=False) + self.assertFalse(self.monitor.latched) + for tick in range(87, 120): + self.step(tick, status=1) + self.assertFalse(self.monitor.latched) + self.assertTrue(self.monitor.seen_active) + + +if __name__ == '__main__': + unittest.main(verbosity=2) diff --git a/opendbc_repo/opendbc/car/gm/values.py b/opendbc_repo/opendbc/car/gm/values.py index da464fae46..9ab212eeaa 100644 --- a/opendbc_repo/opendbc/car/gm/values.py +++ b/opendbc_repo/opendbc/car/gm/values.py @@ -359,6 +359,21 @@ def is_volt_ascm_longitudinal(cp: CarParams) -> bool: return False +def is_volt_ascm_stock(cp: CarParams) -> bool: + """Exact non-pedal ASCM Volt steering owner with factory longitudinal control.""" + try: + configs = cp.safetyConfigs + return (cp.brand == 'gm' and cp.carFingerprint == CAR.CHEVROLET_VOLT_ASCM and + cp.networkLocation == CarParams.NetworkLocation.fwdCamera and cp.pcmCruise and + not cp.openpilotLongitudinalControl and not cp.passive and not cp.dashcamOnly and + not cp.notCar and control_flags(cp) == 0 and len(configs) == 1 and + configs[0].safetyModel == CarParams.SafetyModel.gm and + int(configs[0].safetyParam) in (0x205, 0x605, 0xA05, 0xE05) and + bool(int(configs[0].safetyParam) & int(GMSafetyFlags.ASCM_RADAR)) is not cp.radarUnavailable) + except (AttributeError, IndexError, TypeError, ValueError): + return False + + def is_volt_camera_removed(cp: CarParams, *, longitudinal=None) -> bool: """The exact manual Volt with PT state sources and no factory camera.""" try: diff --git a/openpilot/starpilot/aol/tests/test_gm_ascm_receive_order.py b/openpilot/starpilot/aol/tests/test_gm_ascm_receive_order.py index 0dbb9239a0..2571ecd4b3 100644 --- a/openpilot/starpilot/aol/tests/test_gm_ascm_receive_order.py +++ b/openpilot/starpilot/aol/tests/test_gm_ascm_receive_order.py @@ -182,11 +182,11 @@ def test_raw_projection_owner_excludes_separate_pedal_profiles(topology, alpha, @pytest.mark.parametrize('sdgm', [False, True]) @pytest.mark.parametrize('alpha,sascm', [(False, True), (True, False)]) -def test_raw_projection_owner_excludes_stock_and_missing_sascm(sdgm, alpha, sascm): +def test_raw_projection_owner_includes_stock_ascm_without_changing_stock_sdgm(sdgm, alpha, sascm): from openpilot.starpilot.car.gm.aol import steering_raw_projection_owner cp = sdgm_params(alpha=alpha, sascm=sascm) if sdgm else ascm_params(CAR.CHEVROLET_VOLT_ASCM, alpha=alpha, sascm=sascm) cp.alternativeExperience = 32 - assert not steering_raw_projection_owner(cp) + assert steering_raw_projection_owner(cp) == (not sdgm) @pytest.mark.parametrize('full_op', [False, True]) diff --git a/openpilot/starpilot/aol/tests/test_gm_sdgm_source_wait.py b/openpilot/starpilot/aol/tests/test_gm_sdgm_source_wait.py index ad32efa721..8bb7b8cd9f 100644 --- a/openpilot/starpilot/aol/tests/test_gm_sdgm_source_wait.py +++ b/openpilot/starpilot/aol/tests/test_gm_sdgm_source_wait.py @@ -48,17 +48,20 @@ class Capture: self.messages[name] = messaging.log_from_bytes(raw) -def joined_case(cleanup, bridge, *, c9=False, pedal='none', engage=False): +def joined_case(cleanup, bridge, *, c9=False, pedal='none', engage=False, cp=None): saved = Params() cleanup.callback(saved._finalizer) for key in ('OpenpilotEnabledToggle', 'AlwaysOnLateral', 'AlphaLongitudinalEnabled'): saved.put_bool(key, True, block=True) - cp = sdgm_params(alpha=True, brake_c9=c9) + default_sdgm = cp is None + cp = sdgm_params(alpha=True, brake_c9=c9) if cp is None else cp card = gm_fixture.TestGmAol.card(cp, saved) cleanup.callback(card.vehicle_startup.close) - assert card.CP.flags == 16 and card.CP.alternativeExperience == 32 - if not c9: - assert card.CP.safetyConfigs[0].safetyParam == 0x5007 + assert card.CP.flags == cp.flags and card.CP.alternativeExperience == 32 + assert card.CP.safetyConfigs[0].safetyParam == cp.safetyConfigs[0].safetyParam + assert card.CP.pcmCruise == cp.pcmCruise and card.CP.openpilotLongitudinalControl == cp.openpilotLongitudinalControl + if default_sdgm and not c9: + assert card.CP.flags == 16 and card.CP.safetyConfigs[0].safetyParam == 0x5007 card.ci_initialized = True # No ECU/device I/O; native permission is earned below. card.pm = Capture() case = caller_fixture.IoniqAvailabilityTests(methodName='runTest') diff --git a/openpilot/starpilot/aol/tests/test_gm_stock_ascm_refusal.py b/openpilot/starpilot/aol/tests/test_gm_stock_ascm_refusal.py new file mode 100644 index 0000000000..55dc96a220 --- /dev/null +++ b/openpilot/starpilot/aol/tests/test_gm_stock_ascm_refusal.py @@ -0,0 +1,116 @@ +"""Stock ASCM refusal must withdraw actual callers and publish a takeover alert.""" +from contextlib import ExitStack +import os +from unittest import mock + +import pytest +from opendbc.can import CANPacker +from opendbc.car import Bus +from opendbc.car.gm import gmcan +from opendbc.safety.tests.libsafety import libsafety_py +from opendbc.car.gm.tests.test_ascm_intercept import params +from opendbc.car.gm.values import CAR, GMFlags, DBC +from openpilot.common.prefix import OpenpilotPrefix +from openpilot.selfdrive.selfdrived.events import EventName +from openpilot.starpilot.aol.tests.test_gm_sdgm_source_wait import BASE, joined_case +from openpilot.starpilot.aol.tests.gm_gap_cpp_bridge import gm_host_heartbeat_bridge as gm_host_heartbeat_bridge + + +@pytest.mark.parametrize('c9', [False, True]) +def test_stock_ascm_refusal_withdraws_output_and_publishes_takeover(gm_host_heartbeat_bridge, c9): + cp = params(CAR.CHEVROLET_VOLT_ASCM, accelerator=not c9, sascm=False, alpha=False) + cp.flags |= int(GMFlags.HAS_BSM) + assert cp.pcmCruise and not cp.openpilotLongitudinalControl + assert cp.safetyConfigs[0].safetyParam == (0x605 if c9 else 0x205) + with OpenpilotPrefix(), ExitStack() as cleanup, mock.patch.dict(os.environ, {'SIMULATION': '0', 'AOL_REPLAY_RUNTIME': '1'}): + card, sd, controls, native, current, physical, sample_native, card_receive, drive, apply_command = joined_case( + cleanup, gm_host_heartbeat_bridge, c9=c9, cp=cp) + assert not sd.enabled and not native.get_controls_allowed() + original_update = controls.sm.update_msgs + + def curved_model(timestamp, messages): + # Supply a nonzero model input; Controls, packing and native checks stay real. + changed = [] + for message in messages: + if message.which() == 'modelV2': + message = message.as_builder() + message.modelV2.action.desiredCurvature = .002 + message = message.as_reader() + changed.append(message) + return original_update(timestamp, changed) + + cleanup.enter_context(mock.patch.object(controls.sm, 'update_msgs', side_effect=curved_model)) + positive_torque = False + for tick in range(26, 42): + now = BASE + tick * 10_000_000 + rows = physical(now, tick, eps=1) + card_receive(now, rows) + axis, state, command = drive(now + 1, tick) + apply_command(now + 2, command) + assert axis.lateralActive and command.latActive and sample_native(now + 3) == 1 + positive_torque |= card.last_actuators_output.torqueOutputCan != 0 + assert positive_torque + warning = None + neutral_sent = False + for tick in range(42, 56): + now = BASE + tick * 10_000_000 + rows = physical(now, tick, eps=0) + card_receive(now, rows) + axis, state, command = drive(now + 1, tick) + apply_command(now + 2, command) + mask = sample_native(now + 3) + assert card.CI.CS.steering_authority.latched, { + 'monitorEnabled': card.CI.CS.steering_authority.enabled, + 'axisActive': axis.lateralActive, 'commandActive': command.latActive, + 'requestedTorque': command.actuators.torque, 'nativeMask': mask, 'alert': state.alertType, + } + assert current['last_cs'].steerFaultTemporary and not current['last_cs'].steerFaultPermanent + assert not axis.lateralActive and not command.latActive and not command.longActive + assert mask == 0 + sent = card.pm.messages.get('sendcan') + for frame in sent.sendcan if sent is not None else (): + if frame.address == 0x180 and frame.src == 0: + payload = bytes(frame.dat) + assert not payload[0] & 8 and ((payload[0] & 7) << 8 | payload[1]) == 0 + neutral_sent = True + assert EventName.controlsMismatch in sd.events.names + if state.alertText1: + warning = state + assert neutral_sent and not card.last_actuators_output.torqueOutputCan + assert warning is not None + assert warning.alertText1 == 'TAKE CONTROL IMMEDIATELY' + assert str(warning.alertSound) == 'warningImmediate' + delivered = controls.sm['selfdriveState'] + assert delivered.alertText1 == warning.alertText1 + assert str(delivered.alertSound) == 'warningImmediate' + # A valid small active command is rejected by the native side after OFF. + packer = CANPacker(DBC[card.CP.carFingerprint][Bus.pt]) + address, data, bus = gmcan.create_steering_control(packer, 0, 1, 0, True) + assert not native.safety_tx_hook(libsafety_py.make_CANPacket(address, bus, data)) + # Healthy feedback alone must not clear a refusal or restore actuation. + for tick in range(56, 66): + now = BASE + tick * 10_000_000 + rows = physical(now, tick, eps=1) + card_receive(now, rows) + axis, state, command = drive(now + 1, tick) + apply_command(now + 2, command) + assert card.CI.CS.steering_authority.latched + assert not axis.lateralActive and not command.latActive and sample_native(now + 3) == 0 + for tick in range(66, 71): + now = BASE + tick * 10_000_000 + rows = physical(now, tick, eps=1, main=False) + card_receive(now, rows) + axis, state, command = drive(now + 1, tick) + apply_command(now + 2, command) + assert not command.latActive and sample_native(now + 3) == 0 + recovered = False + for tick in range(71, 87): + now = BASE + tick * 10_000_000 + rows = physical(now, tick, eps=1, main=True) + card_receive(now, rows) + axis, state, command = drive(now + 1, tick) + apply_command(now + 2, command) + mask = sample_native(now + 3) + recovered |= bool(axis.lateralActive and command.latActive and mask == 1) + assert not command.longActive and not state.enabled + assert recovered and not card.CI.CS.steering_authority.latched diff --git a/prebuilt.json b/prebuilt.json index 262fd8618f..ddd06093cf 100644 --- a/prebuilt.json +++ b/prebuilt.json @@ -1 +1 @@ -{"protected_inventory_sha256": "50ee49f3154d54503b9f342aa72fc3e03f597166656a6aa0a0760fae2124fca8", "schema_version": 1} +{"protected_inventory_sha256": "a8c010257884676234f14033c10430e1b02dd6ac43e5a25a81500df23b54a6ee", "schema_version": 1} diff --git a/tools/ci/run_vehicle_tests.py b/tools/ci/run_vehicle_tests.py index 77e2916d26..2a061f7f8e 100644 --- a/tools/ci/run_vehicle_tests.py +++ b/tools/ci/run_vehicle_tests.py @@ -94,6 +94,7 @@ AOL_LIFECYCLE_TARGETS = ( "openpilot/starpilot/aol/tests/test_gm_sdgm_cold_main.py", "openpilot/starpilot/aol/tests/test_gm_main_level_contract.py", "openpilot/starpilot/aol/tests/test_gm_sdgm_source_wait.py", + "openpilot/starpilot/aol/tests/test_gm_stock_ascm_refusal.py", "openpilot/starpilot/aol/tests/test_gm_ascm_receive_order.py", "openpilot/starpilot/aol/tests/test_gm_main_response.py", "openpilot/starpilot/aol/tests/test_gm_held_native.py", @@ -488,6 +489,7 @@ RELEASE_TARGETS = ( "opendbc_repo/opendbc/car/gm/tests/test_volt_camera_removed.py", "opendbc_repo/opendbc/safety/tests/test_gm_volt_sdgm.py", "opendbc_repo/opendbc/car/gm/tests/test_volt_sdgm_control.py", + "opendbc_repo/opendbc/car/gm/tests/test_stock_ascm_steering_authority.py", "opendbc_repo/opendbc/car/gm/tests/test_sdgm_brake_reacquire.py", "opendbc_repo/opendbc/car/gm/tests/test_volt_camera_control.py", "opendbc_repo/opendbc/safety/tests/test_hyundai_mrr35_angle_four.py",