Bolt Gear State

openpilot/selfdrive/car/car_events.py

@@ -6,6 +6,7 @@ from opendbc.car.interfaces import MAX_CTRL_SPEED
 from opendbc.car.toyota.values import ToyotaFlags
 
 from openpilot.selfdrive.selfdrived.events import Events
+from openpilot.starpilot.car.events import event_rules, unsupported_gear
 
 ButtonType = structs.CarState.ButtonEvent.Type
 GearShifter = structs.CarState.GearShifter
@@ -16,6 +17,7 @@ NetworkLocation = structs.CarParams.NetworkLocation
 class CarEvents:
   def __init__(self, CP: structs.CarParams):
     self.CP = CP
+    self.event_rules = event_rules(CP)
 
     self.steering_unpressed = 0
     self.low_speed_alert = False
@@ -24,7 +26,7 @@ class CarEvents:
 
   def update(self, CS: car.CarState, CS_prev: car.CarState, CC: car.CarControl):
     if self.CP.brand in ('body', 'mock'):
-      return Events()
+      return Events(self.event_rules)
 
     events = self.create_common_events(CS, CS_prev)
 
@@ -94,7 +96,7 @@ class CarEvents:
     return events
 
   def create_common_events(self, CS: structs.CarState, CS_prev: car.CarState):
-    events = Events()
+    events = Events(self.event_rules)
 
     CI = interfaces[self.CP.carFingerprint]
     # TODO: cleanup the honda-specific logic
@@ -107,7 +109,8 @@ class CarEvents:
       events.add(EventName.doorOpen)
     if CS.seatbeltUnlatched:
       events.add(EventName.seatbeltNotLatched)
-    if CS.gearShifter != GearShifter.drive and CS.gearShifter not in CI.DRIVABLE_GEARS:
+    if ((CS.gearShifter != GearShifter.drive and CS.gearShifter not in CI.DRIVABLE_GEARS) or
+        unsupported_gear(self.CP, CS.gearShifter)):
       events.add(EventName.wrongGear)
     if CS.gearShifter == GearShifter.reverse:
       events.add(EventName.reverseGear)

openpilot/selfdrive/selfdrived/events.py

@@ -50,10 +50,11 @@ EVENT_NAME = {v: k for k, v in EventName.schema.enumerants.items()}
 
 
 class Events:
-  def __init__(self):
+  def __init__(self, event_rules: dict | None = None):
+    self.event_rules = EVENTS | event_rules if event_rules else EVENTS
     self.events: list[int] = []
     self.static_events: list[int] = []
-    self.event_counters = dict.fromkeys(EVENTS.keys(), 0)
+    self.event_counters = dict.fromkeys(self.event_rules.keys(), 0)
 
   @property
   def names(self) -> list[int]:
@@ -72,7 +73,7 @@ class Events:
     self.events = self.static_events.copy()
 
   def contains(self, event_type: str) -> bool:
-    return any(event_type in EVENTS.get(e, {}) for e in self.events)
+    return any(event_type in self.event_rules.get(e, {}) for e in self.events)
 
   def create_alerts(self, event_types: list[str], callback_args=None):
     if callback_args is None:
@@ -80,10 +81,10 @@ class Events:
 
     ret = []
     for e in self.events:
-      types = EVENTS[e].keys()
+      types = self.event_rules[e].keys()
       for et in event_types:
         if et in types:
-          alert = EVENTS[e][et]
+          alert = self.event_rules[e][et]
           if not isinstance(alert, Alert):
             alert = alert(*callback_args)
 
@@ -102,7 +103,7 @@ class Events:
     for event_name in self.events:
       event = log.OnroadEvent.new_message()
       event.name = event_name
-      for event_type in EVENTS.get(event_name, {}):
+      for event_type in self.event_rules.get(event_name, {}):
         setattr(event, event_type, True)
       ret.append(event)
     return ret

openpilot/selfdrive/selfdrived/selfdrived.py

@@ -173,7 +173,8 @@ class SelfdriveD:
 
     self.CS_prev = car.CarState.new_message()
     self.AM = AlertManager()
-    self.events = Events()
+    from openpilot.starpilot.car.events import event_rules
+    self.events = Events(event_rules(self.CP))
 
     self.initialized = False
     self.enabled = False

openpilot/starpilot/car/events.py

@@ -0,0 +1,12 @@
+def event_rules(cp) -> dict:
+  if cp.brand == "gm":
+    from openpilot.starpilot.car.gm.events import event_rules as gm_rules
+    return gm_rules(cp)
+  return {}
+
+
+def unsupported_gear(cp, gear) -> bool:
+  if cp.brand == "gm":
+    from openpilot.starpilot.car.gm.events import unsupported_gear as gm_unsupported
+    return gm_unsupported(cp, gear)
+  return False

openpilot/starpilot/car/gm/events.py

@@ -0,0 +1,22 @@
+from opendbc.car.gm.values import is_bolt_cc_profile
+from opendbc.car.structs import car
+from openpilot.cereal import log
+from openpilot.selfdrive.selfdrived.events import EVENTS, ET, ImmediateDisableAlert
+
+
+def qualified(cp) -> bool:
+  try:
+    return bool(is_bolt_cc_profile(cp))
+  except (AttributeError, IndexError, TypeError, ValueError):
+    return False
+
+
+def event_rules(cp) -> dict:
+  if not qualified(cp):
+    return {}
+  name = log.OnroadEvent.EventName.wrongGear
+  return {name: {**EVENTS[name], ET.USER_DISABLE: ImmediateDisableAlert("Gear not D")}}
+
+
+def unsupported_gear(cp, gear) -> bool:
+  return qualified(cp) and gear not in (car.CarState.GearShifter.drive, car.CarState.GearShifter.low)

openpilot/starpilot/car/gm/tests/test_gear_events.py

@@ -0,0 +1,133 @@
+import pytest
+
+from opendbc.car import gen_empty_fingerprint, structs
+from opendbc.car.gm.interface import CarInterface
+from opendbc.car.gm.values import BOLT_CC_WORDS, CAR, is_bolt_cc_profile
+from openpilot.cereal import log
+from openpilot.selfdrive.car.car_events import CarEvents
+from openpilot.selfdrive.selfdrived.events import EVENTS, ET, Events
+from openpilot.selfdrive.selfdrived.state import StateMachine
+from openpilot.starpilot.aol.intent import disarming_fault
+from openpilot.starpilot.car.events import event_rules
+
+Gear = structs.CarState.GearShifter
+Name = log.OnroadEvent.EventName
+PROFILES = [(identity, removed, word) for identity, words in BOLT_CC_WORDS.items()
+            for removed, word in enumerate(words)]
+
+
+def params(identity, removed=False):
+  fingerprint = gen_empty_fingerprint()
+  if not removed:
+    fingerprint[2][0x180] = 4
+  return CarInterface.get_params(identity, fingerprint, [], False, False, False)
+
+
+def state(gear, *, enable=False):
+  cs = structs.CarState(gearShifter=gear, vEgo=20.0, canValid=True, buttonEnable=enable)
+  cs.cruiseState.available = True
+  return cs
+
+
+def events_for(cp, cs, previous):
+  produced = CarEvents(cp).update(cs, previous, structs.CarControl())
+  received = Events(event_rules(cp))
+  received.add_from_msg(produced.to_msg())
+  return produced, received
+
+
+@pytest.mark.parametrize('identity,removed,word', PROFILES)
+@pytest.mark.parametrize('gear', (Gear.manumatic, Gear.unknown, Gear.park, Gear.neutral, Gear.reverse, Gear.sport, Gear.brake, Gear.eco))
+def test_checked_bolt_unsupported_gear_blocks_entry_and_disables_immediately(identity, removed, word, gear):
+  cp = params(identity, removed)
+  assert is_bolt_cc_profile(cp)
+  assert cp.safetyConfigs[0].safetyParam == word
+  previous = state(Gear.low)
+  current = state(gear, enable=True)
+  produced, received = events_for(cp, current, previous)
+  for events in (produced, received):
+    assert Name.wrongGear in events.names
+    assert events.contains(ET.NO_ENTRY)
+    assert events.contains(ET.USER_DISABLE)
+    assert events.contains(ET.SOFT_DISABLE)
+    message = next(event for event in events.to_msg() if event.name == Name.wrongGear)
+    assert message.noEntry and message.userDisable and message.softDisable and not message.immediateDisable
+    alerts = events.create_alerts([ET.NO_ENTRY, ET.USER_DISABLE])
+    assert any('Gear not D' in (alert.alert_text_1, alert.alert_text_2) for alert in alerts)
+    assert disarming_fault(events.to_msg(), current) == (gear == Gear.unknown)
+  machine = StateMachine()
+  assert not machine.update(received)[0]
+  _, allowed = events_for(cp, previous, state(Gear.drive))
+  allowed.add(Name.buttonEnable)
+  assert machine.update(allowed)[0]
+  assert not machine.update(received)[0]
+  assert machine.state == log.SelfdriveState.OpenpilotState.disabled
+  assert ET.USER_DISABLE in machine.current_alert_types
+  assert ET.SOFT_DISABLE not in machine.current_alert_types
+  _, recovered = events_for(cp, previous, current)
+  assert not machine.update(recovered)[0]
+  recovered.add(Name.buttonEnable)
+  assert machine.update(recovered)[0]
+
+
+@pytest.mark.parametrize('identity,removed,word', PROFILES)
+@pytest.mark.parametrize('gear', (Gear.drive, Gear.low))
+def test_checked_bolt_supported_gears_are_unchanged(identity, removed, word, gear):
+  cp = params(identity, removed)
+  produced, received = events_for(cp, state(gear), state(Gear.drive))
+  assert Name.wrongGear not in produced.names
+  assert Name.wrongGear not in received.names
+
+
+@pytest.mark.parametrize('identity', (CAR.CHEVROLET_VOLT_CC, CAR.CHEVROLET_VOLT, CAR.CHEVROLET_BOLT_EUV))
+def test_other_actual_gm_factories_keep_manumatic_admission(identity):
+  cp = params(identity)
+  assert not is_bolt_cc_profile(cp)
+  assert event_rules(cp) == {}
+  produced, received = events_for(cp, state(Gear.manumatic), state(Gear.drive))
+  assert Name.wrongGear not in produced.names
+  assert Name.wrongGear not in received.names
+
+
+def test_rules_are_instance_local_and_wrong_profile_cannot_change_global_classification():
+  cp = params(CAR.CHEVROLET_BOLT_CC_2018_2021)
+  rules = event_rules(cp)
+  scoped, ordinary = Events(rules), Events()
+  scoped.add(Name.wrongGear)
+  ordinary.add(Name.wrongGear)
+  assert scoped.contains(ET.USER_DISABLE)
+  assert ordinary.contains(ET.SOFT_DISABLE) and not ordinary.contains(ET.USER_DISABLE)
+  assert ET.USER_DISABLE not in EVENTS[Name.wrongGear]
+  altered = structs.CarParams.new_message(**cp.to_dict())
+  altered.safetyConfigs[0].safetyParam = 4
+  assert event_rules(altered) == {}
+  scoped.clear()
+  assert not scoped.contains(ET.USER_DISABLE)
+
+
+def test_pedal_factory_has_no_scoped_gear_rules():
+  fingerprint = gen_empty_fingerprint()
+  fingerprint[0][0x201] = 6
+  fingerprint[2][0x180] = 4
+  cp = CarInterface.get_params(CAR.CHEVROLET_BOLT_CC_2018_2021, fingerprint, [], False, False, False)
+  assert not is_bolt_cc_profile(cp)
+  assert event_rules(cp) == {}
+
+
+def test_unaffected_volt_reverse_pauses_legacy_aol_output_without_clearing_intent():
+  from openpilot.starpilot.aol.intent import AolCardIntent, AolSettings
+  cp = params(CAR.CHEVROLET_VOLT_CC)
+  owner = AolCardIntent(AolSettings(True, 0.0, 0, 0, (0, 0, 0), (0, 0, 0)))
+  drive = state(Gear.drive)
+  owner.update(drive)
+  assert owner.allowed_latch
+  reverse = state(Gear.reverse)
+  _, events = events_for(cp, reverse, drive)
+  assert Name.reverseGear in events.names
+  assert not disarming_fault(events.to_msg(), reverse)
+  owner.update(reverse, fault_active=disarming_fault(events.to_msg(), reverse))
+  assert owner.allowed_latch
+  assert not owner.output(reverse)[0]
+  owner.update(drive)
+  assert owner.allowed_latch
+  assert owner.output(drive)[0]

openpilot/starpilot/tests/test_bolt_cc_controls_manual.py

@@ -0,0 +1,143 @@
+import os
+import tempfile
+import unittest
+from unittest.mock import patch
+
+from openpilot.cereal import messaging
+from openpilot.common.params import Params
+from openpilot.common.prefix import OpenpilotPrefix
+from openpilot.selfdrive.controls.controlsd import Controls
+from openpilot.selfdrive.selfdrived.selfdrived import SelfdriveD
+from openpilot.selfdrive.selfdrived.events import EventName
+from opendbc.car import structs
+from opendbc.car.gm.bolt_cc import button_bytes
+from opendbc.car.gm.tests.test_bolt_cc import feed, fixture, native, setup
+from opendbc.car.gm.values import CAR, BOLT_CC_WORDS
+from opendbc.safety.tests.libsafety import libsafety_py
+
+
+class TestBoltCcControlsManual(unittest.TestCase):
+  def test_reported_manual_disables_and_requires_fresh_set_release(self):
+    for identity, words in BOLT_CC_WORDS.items():
+      for removed in (False, True):
+        with self.subTest(identity=identity, removed=removed), tempfile.TemporaryDirectory() as params_root, \
+             patch.dict(os.environ, {'PARAMS_ROOT': params_root, 'SIMULATION': '1', 'REPLAY': '1'}), OpenpilotPrefix():
+          params = Params()
+          for key in ('OpenpilotEnabledToggle',):
+            params.put_bool(key, True, block=True)
+          for key in ('AlwaysOnLateral', 'SafeMode', 'AdvancedLateralTune'):
+            params.put_bool(key, False, block=True)
+          cp, ci, packer = fixture(identity, removed=removed)
+          cp.fingerprintSource = structs.CarParams.FingerprintSource.can
+          cp.carFw = []
+          params.put('CarParams', cp.to_bytes(), block=True)
+          controls = Controls()
+          self.assertEqual(controls.CP.to_dict(), cp.to_dict())
+          self.assertEqual(controls.CP.safetyConfigs[0].safetyParam, words[int(removed)])
+          _, template = feed(ci, packer, 900_000_000, active=False, camera=not removed)
+          setup(cp)
+          with patch("openpilot.selfdrive.selfdrived.selfdrived.REPLAY", False):
+            sd = SelfdriveD(CP=cp)
+          previous_control = structs.CarControl()
+          active_frames = 0
+          for tick in range(520):
+            now = 1_000_000_000 + tick * 10_000_000
+            manual = tick < 60 or 340 <= tick < 360
+            gas = False
+            button = 3 if 40 <= tick < 42 or 90 <= tick < 92 or 390 <= tick < 392 else 1
+            replacement = [
+              packer.make_can_msg('ECMPRDNL2', 0, {'PRNDL2': 6, 'ManualMode': int(manual)}),
+              packer.make_can_msg('ECMCruiseControl', 0, {'CruiseActive': int(tick >= 92), 'CruiseSetSpeed': 54}),
+              packer.make_can_msg('AcceleratorPedal2', 0, {'AcceleratorPedal2': 30 if gas else 0}),
+              (0x1E1, button_bytes(button, tick % 4), 0),
+            ]
+            if identity == CAR.CHEVROLET_BOLT_ACC_2022_2023_PEDAL and not removed:
+              replacement.append(packer.make_can_msg('ASCMActiveCruiseControlStatus', 2, {'ACCCmdActive': int(tick >= 92)}))
+            replaced = {frame[0] for frame in replacement}
+            frames = [frame for frame in template if frame[0] not in replaced] + replacement
+            for frame in sorted(frames, key=lambda frame: frame[0] == 0x3D1):
+              if frame[0] in (0x184, 0x3D1, 0x1E1, 0xC9, 0x1C4, 0x1F5, 0x34A, 0x370):
+                self.assertTrue(native('rx', frame, now // 1000))
+            libsafety_py.libsafety.safety_tick()
+            cs = ci.update([(now, frames)])
+            self.assertTrue(cs.canValid)
+            self.assertEqual(cs.gearShifter, structs.CarState.GearShifter.manumatic if manual else structs.CarState.GearShifter.low)
+            cs.vCruise = cs.vCruiseCluster = 100.
+            messages = []
+
+            def message(service, size=None, now=now, messages=messages):
+              msg = messaging.new_message(service, size, valid=True, logMonoTime=now)
+              messages.append(msg)
+              return getattr(msg, service)
+
+            state = message('carState')
+            state.from_dict(cs.to_dict())
+            parameters = message('vehicleParameters')
+            parameters.stiffnessFactor, parameters.steerRatio = 1., cp.steerRatio
+            message('longitudinalPlan').aTarget = 1.
+            message('modelV2').action.desiredCurvature = .00001
+            message('lateralDelay').lateralDelay = .1
+            message('carOutput')
+            peers = []
+            for service in sd.sm.data:
+              if service in ('alertDebug', 'lateralManeuverPlan', 'laneChangeAssistWire'):
+                continue
+              size = 1 if service == 'pandaStates' else 0 if service == 'laneChangeAssistWire' else None
+              peer = messaging.new_message(service, size,
+                                           valid=True, logMonoTime=now)
+              payload = getattr(peer, service)
+              if service == 'pandaStates':
+                payload[0].safetyModel = cp.safetyConfigs[0].safetyModel
+                payload[0].safetyParam = cp.safetyConfigs[0].safetyParam
+                payload[0].alternativeExperience = cp.alternativeExperience
+                payload[0].controlsAllowed = bool(libsafety_py.libsafety.get_controls_allowed())
+                payload[0].safetyRxChecksInvalid = not bool(libsafety_py.libsafety.safety_config_valid())
+              elif service == 'extrinsicsCalibration':
+                payload.calStatus = 'calibrated'
+                payload.rpyCalib = [0., 0., 0.]
+              elif service == 'vehicleParameters':
+                payload.stiffnessFactor, payload.steerRatio, payload.valid = 1., cp.steerRatio, True
+              elif service == 'deviceMotion':
+                payload.inputsOK = payload.posenetOK = True
+                payload.velocityDevice.x = cs.vEgo
+              elif service == 'controlsState':
+                payload.lateralControlState.init('torqueState')
+              elif service == 'carControl':
+                payload.from_dict(previous_control.to_dict())
+              elif service == 'deviceState':
+                payload.freeSpacePercent = 50.
+              peers.append(peer.as_reader())
+            car_state_event = messages[0].as_reader()
+            with patch('openpilot.selfdrive.selfdrived.selfdrived.REPLAY', False), \
+                 patch('openpilot.selfdrive.selfdrived.selfdrived.messaging.recv_one', return_value=car_state_event), \
+                 patch.object(sd.sm, 'update', side_effect=lambda timeout, now=now, peers=peers, sd=sd:
+                              sd.sm.update_msgs(now / 1e9, peers)):
+              sd.step()
+            self.assertEqual(sd.mismatch_counter, int(tick == 340), tick)
+            self.assertNotIn(EventName.controlsMismatch, sd.events.names)
+            host = message('selfdriveState')
+            host.enabled, host.active, host.state = sd.enabled, sd.active, sd.state_machine.state
+            event_message = messaging.new_message('onroadEvents', len(sd.events.to_msg()), valid=True, logMonoTime=now)
+            event_message.onroadEvents = sd.events.to_msg()
+            messages.append(event_message)
+            controls.sm.update_msgs(now / 1e9, [msg.as_reader() for msg in messages])
+            cc, lateral_log = controls.state_control()
+            self.assertEqual(cc.enabled, sd.enabled)
+            controls.publish(cc, lateral_log)
+            self.assertEqual(cc.cruiseControl.cancel, cs.cruiseState.enabled and (not cc.enabled or not cp.pcmCruise))
+            _, commands = ci.apply(cc.as_reader(), now + 1_000_000)
+            for frame in commands:
+              self.assertTrue(native('tx', frame, (now + 1_000_000) // 1000), (tick, hex(frame[0])))
+              if manual and frame[0] == 0x180:
+                self.assertEqual(((frame[1][0] & 7) << 8) | frame[1][1], 0)
+            if 95 <= tick < 340 or tick >= 395:
+              self.assertTrue(cc.enabled and cc.latActive, tick)
+              self.assertTrue(libsafety_py.libsafety.get_controls_allowed(), tick)
+              active_frames += 1
+            if manual or 360 <= tick < 390:
+              self.assertFalse(cc.enabled or cc.latActive or cc.longActive, tick)
+            if manual:
+              self.assertFalse(libsafety_py.libsafety.get_controls_allowed(), tick)
+              self.assertEqual(cc.actuators.torque, 0.)
+            previous_control = cc
+          self.assertGreater(active_frames, 200)

tools/ci/run_vehicle_tests.py

@@ -79,6 +79,8 @@ RELEASE_TARGETS = (
   "opendbc_repo/opendbc/safety/tests/test_gm_bolt_cc.py",
   "opendbc_repo/opendbc/car/gm/tests/test_bolt_cc_low.py",
   "openpilot/starpilot/tests/test_bolt_cc_controls_low.py",
+  "openpilot/starpilot/tests/test_bolt_cc_controls_manual.py",
+  "openpilot/starpilot/car/gm/tests/test_gear_events.py",
   "openpilot/starpilot/tests/test_bolt_disable_active_pedal.py::TestBoltDisableActivePedal",
   "opendbc_repo/opendbc/car/gm/tests/test_bolt_acc_cc.py",
   "opendbc_repo/opendbc/car/gm/tests/test_bolt_factory_acc.py",