Detect unresponsive steering on stock ASCM Volts

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)
 

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):

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)

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:

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])

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')

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

prebuilt.json

@@ -1 +1 @@
-{"protected_inventory_sha256": "50ee49f3154d54503b9f342aa72fc3e03f597166656a6aa0a0760fae2124fca8", "schema_version": 1}
+{"protected_inventory_sha256": "a8c010257884676234f14033c10430e1b02dd6ac43e5a25a81500df23b54a6ee", "schema_version": 1}

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",