From 942a562a0aec101a4538e83eed302914dfb75bc8 Mon Sep 17 00:00:00 2001 From: firestar5683 <168790843+firestar5683@users.noreply.github.com> Date: Sun, 4 Oct 2026 22:31:03 -0500 Subject: [PATCH] Bolt Gear State --- openpilot/selfdrive/car/car_events.py | 9 +- openpilot/selfdrive/selfdrived/events.py | 13 +- openpilot/selfdrive/selfdrived/selfdrived.py | 3 +- openpilot/starpilot/car/events.py | 12 ++ openpilot/starpilot/car/gm/events.py | 22 +++ .../car/gm/tests/test_gear_events.py | 133 ++++++++++++++++ .../tests/test_bolt_cc_controls_manual.py | 143 ++++++++++++++++++ tools/ci/run_vehicle_tests.py | 2 + 8 files changed, 327 insertions(+), 10 deletions(-) create mode 100644 openpilot/starpilot/car/events.py create mode 100644 openpilot/starpilot/car/gm/events.py create mode 100644 openpilot/starpilot/car/gm/tests/test_gear_events.py create mode 100644 openpilot/starpilot/tests/test_bolt_cc_controls_manual.py diff --git a/openpilot/selfdrive/car/car_events.py b/openpilot/selfdrive/car/car_events.py index 991e1abdb8..f3f397ad24 100644 --- a/openpilot/selfdrive/car/car_events.py +++ b/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) diff --git a/openpilot/selfdrive/selfdrived/events.py b/openpilot/selfdrive/selfdrived/events.py index 2178a6da3f..77595ec7c5 100755 --- a/openpilot/selfdrive/selfdrived/events.py +++ b/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 diff --git a/openpilot/selfdrive/selfdrived/selfdrived.py b/openpilot/selfdrive/selfdrived/selfdrived.py index 8d791443e0..0be53b229a 100755 --- a/openpilot/selfdrive/selfdrived/selfdrived.py +++ b/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 diff --git a/openpilot/starpilot/car/events.py b/openpilot/starpilot/car/events.py new file mode 100644 index 0000000000..7ce6725a73 --- /dev/null +++ b/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 diff --git a/openpilot/starpilot/car/gm/events.py b/openpilot/starpilot/car/gm/events.py new file mode 100644 index 0000000000..5c708b271f --- /dev/null +++ b/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) diff --git a/openpilot/starpilot/car/gm/tests/test_gear_events.py b/openpilot/starpilot/car/gm/tests/test_gear_events.py new file mode 100644 index 0000000000..09b852c14a --- /dev/null +++ b/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] diff --git a/openpilot/starpilot/tests/test_bolt_cc_controls_manual.py b/openpilot/starpilot/tests/test_bolt_cc_controls_manual.py new file mode 100644 index 0000000000..3211c38f81 --- /dev/null +++ b/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) diff --git a/tools/ci/run_vehicle_tests.py b/tools/ci/run_vehicle_tests.py index 61ff4b149c..b91fad69c5 100644 --- a/tools/ci/run_vehicle_tests.py +++ b/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",