From e6aea9f9fa6cdc9dae0c57b63ddda87ed5853324 Mon Sep 17 00:00:00 2001 From: firestar5683 <168790843+firestar5683@users.noreply.github.com> Date: Fri, 25 Sep 2026 14:53:18 -0500 Subject: [PATCH] Bolt Cruise Implement Bolt cruise control with its own host and native admission gates. --- opendbc_repo/opendbc/car/gm/bolt_cc.py | 263 +++++++++++++++ .../opendbc/car/gm/tests/test_bolt_cc.py | 269 ++++++++++++++++ .../opendbc/safety/modes/gm_bolt_cc.h | 279 ++++++++++++++++ .../opendbc/safety/modes/gm_bolt_cc_gate.h | 169 ++++++++++ .../opendbc/safety/tests/test_gm_bolt_cc.py | 302 ++++++++++++++++++ 5 files changed, 1282 insertions(+) create mode 100644 opendbc_repo/opendbc/car/gm/bolt_cc.py create mode 100644 opendbc_repo/opendbc/car/gm/tests/test_bolt_cc.py create mode 100644 opendbc_repo/opendbc/safety/modes/gm_bolt_cc.h create mode 100644 opendbc_repo/opendbc/safety/modes/gm_bolt_cc_gate.h create mode 100644 opendbc_repo/opendbc/safety/tests/test_gm_bolt_cc.py diff --git a/opendbc_repo/opendbc/car/gm/bolt_cc.py b/opendbc_repo/opendbc/car/gm/bolt_cc.py new file mode 100644 index 0000000000..92c3652069 --- /dev/null +++ b/opendbc_repo/opendbc/car/gm/bolt_cc.py @@ -0,0 +1,263 @@ +"""Cruise button ownership for the exact Bolt profiles.""" + +from dataclasses import dataclass +import math + +IDS = ('CHEVROLET_BOLT_CC_2017', 'CHEVROLET_BOLT_CC_2018_2021', 'CHEVROLET_BOLT_CC_2022_2023') +ACC_ID = 'CHEVROLET_BOLT_ACC_2022_2023_PEDAL' +MIN_SPEED = 24 * 0.44704 + + +@dataclass(frozen=True) +class BoltCcProfile: + identity: str + camera_removed: bool = False + + def __post_init__(self): + if self.identity not in (*IDS, ACC_ID): + raise ValueError('not a supported Bolt CC identity') + + @property + def adaptive_stock(self): + return self.identity == ACC_ID + + @property + def camera_required(self): + return self.adaptive_stock and not self.camera_removed + + +def button_bytes(button, counter): + if button not in (1, 2, 3, 6) or counter not in range(4): + raise ValueError('invalid button tuple') + checksum = 255 + counter * 0x4EF - (button - 1) * 16 + return bytes((0, 0, 0, 1, counter, (button << 4) | ((checksum >> 8) & 15), checksum & 255)) + + +class BoltCcOwner: + def __init__(self, profile): + self.profile = profile + self.frame = 0 + self.last_send = 0 + self.last_direction = 0 + self.last_direction_frame = -151 + self.pending = 0 + self.pending_frame = 0 + self.credit = None + self.last_counter = None + self.last_button_stamp = 0 + self.cruise_active = False + self.cruise_stamp = 0 + self.camera_active = False + self.camera_stamp = 0 + self.prev_enabled = False + self.cancel_pending = False + self.cancel_stamp = 0 + + def cruise(self, raw, stamp): + if len(raw) != 8 or stamp <= 0 or stamp <= self.cruise_stamp: + return False + self.cruise_active = bool(raw[4] & 128) + self.cruise_stamp = stamp + if not self.cruise_active: + self.cancel_pending = False + return True + + def camera(self, raw, stamp): + if not self.profile.camera_required or len(raw) != 6 or stamp <= 0 or stamp <= self.camera_stamp: + return False + self.camera_active = bool(raw[2] & 128) + self.camera_stamp = stamp + if not self.camera_active: + self.cancel_pending = False + return True + + def button(self, raw, stamp): + self.credit = None + if len(raw) != 7 or stamp <= 0 or stamp <= self.last_button_stamp: + return False + self.last_button_stamp = stamp + counter = raw[4] + button = (raw[5] >> 4) & 7 + if counter not in range(4) or button not in (1, 2, 3, 6) or raw != button_bytes(button, counter): + return False + forward = self.last_counter is None or counter == (self.last_counter + 1) % 4 + self.last_counter = counter + if not forward or button != 1: + return False + self.credit = (counter, stamp) + return True + + def request(self, speed, stock_speed, accel, metric=False): + if not all(math.isfinite(x) for x in (speed, stock_speed, accel)): + return 0, math.inf + convert = 3.6 if metric else 1 / 0.44704 + setpoint = int(round(stock_speed * convert)) + projected = (speed * 1.01 + 3 * accel) * convert + desired = int(round(projected)) + deadband = 0.0 if self.profile.adaptive_stock else 0.75 * (1.609344 if metric else 1.0) + comparison = desired if self.profile.adaptive_stock else projected + rate = 1.0 if abs(accel) <= 0.15 else 0.2 + if MIN_SPEED - desired / convert > 3.25: + button = 6 + elif comparison < setpoint - deadband and setpoint > MIN_SPEED * convert + 1: + button = 3 + elif comparison > setpoint + deadband: + button = 2 + else: + button = 0 + if self.profile.adaptive_stock: + return button, rate + if button not in (2, 3): + self.pending = 0 + return button, rate + if self.last_direction == 3 and button == 2 and (self.frame - self.last_direction_frame) * 0.01 <= 1.5: + if self.pending != button: + self.pending = button + self.pending_frame = self.frame + return 0, rate + if (self.frame - self.pending_frame) * 0.01 < 0.6: + return 0, rate + self.pending = 0 + return button, rate + + def update( + self, + now, + speed, + stock_speed, + accel, + *, + enabled=True, + long_active=True, + drive=True, + brake=False, + gas=False, + sources_current=True, + metric=False, + hud_speed=None, + ): + self.frame += 1 + stock_active = self.camera_active if self.profile.camera_required else self.cruise_active + if self.prev_enabled and not enabled and stock_active: + self.cancel_pending = True + self.cancel_stamp = now + if self.cancel_pending and not 0 <= now - self.cancel_stamp <= 100_000_000: + self.cancel_pending = False + if enabled: + self.cancel_pending = False + self.prev_enabled = enabled + stock_ready = ( + self.credit is not None + and self.cruise_active + and self.cruise_stamp > 0 + and 0 <= now - self.cruise_stamp <= 300_000_000 + and 0 < now - self.credit[1] <= 100_000_000 + and sources_current + ) + camera_ready = (not self.profile.camera_required or + (self.camera_active and self.camera_stamp > 0 and 0 <= now - self.camera_stamp <= 300_000_000)) + cancel_ready = (stock_ready if not self.profile.camera_required else + camera_ready and self.credit is not None and 0 < now - self.credit[1] <= 100_000_000 and sources_current) + if self.cancel_pending and cancel_ready: + counter, stamp = self.credit + self.credit = None + self.cancel_pending = False + return (0x1E1, button_bytes(6, (counter + 1) % 4), 2 if self.profile.camera_required else 0) + stock_ready = stock_ready and camera_ready + helper_requested = long_active and speed >= MIN_SPEED + if ( + not helper_requested + and stock_ready + and enabled + and drive + and not brake + and gas + and self.frame % 52 == 0 + and hud_speed is not None + and all(math.isfinite(v) for v in (stock_speed, speed, hud_speed)) + and stock_speed < speed < hud_speed + ): + counter, stamp = self.credit + self.credit = None + self.last_send = self.frame + return (0x1E1, button_bytes(3, (counter + 1) % 4), 0) + valid = stock_ready and drive and not brake and not gas and enabled and speed >= MIN_SPEED + if not valid or not long_active: + self.credit = None + self.last_send = self.frame + return None + if self.frame % 4: + return None + button, rate = self.request(speed, stock_speed, accel, metric) + if not button or (self.frame - self.last_send) * 0.01 <= rate: + return None + # Helper CANCEL uses projected speed only after the actual-speed caller floor. + if button != 6 and speed < MIN_SPEED: + return None + counter, stamp = self.credit + self.credit = None + self.last_send = self.frame + if button in (2, 3): + self.last_direction = button + self.last_direction_frame = self.frame + return (0x1E1, button_bytes(button, (counter + 1) % 4), 0) + + +def pscm_passthrough(data): + if len(data) != 8: + raise ValueError('invalid PSCM length') + masks = (0x3F, 0xFF, 0x3F, 0xFF, 0x7B, 0xFF, 7, 0xFF) + out = bytearray(d & m for d, m in zip(data, masks, strict=True)) + mod = 32 if not (data[2] & 32) else 0 + checksum = (((data[4] & 3) << 8) | data[5]) + mod + out[2] |= 32 + out[4] = (out[4] & 0xFC) | ((checksum >> 8) & 3) + out[5] = checksum & 255 + return bytes(out) + + +def auxiliary_messages(owner, pscm, source_stamp, now): + result = [] + if owner.frame % 10 == 0 and source_stamp > 0 and 0 <= now - source_stamp <= 100_000_000: + result.append((0x184, pscm_passthrough(pscm), 2)) + if owner.profile.camera_removed and owner.frame % 100 == 0: + result.extend([(0x409, bytes(7), 0), (0x40A, bytes(7), 0)]) + return result + + +class BoltCcLongitudinalPolicy: + """Literal conventional-Bolt PID calibration; no Volt stop/regen shaping.""" + + friction_variant = False + stopping_decel_rate = 11.18 + kp = ((10.7, 10.8, 28.0), (0.0, 5.0, 2.0)) + + def reset(self): + pass + + def prepare_pid(self, pid, target, error, speed, last_output, accel_limits, **kwargs): + if pid.i > 0.0 and target < -0.05 and error < -0.25 and not (speed <= 0.35 and target > -0.40): + magnitude = abs(error) + if magnitude < 0.75: + bleed = 0.55 - 0.30 * (magnitude - 0.25) / 0.50 + elif magnitude < 1.5: + bleed = 0.25 * (1.5 - magnitude) / 0.75 + else: + bleed = 0.0 + pid.i *= bleed + return False + + def feedforward(self, target, speed, last_output): + return target + + def shape_output(self, output, target, error, speed): + if output > 0.0 and target < -0.10 and error < -0.35 and not (speed <= 0.35 and target > -0.40): + positive_cap = 0.0 if target <= -0.6 else 0.05 * (target + 0.6) / 0.5 + output = min(output, positive_cap) + return output + + +def policy_for(cp): + from opendbc.car.gm.values import is_bolt_cc_profile + + return BoltCcLongitudinalPolicy() if is_bolt_cc_profile(cp) and cp.openpilotLongitudinalControl else None diff --git a/opendbc_repo/opendbc/car/gm/tests/test_bolt_cc.py b/opendbc_repo/opendbc/car/gm/tests/test_bolt_cc.py new file mode 100644 index 0000000000..6ba524caf3 --- /dev/null +++ b/opendbc_repo/opendbc/car/gm/tests/test_bolt_cc.py @@ -0,0 +1,269 @@ +from unittest.mock import patch +import unittest +from opendbc.can import CANPacker +from opendbc.car import Bus, gen_empty_fingerprint, structs +from opendbc.car.gm.interface import CarInterface +from opendbc.car.gm.values import CAR, DBC, BOLT_CC_WORDS, GMFlags, is_bolt_cc_profile +from opendbc.car.gm.bolt_cc import button_bytes +from opendbc.safety.tests.libsafety import libsafety_py + +BOLT_CC_WORDS = {identity: words for identity, words in BOLT_CC_WORDS.items() if identity != CAR.CHEVROLET_BOLT_ACC_2022_2023_PEDAL} + + +def setup(cp): + safety = libsafety_py.libsafety + safety.init_tests() + assert safety.set_safety_hooks(int(structs.CarParams.SafetyModel.gm), cp.safetyConfigs[0].safetyParam) == 0 + + +def native(direction, msg, us): + address, data, bus = msg + safety = libsafety_py.libsafety + safety.set_timer(us) + packet = libsafety_py.make_CANPacket(address, bus, data) + return safety.safety_rx_hook(packet) if direction == "rx" else safety.safety_tx_hook(packet) + + +class Settings: + def __init__(self, pedal=False, metric=False): + self.settings = {'GMPedalLongitudinal': pedal, 'IsMetric': metric} + + def get_bool(self, key): + return self.settings.get(key, False) + + +def params(identity, alpha=False, present=False, pedal=False, removed=False, metric=False): + fp = gen_empty_fingerprint() + if present: + fp[0][0x201] = 6 + if not removed: + fp[2][0x180] = 4 + with patch('opendbc.car.gm.interface.Params', return_value=Settings(pedal, metric)): + return CarInterface.get_params(identity, fp, [], alpha, False, False) + + +def fixture(identity, removed=False, metric=False, alpha=False, present=False): + cp = params(identity, removed=removed, metric=metric, alpha=alpha, present=present) + ci = CarInterface(cp) + ci.CC.bolt_cc_metric = metric + return cp, ci, CANPacker(DBC[identity][Bus.pt]) + + +def feed(ci, packer, now, counter=0, gas=False, active=True, speed=20, stock=15, brake=False, regen=False): + vals = { + 'PSCMStatus': {}, + 'ECMCruiseControl': {'CruiseActive': int(active), 'CruiseSetSpeed': stock * 3.6}, + 'ECMEngineStatus': {'CruiseMainOn': 1, 'BrakePressed': int(brake)}, + 'AcceleratorPedal2': {'AcceleratorPedal2': 30 if gas else 0}, + 'ECMPRDNL2': {'PRNDL2': 4}, + 'EBCMWheelSpdRear': {'RLWheelSpd': speed * 3.6, 'RRWheelSpd': speed * 3.6, 'RLWheelDir': 1, 'RRWheelDir': 1}, + 'EBCMWheelSpdFront': {'FLWheelSpd': speed * 3.6, 'FRWheelSpd': speed * 3.6}, + 'ESPStatus': {'TractionControlOn': 1}, + 'BCMDoorBeltStatus': {'LeftSeatBelt': 1}, + 'BCMGeneralPlatformStatus': {}, + 'BCMTurnSignals': {}, + 'PSCMSteeringAngle': {}, + 'ECMAcceleratorPos': {}, + 'EBCMFrictionBrakeStatus': {}, + 'EBCMRegenPaddle': {'RegenPaddle': int(regen)}, + } + msgs = [packer.make_can_msg(n, 0, v) for n, v in vals.items()] + msgs.append((0x1E1, button_bytes(1, counter), 0)) + msgs.append(packer.make_can_msg('ASCMLKASteeringCmd', 2, {})) + msgs.append(packer.make_can_msg('AEBCmd', 2, {})) + # Prime lazy subscriptions, then prove actual CI's validity (no fabricated canValid). + ci.update([(now - 1_000_000, msgs)]) + out = ci.update([(now, msgs)]) + return out, sorted(msgs, key=lambda m: m[0] == 0x3d1) + + +def control(long_active=True, enabled=True, gas=False): + cc = structs.CarControl() + cc.enabled = enabled + cc.latActive = True + cc.longActive = long_active + cc.actuators.accel = 1 + cc.actuators.torque = 0.02 + cc.hudControl.setSpeed = 30 + return cc + + +class Production(unittest.TestCase): + def test_final_params_matrix_and_pedal_separation(self): + for identity, words in BOLT_CC_WORDS.items(): + for alpha in (False, True): + for present in (False, True): + for setting in (False, True): + for removed in (False, True): + cp = params(identity, alpha=alpha, present=present, pedal=setting, removed=removed) + if present and setting: + self.assertFalse(is_bolt_cc_profile(cp)) + self.assertTrue(cp.flags & GMFlags.PEDAL_LONG.value) + else: + self.assertTrue(is_bolt_cc_profile(cp)) + self.assertEqual(cp.safetyConfigs[0].safetyParam, words[removed]) + self.assertTrue(cp.openpilotLongitudinalControl) + self.assertFalse(cp.pcmCruise or cp.alphaLongitudinalAvailable) + self.assertAlmostEqual(cp.minEnableSpeed, 24 * 0.44704, places=5) + + def test_exact_profile_rejects_wrong_model_extra_configs_and_side_owner(self): + cp = params(CAR.CHEVROLET_BOLT_CC_2017) + for field, value in (("passive", True), ("dashcamOnly", True), ("notCar", True), + ("pcmCruise", True), ("alphaLongitudinalAvailable", True), + ("networkLocation", structs.CarParams.NetworkLocation.gateway)): + altered = cp.as_reader().as_builder() + setattr(altered, field, value) + self.assertFalse(is_bolt_cc_profile(altered), field) + altered = cp.as_reader().as_builder() + altered.safetyConfigs[0].safetyModel = structs.CarParams.SafetyModel.toyota + self.assertFalse(is_bolt_cc_profile(altered)) + altered = cp.as_reader().as_builder() + altered.safetyConfigs = [cp.safetyConfigs[0].to_dict(), cp.safetyConfigs[0].to_dict()] + self.assertFalse(is_bolt_cc_profile(altered)) + + def test_actual_parser_controller_native_all_generation_routes(self): + for identity in BOLT_CC_WORDS: + for removed in (False, True): + for alpha, present in ((False, False), (False, True), (True, False), (True, True)): + cp, ci, packer = fixture(identity, removed, alpha=alpha, present=present) + out, msgs = feed(ci, packer, 1_000_000_000) + self.assertTrue(out.canValid) + self.assertFalse(out.canTimeout) + self.assertTrue(out.cruiseState.enabled) + self.assertFalse(out.cruiseState.nonAdaptive) + self.assertAlmostEqual(out.cruiseState.speed, 15, places=2) + setup(cp) + for msg in msgs: + if msg[0] in (0x184, 0x3D1, 0x1E1, 0xC9, 0x1C4, 0x1F5, 0x34A): + self.assertTrue(native('rx', msg, 1_000_000)) + ci.CC.frame = 104 + _, commands = ci.apply(control().as_reader(), 1_001_000_000) + button = next(m for m in commands if m[0] == 0x1E1) + self.assertEqual(button[1], button_bytes(2, 1)) + self.assertTrue(native('tx', button, 1_001_000)) + self.assertFalse(any(m[0] in (0x200, 0x315, 0x2CB, 0xBD, 0x3D1) for m in commands)) + + def test_gas_preserves_lateral_and_set_exception_then_release(self): + for identity in BOLT_CC_WORDS: + cp, ci, packer = fixture(identity) + out, msgs = feed(ci, packer, 1_000_000_000, gas=True) + self.assertTrue(out.gasPressed) + setup(cp) + for msg in msgs: + if msg[0] in (0x184, 0x3D1, 0x1E1, 0xC9, 0x1C4, 0x1F5, 0x34A): + native('rx', msg, 1_000_000) + self.assertTrue(libsafety_py.libsafety.get_controls_allowed()) + ci.CC.frame = 104 + _, commands = ci.apply(control(False).as_reader(), 1_001_000_000) + button = next(m for m in commands if m[0] == 0x1E1) + self.assertEqual(button[1], button_bytes(3, 1)) + self.assertTrue(native('tx', button, 1_001_000)) + steer = next(m for m in commands if m[0] == 0x180) + self.assertTrue(native('tx', steer, 1_001_001)) + self.assertFalse(native('tx', (0x1E1, button_bytes(2, 1), 0), 1_001_002)) + # Release gas, still-active PCM; a new physical neutral counter restores ordinary long. + out, msgs = feed(ci, packer, 1_010_000_000, counter=1, gas=False) + for msg in msgs: + if msg[0] in (0x184, 0x3D1, 0x1E1, 0xC9, 0x1C4, 0x1F5, 0x34A): + native('rx', msg, 1_010_000) + self.assertTrue(libsafety_py.libsafety.get_controls_allowed()) + ci.CC.frame = 128 + _, commands = ci.apply(control().as_reader(), 1_011_000_000) + button = next(m for m in commands if m[0] == 0x1E1) + self.assertEqual(button[1], button_bytes(2, 2)) + self.assertTrue(native('tx', button, 1_011_000)) + + def test_actual_controller_metric_boundary_and_disable_long(self): + identity = CAR.CHEVROLET_BOLT_CC_2018_2021 + for metric in (False, True): + cp, ci, packer = fixture(identity, metric=metric) + out, _ = feed(ci, packer, 1_000_000_000, speed=12, stock=12) + self.assertTrue(out.canValid) + ci.CC.frame = 104 + cc = control() + cc.actuators.accel = -0.17 + _, commands = ci.apply(cc.as_reader(), 1_001_000_000) + buttons = [m for m in commands if m[0] == 0x1E1] + self.assertEqual(len(buttons), 0 if metric else 1) + if buttons: + self.assertEqual(buttons[0][1], button_bytes(3, 1)) + cp = params(identity) + cp.openpilotLongitudinalControl = False + ci = CarInterface(cp) + packer = CANPacker(DBC[identity][Bus.pt]) + out, _ = feed(ci, packer, 1_000_000_000) + self.assertTrue(out.canValid) + ci.CC.frame = 104 + _, commands = ci.apply(control().as_reader(), 1_001_000_000) + self.assertFalse(any(m[0] in (0x1E1, 0x200, 0x315, 0x2CB, 0xBD) for m in commands)) + + def test_real_paddle_disengagement_and_physical_button_rearm(self): + cp, ci, packer = fixture(CAR.CHEVROLET_BOLT_CC_2017) + out, msgs = feed(ci, packer, 1_000_000_000, regen=True) + self.assertTrue(out.regenBraking) + setup(cp) + for msg in msgs: + if msg[0] in (0x184, 0x3D1, 0x1E1, 0xC9, 0x1C4, 0x1F5, 0x34A, 0xBD): + native('rx', msg, 1_000_000) + self.assertFalse(libsafety_py.libsafety.get_controls_allowed()) + ci.CC.frame = 104 + _, commands = ci.apply(control().as_reader(), 1_001_000_000) + self.assertFalse(any(m[0] == 0x1E1 for m in commands)) + self.assertFalse(native('tx', (0x1E1, button_bytes(2, 1), 0), 1_001_000)) + self.assertTrue(native('rx', packer.make_can_msg('EBCMRegenPaddle', 0, {}), 1_002_000)) + self.assertFalse(libsafety_py.libsafety.get_controls_allowed()) + self.assertTrue(native('rx', (0x1E1, button_bytes(3, 1), 0), 1_003_000)) + self.assertFalse(libsafety_py.libsafety.get_controls_allowed()) + self.assertTrue(native('rx', (0x1E1, button_bytes(1, 2), 0), 1_004_000)) + self.assertTrue(libsafety_py.libsafety.get_controls_allowed()) + self.assertTrue(native('tx', (0x1E1, button_bytes(2, 3), 0), 1_005_000)) + + def test_wrong_length_sources_cannot_create_fresh_raw_credit(self): + for bad_length in (3, 9): + cp, ci, packer = fixture(CAR.CHEVROLET_BOLT_CC_2018_2021) + out, _ = feed(ci, packer, 1_000_000_000) + self.assertTrue(out.canValid) + bad = (0x1E1, bytes(bad_length), 0) + good = (0x1E1, button_bytes(1, 1), 0) + out = ci.update([(1_010_000_000, [bad, good])]) + self.assertFalse(out.canValid) + self.assertEqual(ci.CS.bolt_cc_sources[2], (0, b"")) + ci.CC.frame = 104 + _, commands = ci.apply(control().as_reader(), 1_011_000_000) + self.assertFalse(any(m[0] == 0x1E1 for m in commands)) + ci.update([(1_012_000_000, [good])]) + self.assertEqual(ci.CS.bolt_cc_sources[2], (1_012_000_000, good[1])) + + def test_invalid_optional_paddle_source_stays_not_ready_until_new_valid_packet(self): + cp, ci, packer = fixture(CAR.CHEVROLET_BOLT_CC_2018_2021) + out, _ = feed(ci, packer, 1_000_000_000) + self.assertTrue(out.canValid) + out = ci.update([(1_010_000_000, [(0xbd, bytes(3), 0)])]) + self.assertFalse(out.canValid) + out = ci.update([(1_011_000_000, [])]) + self.assertFalse(out.canValid) + out = ci.update([(1_012_000_000, [(0xbd, bytes(7), 0)])]) + self.assertTrue(out.canValid) + + def test_global_health_revocation_does_not_auto_rearm_on_fresh_sources(self): + cp, ci, packer = fixture(CAR.CHEVROLET_BOLT_CC_2018_2021) + _, msgs = feed(ci, packer, 1_000_000_000) + setup(cp) + for msg in msgs: + if msg[0] in (0x184, 0x3D1, 0x1E1, 0xC9, 0x1C4, 0x1F5, 0x34A, 0xBD): + native('rx', msg, 1_000_000) + self.assertTrue(libsafety_py.libsafety.get_controls_allowed()) + libsafety_py.libsafety.set_timer(3_000_000) + libsafety_py.libsafety.safety_tick_current_safety_config() + self.assertFalse(libsafety_py.libsafety.get_controls_allowed()) + _, msgs = feed(ci, packer, 3_001_000_000, counter=1) + for msg in msgs: + if msg[0] in (0x184, 0x3D1, 0x1E1, 0xC9, 0x1C4, 0x1F5, 0x34A, 0xBD): + native('rx', msg, 3_001_000) + self.assertFalse(libsafety_py.libsafety.get_controls_allowed()) + self.assertTrue(native('rx', (0x1E1, button_bytes(2, 2), 0), 3_002_000)) + self.assertTrue(libsafety_py.libsafety.get_controls_allowed()) + + +if __name__ == '__main__': + unittest.main() diff --git a/opendbc_repo/opendbc/safety/modes/gm_bolt_cc.h b/opendbc_repo/opendbc/safety/modes/gm_bolt_cc.h new file mode 100644 index 0000000000..fc629a5a63 --- /dev/null +++ b/opendbc_repo/opendbc/safety/modes/gm_bolt_cc.h @@ -0,0 +1,279 @@ +#pragma once + +#include "gm_bolt_cc_gate.h" + +static void gm_rx_hook(const CANPacket_t *msg); +static bool gm_fwd_hook(int bus_num, int addr); +static bool gm_aol_mode_valid(void); + +static GmBoltCcGate gm_bolt_cc_state; +static bool gm_bolt_cc_removed; +static bool gm_bolt_cc_stock_only; +static bool gm_bolt_cc_stock_main; +static unsigned int gm_bolt_cc_generation; +static uint8_t gm_bolt_cc_pscm[8]; +static uint64_t gm_bolt_cc_pscm_tx_ns; +static uint32_t gm_bolt_cc_clock_prev; +static uint64_t gm_bolt_cc_clock_epoch; + +static uint64_t gm_bolt_cc_now_ns(void) { + const uint32_t us = microsecond_timer_get(); + if (us < gm_bolt_cc_clock_prev) { + gm_bolt_cc_clock_epoch += 1ULL << 32U; + } + gm_bolt_cc_clock_prev = us; + return (gm_bolt_cc_clock_epoch + us) * 1000ULL; +} + +static void gm_bolt_cc_profile_rx(const CANPacket_t *msg) { + const uint64_t now = gm_bolt_cc_now_ns(); + // Shared safety revocation remains latched until a physical rearm edge. + if (!controls_allowed && gm_bolt_cc_state.armed) { + gm_bolt_cc_state.armed = false; + } + if (gm_bolt_cc_receive(&gm_bolt_cc_state, msg->addr, msg->bus, msg->data, GET_LEN(msg), now)) { + if (msg->bus == 0U) { + if (msg->addr == 0x184U) { + for (unsigned int i = 0U; i < 8U; i++) { + gm_bolt_cc_pscm[i] = msg->data[i]; + } + const unsigned int torque_raw = (((unsigned int)msg->data[6] & 7U) << 8U) | (unsigned int)msg->data[7]; + update_sample(&torque_driver, to_signed((int)torque_raw, 11)); + } + if (msg->addr == 0x34AU) { + vehicle_moving = gm_bolt_cc_state.speed > (0.311F / 3.6F); + } + if (gm_bolt_cc_stock_only && (msg->addr == 0xC9U)) { + gm_bolt_cc_stock_main = GET_BIT(msg, 29U); + if (!gm_bolt_cc_stock_main) { gm_bolt_cc_state.credit = false; } + } + brake_pressed = gm_bolt_cc_state.brake; + gas_pressed = gm_bolt_cc_state.gas; + regen_braking = gm_bolt_cc_state.regen; + cruise_engaged_prev = gm_bolt_cc_state.active; + controls_allowed = gm_bolt_cc_state.armed && gm_bolt_cc_state.active; + } + } +} + +static bool gm_bolt_cc_profile_tx(const CANPacket_t *msg) { + const uint64_t now = gm_bolt_cc_now_ns(); + bool fresh = true; + for (unsigned int i = 0U; i < 7U; i++) { + fresh = fresh && gm_bolt_cc_state.seen[i] && (now >= gm_bolt_cc_state.stamps[i]) && + ((now - gm_bolt_cc_state.stamps[i]) <= 300000000ULL); + } + bool allowed = false; + if (msg->addr == 0x1E1U) { + const bool stock_cancel = !gm_bolt_cc_stock_only || (gm_bolt_cc_stock_main && (((msg->data[5] >> 4U) & 7U) == 6U)); + allowed = stock_cancel && gm_bolt_cc_transmit(&gm_bolt_cc_state, msg->addr, msg->bus, msg->data, GET_LEN(msg), now, + longitudinal_controls_allowed(), get_longitudinal_allowed()); + } else if ((msg->addr == 0x180U) && (msg->bus == 0U) && (GET_LEN(msg) == 4U)) { + allowed = fresh && (!gm_aol_lateral_allowed() || gm_bolt_cc_state.drive) && gm_tx_hook(msg); + } else if ((msg->addr == 0x184U) && (msg->bus == 2U) && (GET_LEN(msg) == 8U) && (gm_bolt_cc_generation != 0U)) { + const uint8_t masks[8] = {0x3FU, 0xFFU, 0x3FU, 0xFFU, 0x7BU, 0xFFU, 7U, 0xFFU}; + uint8_t expected[8]; + for (unsigned int i = 0U; i < 8U; i++) { + expected[i] = gm_bolt_cc_pscm[i] & masks[i]; + } + unsigned int checksum = (((unsigned int)gm_bolt_cc_pscm[4] & 3U) << 8U) | (unsigned int)gm_bolt_cc_pscm[5]; + if ((gm_bolt_cc_pscm[2] & 32U) == 0U) { + checksum += 32U; + } + expected[2] = (uint8_t)((unsigned int)expected[2] | 32U); + expected[4] = (uint8_t)(((unsigned int)expected[4] & 0xFCU) | ((checksum >> 8U) & 3U)); + expected[5] = (uint8_t)(checksum & 255U); + bool payload_matches = true; + for (unsigned int i = 0U; i < 8U; i++) { + payload_matches &= expected[i] == msg->data[i]; + } + const bool valid = gm_bolt_cc_state.seen[0] && (now >= gm_bolt_cc_state.stamps[0]) && + ((now - gm_bolt_cc_state.stamps[0]) <= 100000000ULL) && + ((gm_bolt_cc_pscm_tx_ns == 0ULL) || ((now - gm_bolt_cc_pscm_tx_ns) >= 100000000ULL)) && + payload_matches; + if (valid) { + gm_bolt_cc_pscm_tx_ns = now; + } + allowed = valid; + } else if (gm_bolt_cc_removed && (gm_bolt_cc_generation != 0U) && ((msg->addr == 0x409U) || (msg->addr == 0x40AU)) && + (msg->bus == 0U) && (GET_LEN(msg) == 7U)) { + allowed = true; + for (unsigned int i = 0U; i < 7U; i++) { + allowed &= msg->data[i] == 0U; + } + } else { + allowed = false; + } + return allowed; +} + +static bool gm_bolt_cc_profile_fwd(int bus, int addr) { + // Preserve the original no-ACC camera withholding policy without a cruise spoof. + return ((bus == 2) && (addr == 0x180)) || ((bus == 0) && ((addr == 0x184) || (addr == 0x3D1))); +} + +static safety_config gm_bolt_cc_profile_init(uint16_t tag) { + const unsigned int tag_value = tag; + const unsigned int generation = ((tag_value >= 1U) && (tag_value <= 6U)) ? (((tag_value - 1U) / 2U) + 1U) : 0U; + const bool gm_bolt_cc_adaptive = (tag == 7U) || (tag == 8U); + const bool gm_bolt_cc_camera_required = tag == 7U; + gm_bolt_cc_generation = gm_bolt_cc_adaptive ? 3U : generation; + gm_bolt_cc_removed = (gm_bolt_cc_generation != 0U) && ((tag % 2U) == 0U); + for (unsigned int i = 0U; i < 8U; i++) { + gm_bolt_cc_pscm[i] = 0U; + } + gm_bolt_cc_pscm_tx_ns = 0ULL; + gm_bolt_cc_clock_prev = microsecond_timer_get(); + gm_bolt_cc_clock_epoch = 0ULL; + uint16_t base_param = 1U; + if (gm_bolt_cc_generation == 1U) { + base_param = 33U; + } + (void)gm_init(base_param); + gm_bolt_cc_reset(&gm_bolt_cc_state, gm_bolt_cc_generation); + gm_bolt_cc_state.camera_required = gm_bolt_cc_camera_required; + + static RxCheck rx[] = { + {.msg = {{0x184U, 0U, 8U, 10U, .ignore_checksum = true, .ignore_counter = true, .ignore_quality_flag = true}, {0}, {0}}}, + {.msg = {{0x3D1U, 0U, 8U, 10U, .ignore_checksum = true, .ignore_counter = true, .ignore_quality_flag = true}, {0}, {0}}}, + {.msg = {{0x1E1U, 0U, 7U, 10U, .ignore_checksum = true, .ignore_counter = true, .ignore_quality_flag = true}, {0}, {0}}}, + {.msg = {{0xC9U, 0U, 8U, 10U, .ignore_checksum = true, .ignore_counter = true, .ignore_quality_flag = true}, {0}, {0}}}, + {.msg = {{0x1C4U, 0U, 8U, 10U, .ignore_checksum = true, .ignore_counter = true, .ignore_quality_flag = true}, {0}, {0}}}, + {.msg = {{0x1F5U, 0U, 8U, 10U, .ignore_checksum = true, .ignore_counter = true, .ignore_quality_flag = true}, {0}, {0}}}, + {.msg = {{0x34AU, 0U, 5U, 10U, .ignore_checksum = true, .ignore_counter = true, .ignore_quality_flag = true}, {0}, {0}}} + }; + static RxCheck adaptive_rx[] = { + {.msg = {{0x184U, 0U, 8U, 10U, .ignore_checksum = true, .ignore_counter = true, .ignore_quality_flag = true}, {0}, {0}}}, + {.msg = {{0x3D1U, 0U, 8U, 10U, .ignore_checksum = true, .ignore_counter = true, .ignore_quality_flag = true}, {0}, {0}}}, + {.msg = {{0x1E1U, 0U, 7U, 10U, .ignore_checksum = true, .ignore_counter = true, .ignore_quality_flag = true}, {0}, {0}}}, + {.msg = {{0xC9U, 0U, 8U, 10U, .ignore_checksum = true, .ignore_counter = true, .ignore_quality_flag = true}, {0}, {0}}}, + {.msg = {{0x1C4U, 0U, 8U, 10U, .ignore_checksum = true, .ignore_counter = true, .ignore_quality_flag = true}, {0}, {0}}}, + {.msg = {{0x1F5U, 0U, 8U, 10U, .ignore_checksum = true, .ignore_counter = true, .ignore_quality_flag = true}, {0}, {0}}}, + {.msg = {{0x370U, 2U, 6U, 25U, .ignore_checksum = true, .ignore_counter = true, .ignore_quality_flag = true}, {0}, {0}}}, + {.msg = {{0x34AU, 0U, 5U, 10U, .ignore_checksum = true, .ignore_counter = true, .ignore_quality_flag = true}, {0}, {0}}} + }; + static const CanMsg tx[] = { + {0x180U, 0U, 4U, .check_relay = true}, {0x1E1U, 0U, 7U, .check_relay = false}, {0x184U, 2U, 8U, .check_relay = true} + }; + static const CanMsg removed_tx[] = { + {0x180U, 0U, 4U, .check_relay = false}, {0x1E1U, 0U, 7U, .check_relay = false}, {0x184U, 2U, 8U, .check_relay = false}, + {0x409U, 0U, 7U, .check_relay = false}, {0x40AU, 0U, 7U, .check_relay = false} + }; + static const CanMsg adaptive_tx[] = { + {0x180U, 0U, 4U, .check_relay = true}, {0x1E1U, 0U, 7U, .check_relay = false}, + {0x184U, 2U, 8U, .check_relay = true}, {0x1E1U, 2U, 7U, .check_relay = false} + }; + static const CanMsg denied_tx[] = {{0U, 0U, 0U, .check_relay = false}}; + safety_config ret = BUILD_SAFETY_CFG(rx, tx); + if (gm_bolt_cc_camera_required) { + SET_RX_CHECKS(adaptive_rx, ret); + SET_TX_MSGS(adaptive_tx, ret); + } + if (gm_bolt_cc_removed) { + SET_TX_MSGS(removed_tx, ret); + } + if (gm_bolt_cc_stock_only) { + static const CanMsg stock_tx[] = { + {0x180U, 0U, 4U, .check_relay = false}, {0x1E1U, 0U, 7U, .check_relay = false}, + {0x184U, 2U, 8U, .check_relay = false} + }; + SET_TX_MSGS(stock_tx, ret); + } + if (gm_bolt_cc_generation == 0U) { + SET_TX_MSGS(denied_tx, ret); + } + return ret; +} + +// Complete-word dispatch keeps every other GM profile on its existing hooks. + +static unsigned int gm_bolt_cc_tag(uint16_t word) { + static const uint16_t GM_BOLT_CC_WORDS[6] = {0xC110U, 0xC111U, 0xC120U, 0xC121U, 0xC130U, 0xC131U}; + + unsigned int tag = (word == 0xC140U) ? 7U : 0U; + switch (word) { + case 0xE710U: tag = 2U; break; + case 0xE711U: tag = 4U; break; + case 0xE712U: tag = 6U; break; + case 0xE713U: tag = 8U; break; + default: break; + } + if (word == 0xC141U) { + tag = 8U; + } + for (unsigned int i = 0U; i < 6U; i++) { + if (word == GM_BOLT_CC_WORDS[i]) { + tag = i + 1U; + break; + } + } + return tag; +} + +static bool gm_bolt_cc_selected; + +static safety_config gm_bolt_cc_init(uint16_t word) { + gm_hybrid_rejected = ((word & 0xFF00U) == 0xE800U) && ((word > 0xE805U) || + ((alternative_experience != 0) && (alternative_experience != 32))); + gm_hybrid_selected = (word >= 0xE800U) && (word <= 0xE805U); + safety_config ret; + if (gm_hybrid_selected) { + gm_bolt_cc_selected = false; + ret = gm_hybrid_init(word); + } else { + gm_bolt_cc_stock_main = false; + gm_bolt_cc_stock_only = (word == 0xE710U) || (word == 0xE711U) || (word == 0xE712U) || (word == 0xE713U); + const unsigned int tag = gm_bolt_cc_tag(word); + gm_bolt_cc_selected = tag != 0U; + ret = (tag != 0U) ? gm_bolt_cc_profile_init((uint16_t)tag) : gm_init(word); + gm_aol_initialize(gm_aol_profile_word(word) && gm_aol_mode_valid()); + gm_aol_stock_only = gm_bolt_cc_stock_only || (word == 16U) || (word == 0xC162U); + gm_aol_stock_gateway = word == 16U; + } + gm_cruise_status_reset(word); + return ret; +} + +static void gm_bolt_cc_rx(const CANPacket_t *msg) { + gm_cruise_status_prune(); + const bool ignore_status = (gm_cruise_status_bolt || gm_cruise_status_hybrid) && (msg->addr == 0x3D1U) && + (msg->returned || msg->rejected); + if (!ignore_status) { + if (gm_hybrid_selected) { + gm_hybrid_rx(msg); + } else if (gm_bolt_cc_selected) { + gm_bolt_cc_profile_rx(msg); + } else { + gm_rx_hook(msg); + } + gm_aol_observe(msg); + gm_cruise_status_observe(msg); + } +} + +static void gm_bolt_cc_optional_rx(const CANPacket_t *msg) { + gm_cruise_status_prune(); + if (gm_hybrid_selected) { + gm_hybrid_rx(msg); + } else { + gm_one_pedal_observe(msg); + if (gm_bolt_cc_selected && (msg->addr == 0xBDU) && (msg->bus == 0U) && (GET_LEN(msg) == 7U)) { + gm_bolt_cc_profile_rx(msg); + } + } + gm_cruise_status_observe(msg); +} + +static bool gm_bolt_cc_tx(const CANPacket_t *msg) { + gm_cruise_status_prune(); + return !gm_hybrid_rejected && + ((msg->addr == 0x3D1U) ? gm_cruise_status_tx(msg) : + (gm_hybrid_selected ? gm_hybrid_tx(msg) : (gm_bolt_cc_selected ? gm_bolt_cc_profile_tx(msg) : gm_tx_hook(msg)))); +} + +static bool gm_bolt_cc_fwd(int bus, int addr) { + bool block; + if (gm_hybrid_selected) { block = ((bus == 2) && (addr == 0x180)) || ((bus == 0) && (addr == 0x184)); } + else { block = gm_bolt_cc_selected ? gm_bolt_cc_profile_fwd(bus, addr) : gm_fwd_hook(bus, addr); } + return block || gm_cruise_status_block(bus, addr); +} diff --git a/opendbc_repo/opendbc/safety/modes/gm_bolt_cc_gate.h b/opendbc_repo/opendbc/safety/modes/gm_bolt_cc_gate.h new file mode 100644 index 0000000000..6dc12121c4 --- /dev/null +++ b/opendbc_repo/opendbc/safety/modes/gm_bolt_cc_gate.h @@ -0,0 +1,169 @@ +#pragma once + +#include +#include + +// Conventional cruise buttons only; no pedal, friction or scheduled outputs. +typedef struct { + uint8_t generation; + uint8_t counter; + uint8_t last_button; + bool credit; + bool button_seen; + bool active; + bool camera_required; + bool camera_active; + uint64_t camera_ts; + bool brake; + bool gas; + bool drive; + bool armed; + bool regen; + float speed; + uint64_t regen_ts; + uint64_t button_ts; + uint64_t cruise_ts; + bool seen[7]; + uint64_t stamps[7]; +} GmBoltCcGate; + +static void gm_bolt_cc_reset(GmBoltCcGate *gate, unsigned int generation) { + *gate = (GmBoltCcGate){0}; + gate->generation = (uint8_t)(((generation >= 1U) && (generation <= 3U)) ? generation : 0U); +} + +static bool gm_bolt_cc_button_tuple(const uint8_t *data, unsigned int button, unsigned int counter) { + bool valid = false; + if ((counter <= 3U) && ((button == 1U) || (button == 2U) || (button == 3U) || (button == 6U))) { + const unsigned int checksum = 255U + (counter * 0x4EFU) - ((button - 1U) * 16U); + valid = (data[0] == 0U) && (data[1] == 0U) && (data[2] == 0U) && (data[3] == 1U) && + (data[4] == counter) && (data[5] == ((button << 4U) | ((checksum >> 8U) & 15U))) && + (data[6] == (checksum & 255U)); + } + return valid; +} + +static bool gm_bolt_cc_receive(GmBoltCcGate *gate, uint32_t addr, unsigned int bus, const uint8_t *data, unsigned int len, uint64_t now) { + static const uint32_t GM_BOLT_CC_SOURCES[7] = {0x184U, 0x3D1U, 0x1E1U, 0xC9U, 0x1C4U, 0x1F5U, 0x34AU}; + static const unsigned int GM_BOLT_CC_LENGTHS[7] = {8U, 8U, 7U, 8U, 8U, 8U, 5U}; + + bool accepted = false; + if (gate->camera_required && (addr == 0x370U) && (bus == 2U) && (len == 6U) && (now > gate->camera_ts)) { + gate->camera_active = (data[2] & 128U) != 0U; + gate->camera_ts = now; + accepted = true; + } + if ((gate->generation != 0U) && (bus == 0U) && (now != 0ULL)) { + if ((addr == 0xBDU) && (len == 7U) && (now > gate->regen_ts)) { + gate->regen_ts = now; + gate->regen = ((unsigned int)data[0] >> 4U) != 0U; + if (gate->regen) { + gate->credit = false; + gate->armed = false; + } + accepted = true; + } else { + int index = -1; + for (unsigned int i = 0U; i < 7U; i++) { + if (addr == GM_BOLT_CC_SOURCES[i]) { + index = (int)i; + } + } + if ((index >= 0) && (len == GM_BOLT_CC_LENGTHS[index]) && (!gate->seen[index] || (now > gate->stamps[index]))) { + gate->seen[index] = true; + gate->stamps[index] = now; + + if (addr == 0x3D1U) { + const bool active = (data[4] & 128U) != 0U; + if (active && !gate->active && gate->drive && !gate->brake && !gate->regen) { + gate->armed = true; + } + gate->active = active; + gate->cruise_ts = now; + if (!gate->active) { + if (!gate->camera_required) { + gate->credit = false; + } + gate->armed = false; + } + } + if (addr == 0xC9U) { + gate->brake = (data[5] & 1U) != 0U; + if (gate->brake) { + gate->credit = false; + gate->armed = false; + } + } + if (addr == 0x1C4U) { + gate->gas = data[5] != 0U; + if (gate->gas) { + gate->credit = false; + } + } + if (addr == 0x1F5U) { + const unsigned int gear = data[3] & 15U; + gate->drive = ((gear == 4U) || (gear == 6U)) && ((data[5] & 2U) == 0U); + if (!gate->drive) { + gate->credit = false; + gate->armed = false; + } + } + if (addr == 0x34AU) { + const unsigned int left = ((unsigned int)data[0] << 8U) | (unsigned int)data[1]; + const unsigned int right = ((unsigned int)data[2] << 8U) | (unsigned int)data[3]; + gate->speed = (float)((left < right) ? left : right) * (0.0311F / 3.6F); + } + if (addr == 0x1E1U) { + gate->credit = false; + const unsigned int counter = data[4]; + const bool forward = !gate->button_seen || (counter == ((gate->counter + 1U) % 4U)); + gate->button_seen = true; + gate->counter = (uint8_t)counter; + gate->button_ts = now; + gate->credit = forward && gm_bolt_cc_button_tuple(data, 1U, counter); + const unsigned int button = ((unsigned int)data[5] >> 4U) & 7U; + const bool physical_valid = forward && gm_bolt_cc_button_tuple(data, button, counter); + if (physical_valid && gate->active && gate->drive && !gate->brake && !gate->regen && + (((button == 2U) && (gate->last_button != 2U)) || ((button == 1U) && (gate->last_button == 3U)))) { + gate->armed = true; + } + gate->last_button = (uint8_t)(physical_valid ? button : 0U); + if ((((unsigned int)data[5] >> 4U) & 7U) == 6U) { + gate->armed = false; + } + } + accepted = true; + } + } + } + return accepted; +} + +static bool gm_bolt_cc_transmit(GmBoltCcGate *gate, uint32_t addr, unsigned int bus, const uint8_t *data, unsigned int len, + uint64_t now, bool controls, bool long_active) { + bool valid = false; + if ((gate->generation != 0U) && (addr == 0x1E1U) && ((bus == 0U) || (gate->camera_required && (bus == 2U))) && (len == 7U)) { + const unsigned int button = ((unsigned int)data[5] >> 4U) & 7U; + const bool camera_ready = !gate->camera_required || (gate->camera_active && (gate->camera_ts > 0ULL) && + (now >= gate->camera_ts) && ((now - gate->camera_ts) <= 300000000ULL)); + const bool camera_cancel = gate->camera_required && (bus == 2U) && (button == 6U); + bool ready = gate->credit && camera_ready && (camera_cancel || gate->active); + const bool acceleration_owned = gate->armed && gate->drive && !gate->brake && !gate->regen && !gate->gas && controls && long_active; + for (unsigned int i = 0U; i < 7U; i++) { + ready = ready && gate->seen[i] && (now >= gate->stamps[i]) && ((now - gate->stamps[i]) <= 300000000ULL); + } + ready = ready && (now > gate->button_ts) && ((now - gate->button_ts) <= 100000000ULL); + const unsigned int counter = (gate->counter + 1U) % 4U; + const bool cancel = button == 6U; + // The original gas SET caller is separate from ordinary longActive requests. + // Native requires the physical slot and gas/Drive/control ownership; host + // requires stock speed < actual speed < HUD speed and the frame52 cadence. + const bool gas_set = gate->armed && gate->drive && !gate->brake && !gate->regen && gate->gas && controls && (button == 3U); + valid = ready && ((bus == 0U) || camera_cancel) && (cancel || gas_set || + (acceleration_owned && (gate->speed >= 10.72896F) && ((button == 2U) || (button == 3U)))) && + gm_bolt_cc_button_tuple(data, button, counter); + // A matching TX attempt consumes the physical slot even when malformed. + gate->credit = false; + } + return valid; +} diff --git a/opendbc_repo/opendbc/safety/tests/test_gm_bolt_cc.py b/opendbc_repo/opendbc/safety/tests/test_gm_bolt_cc.py new file mode 100644 index 0000000000..b2600eca46 --- /dev/null +++ b/opendbc_repo/opendbc/safety/tests/test_gm_bolt_cc.py @@ -0,0 +1,302 @@ +"""Conventional Bolt ownership through the registered GM hooks.""" + +import unittest +from opendbc.car.gm.bolt_cc import button_bytes +from opendbc.car.structs import CarParams +from opendbc.safety.tests.libsafety import libsafety_py + +WORDS = (0xC110, 0xC111, 0xC120, 0xC121, 0xC130, 0xC131) + + +def setup(word): + safety = libsafety_py.libsafety + safety.init_tests() + assert safety.set_safety_hooks(int(CarParams.SafetyModel.gm), word) == 0 + + +def packet(address, data, us=1_000_000, *, transmit=False, bus=0): + safety = libsafety_py.libsafety + safety.set_timer(us) + msg = libsafety_py.make_CANPacket(address, bus, data) + return safety.safety_tx_hook(msg) if transmit else safety.safety_rx_hook(msg) + + +def ready(word, *, gas=False, gear=4, manual=False): + setup(word) + wheel = int(20 * 3.6 / 0.0311) + frames = [ + (0x1F5, bytes((0, 0, 0, gear, 0, 2 * int(manual), 0, 0))), + (0xC9, bytes(8)), + (0xBD, bytes(7)), + (0x1C4, bytes((0, 0, 0, 0, 0, int(gas), 0, 0))), + (0x184, bytes(8)), + (0x34A, bytes((wheel >> 8, wheel & 255, wheel >> 8, wheel & 255, 9))), + (0x1E1, button_bytes(1, 0)), + (0x3D1, bytes((0, 0, 0, 0, 128, 0, 0, 0))), + ] + for address, data in frames: + packet(address, data) + return frames + + +class TestGmBoltCcSafety(unittest.TestCase): + def test_forward_gear_transitions_retain_permission_but_revocation_requires_button_edge(self): + for word in (*WORDS, 0xC140, 0xC141): + for denied_gear, manual in ((0, False), (1, False), (2, False), (3, False), + (5, False), (7, False), (4, True), (6, True)): + with self.subTest(word=word, denied_gear=denied_gear, manual=manual): + ready(word, gear=6) + safety = libsafety_py.libsafety + self.assertTrue(safety.get_controls_allowed()) + for step, gear in enumerate((4, 6), start=1): + packet(0x1F5, bytes((0, 0, 0, gear, 0, 0, 0, 0)), 1_000_000 + step * 1000) + self.assertTrue(safety.get_controls_allowed()) + packet(0x1F5, bytes((0, 0, 0, denied_gear, 0, 2 * int(manual), 0, 0)), 1_003_000) + self.assertFalse(safety.get_controls_allowed()) + packet(0x1F5, bytes((0, 0, 0, 6, 0, 0, 0, 0)), 1_004_000) + self.assertFalse(safety.get_controls_allowed()) + packet(0x1E1, button_bytes(1, 1), 1_005_000) + self.assertFalse(safety.get_controls_allowed()) + packet(0x1E1, button_bytes(3, 2), 1_006_000) + self.assertFalse(safety.get_controls_allowed()) + packet(0x1E1, button_bytes(1, 3), 1_007_000) + self.assertTrue(safety.get_controls_allowed()) + + def test_low_gear_physical_cruise_and_set_rearm_for_every_bolt_profile(self): + for word in (*WORDS, 0xC140, 0xC141): + for gear in (4, 6): + with self.subTest(word=word, gear=gear): + ready(word, gear=gear) + safety = libsafety_py.libsafety + self.assertTrue(safety.get_controls_allowed()) + self.assertTrue(packet(0x180, bytes((8, 1, 0, 0)), 1_001_000, transmit=True)) + packet(0xC9, bytes((0, 0, 0, 0, 0, 1, 0, 0)), 1_002_000) + self.assertFalse(safety.get_controls_allowed()) + packet(0xC9, bytes(8), 1_003_000) + self.assertFalse(safety.get_controls_allowed()) + packet(0x1E1, button_bytes(3, 1), 1_004_000) + self.assertFalse(safety.get_controls_allowed()) + packet(0x1E1, button_bytes(1, 2), 1_005_000) + self.assertTrue(safety.get_controls_allowed()) + packet(0x1F5, bytes((0, 0, 0, 2, 0, 0, 0, 0)), 1_006_000) + self.assertFalse(safety.get_controls_allowed()) + + def test_non_forward_and_manual_gears_cannot_grant_bolt_permission(self): + for word in (*WORDS, 0xC140, 0xC141): + for gear, manual in ((0, False), (1, False), (2, False), (3, False), + (5, False), (7, False), (4, True), (6, True)): + with self.subTest(word=word, gear=gear, manual=manual): + ready(word, gear=gear, manual=manual) + safety = libsafety_py.libsafety + self.assertFalse(safety.get_controls_allowed()) + packet(0x1E1, button_bytes(3, 1), 1_001_000) + packet(0x1E1, button_bytes(1, 2), 1_002_000) + self.assertFalse(safety.get_controls_allowed()) + self.assertFalse(packet(0x180, bytes((8, 1, 0, 0)), 1_003_000, transmit=True)) + + def test_pcm_rising_cannot_override_held_park_brake_or_paddle(self): + for word in WORDS: + for source, blocked, clear in ( + (0x1F5, bytes((0, 0, 0, 1, 0, 0, 0, 0)), bytes((0, 0, 0, 4, 0, 0, 0, 0))), + (0xC9, bytes((0, 0, 0, 0, 0, 1, 0, 0)), bytes(8)), + (0xBD, bytes((16, 0, 0, 0, 0, 0, 0)), bytes(7)), + ): + with self.subTest(word=word, source=source): + ready(word) + packet(0x3D1, bytes(8), 1_002_000) + packet(source, blocked, 1_003_000) + packet(0x3D1, bytes((0, 0, 0, 0, 128, 0, 0, 0)), 1_004_000) + self.assertFalse(libsafety_py.libsafety.get_controls_allowed()) + self.assertFalse(packet(0x180, bytes((8, 1, 0, 0)), 1_005_000, transmit=True)) + packet(source, clear, 1_006_000) + packet(0x1E1, button_bytes(1, 1), 1_007_000) + self.assertFalse(libsafety_py.libsafety.get_controls_allowed()) + packet(0x1E1, button_bytes(3, 2), 1_008_000) + self.assertFalse(libsafety_py.libsafety.get_controls_allowed()) + packet(0x1E1, button_bytes(1, 3), 1_009_000) + self.assertTrue(libsafety_py.libsafety.get_controls_allowed()) + self.assertTrue(packet(0x1E1, button_bytes(2, 0), 1_010_000, transmit=True)) + + def test_required_health_and_global_revocation_latch(self): + for word in WORDS: + frames = ready(word) + safety = libsafety_py.libsafety + self.assertTrue(safety.get_controls_allowed()) + safety.set_timer(3_000_000) + safety.safety_tick() + self.assertFalse(safety.get_controls_allowed()) + for address, data in frames: + packet(address, button_bytes(1, 1) if address == 0x1E1 else data, 3_001_000) + self.assertFalse(safety.get_controls_allowed()) + packet(0x1E1, button_bytes(2, 2), 3_002_000) + self.assertTrue(safety.get_controls_allowed()) + + def test_gas_lateral_and_exact_unowned_transmit_denials(self): + for word in WORDS: + ready(word, gas=True) + self.assertTrue(libsafety_py.libsafety.get_controls_allowed()) + self.assertTrue(packet(0x180, bytes((8, 1, 0, 0)), 1_001_000, transmit=True)) + self.assertFalse(packet(0x1E1, button_bytes(2, 1), 1_001_000, transmit=True)) + ready(word, gas=True) + self.assertTrue(packet(0x1E1, button_bytes(3, 1), 1_001_000, transmit=True)) + for address, length in ((0x200, 6), (0x315, 5), (0x2CB, 8), (0xBD, 7), (0x1F5, 8), (0x3D1, 8)): + self.assertFalse(packet(address, bytes(length), 1_002_000, transmit=True)) + ready(word) + bad = bytearray(button_bytes(2, 1)) + bad[6] ^= 1 + self.assertFalse(packet(0x1E1, bad, 1_001_000, transmit=True)) + self.assertFalse(packet(0x1E1, button_bytes(2, 1), 1_002_000, transmit=True)) + + def test_invalid_sibling_words_and_existing_camera_owner(self): + for word in (0xC112, 0xC125, 0xC000, 0xFFFF): + setup(word) + self.assertFalse(packet(0x180, bytes(4), transmit=True)) + self.assertFalse(packet(0x1E1, button_bytes(3, 1), transmit=True)) + setup(1) + packet(0x1C4, bytes((0, 32, 0, 0, 0, 0, 0, 0))) + self.assertTrue(libsafety_py.libsafety.get_controls_allowed()) + self.assertTrue(packet(0x180, bytes(4), 1_001_000, transmit=True)) + self.assertFalse(packet(0x200, bytes(6), 1_001_000, transmit=True)) + + def test_timer_rollover_preserves_fresh_physical_sources(self): + safety = libsafety_py.libsafety + safety.set_alternative_experience(0) + for word in WORDS: + frames = ready(word) + setup(word) + for address, data in frames: + packet(address, data, 0xFFFFF000) + self.assertTrue(safety.get_controls_allowed()) + self.assertTrue(packet(0x180, bytes((8, 1, 0, 0)), 1000, transmit=True)) + packet(0xC9, bytes((0, 0, 0, 0, 0, 1, 0, 0)), 2000) + self.assertFalse(safety.get_controls_allowed()) + self.assertFalse(packet(0x180, bytes((8, 1, 0, 0)), 3000, transmit=True)) + + def test_pscm_copy_requires_exact_payload_fresh_source_and_cadence(self): + libsafety_py.libsafety.set_alternative_experience(0) + for word in WORDS: + for source, expected in ((bytes(8), bytes((0, 0, 32, 0, 0, 32, 0, 0))), + (bytes((0, 0, 32, 0, 0, 32, 0, 0)), bytes((0, 0, 32, 0, 0, 32, 0, 0)))): + with self.subTest(word=word, source=source): + ready(word) + packet(0x184, source, 1_000_001) + for index in range(8): + bad = bytearray(expected) + bad[index] ^= 1 + self.assertFalse(packet(0x184, bad, 1_001_000, transmit=True, bus=2)) + self.assertTrue(packet(0x184, expected, 1_001_000, transmit=True, bus=2)) + self.assertFalse(packet(0x184, expected, 1_002_000, transmit=True, bus=2)) + self.assertFalse(packet(0x184, expected, 1_101_001, transmit=True, bus=2)) + packet(0x184, source, 1_101_002) + self.assertTrue(packet(0x184, expected, 1_101_003, transmit=True, bus=2)) + for address in (0x409, 0x40A): + ready(word) + self.assertEqual(packet(address, bytes(7), 1_001_000, transmit=True), word in (0xC111, 0xC121, 0xC131)) + for index in range(7): + bad = bytearray(7) + bad[index] = 1 + self.assertFalse(packet(address, bad, 1_002_000, transmit=True)) + + def test_camera_cancel_requires_fresh_active_camera_and_consumes_credit(self): + libsafety_py.libsafety.set_alternative_experience(0) + for camera in (bytes(6), bytes((0, 0, 128, 0, 0, 0))): + ready(0xC140) + packet(0x370, camera, 1_000_001, bus=2) + expected = bool(camera[2] & 128) + self.assertEqual(packet(0x1E1, button_bytes(6, 1), 1_001_000, transmit=True, bus=2), expected) + self.assertFalse(packet(0x1E1, button_bytes(6, 1), 1_002_000, transmit=True, bus=2)) + frames = ready(0xC140) + packet(0x370, bytes((0, 0, 128, 0, 0, 0)), 1_000_001, bus=2) + for address, data in frames: + packet(address, button_bytes(1, 1) if address == 0x1E1 else data, 1_301_000) + self.assertFalse(packet(0x1E1, button_bytes(6, 2), 1_302_000, transmit=True, bus=2)) + packet(0x370, bytes((0, 0, 128, 0, 0, 0)), 1_303_000, bus=2) + packet(0x1E1, button_bytes(1, 2), 1_304_000) + self.assertTrue(packet(0x1E1, button_bytes(6, 3), 1_305_000, transmit=True, bus=2)) + + def test_physical_cancel_withdraws_and_set_release_rearms(self): + libsafety_py.libsafety.set_alternative_experience(0) + for word in WORDS: + ready(word) + packet(0x1E1, button_bytes(6, 1), 1_001_000) + self.assertFalse(libsafety_py.libsafety.get_controls_allowed()) + packet(0x1E1, button_bytes(1, 2), 1_002_000) + self.assertFalse(libsafety_py.libsafety.get_controls_allowed()) + packet(0x1E1, button_bytes(3, 3), 1_003_000) + packet(0x1E1, button_bytes(1, 0), 1_004_000) + self.assertTrue(libsafety_py.libsafety.get_controls_allowed()) + + def test_invalid_profile_initializer_rejects_transmit_admission(self): + setup(0xC110) + self.assertFalse(libsafety_py.libsafety.safety_test_gm_bolt_unowned_tx()) + self.assertTrue(libsafety_py.libsafety.safety_test_gm_bolt_invalid_profile()) + setup(0xC110) + self.assertFalse(packet(0x200, bytes(6), transmit=True)) + + def test_selected_forwarding_policy_preserves_owned_blocks_and_unowned_routing(self): + for word in WORDS: + libsafety_py.libsafety.set_alternative_experience(0) + setup(word) + for bus in (0, 1, 2): + for address in (0x180, 0x184, 0x3D1, 0x321): + with self.subTest(word=word, bus=bus, address=hex(address)): + blocked = (bus == 2 and address == 0x180) or (bus == 0 and address in (0x184, 0x3D1)) + self.assertEqual(libsafety_py.libsafety.safety_test_selected_fwd(bus, address), blocked) + destination = -1 if blocked or bus == 1 else 2 if bus == 0 else 0 + self.assertEqual(libsafety_py.libsafety.safety_fwd_hook(bus, address), destination) + + +class TestGmRemovedBoltStockSafety(unittest.TestCase): + def test_reduced_stock_has_cancel_only_physical_button_authority(self): + for word in (0xE710, 0xE711, 0xE712, 0xE713): + for button in (2, 3, 6): + with self.subTest(word=word, button=button): + ready(word) + packet(0xC9, bytes((0, 0, 0, 32, 0, 0, 0, 0)), 1_000_001) + self.assertEqual(packet(0x1E1, button_bytes(button, 1), 1_001_000, transmit=True), button == 6) + if button == 6: + self.assertFalse(packet(0x1E1, button_bytes(6, 1), 1_002_000, transmit=True)) + for address, length in ((0x200, 6), (0x315, 5), (0xBD, 7), (0x1F5, 8), (0x409, 7), (0x40A, 7)): + self.assertFalse(packet(address, bytes(length), 1_002_000, transmit=True)) + self.assertFalse(packet(0x1E1, button_bytes(6, 1), 1_002_000, transmit=True, bus=2)) + + def test_reduced_stock_requires_current_cruise_and_neutral_credit(self): + for word in (0xE710, 0xE711, 0xE712, 0xE713): + with self.subTest(word=word): + ready(word) + packet(0xC9, bytes((0, 0, 0, 32, 0, 0, 0, 0)), 1_000_001) + self.assertFalse(packet(0x1E1, button_bytes(6, 1), 1_100_001, transmit=True)) + ready(word) + packet(0xC9, bytes((0, 0, 0, 32, 0, 0, 0, 0)), 1_000_001) + packet(0x3D1, bytes(8), 1_001_000) + self.assertFalse(packet(0x1E1, button_bytes(6, 1), 1_002_000, transmit=True)) + ready(word) + packet(0xC9, bytes((0, 0, 0, 32, 0, 0, 0, 0)), 1_000_001) + self.assertTrue(packet(0x1E1, button_bytes(6, 1), 1_001_000, transmit=True)) + packet(0x1E1, button_bytes(1, 0), 1_002_000) + self.assertFalse(packet(0x1E1, button_bytes(6, 1), 1_003_000, transmit=True)) + + def test_stock_dispatch_resets_to_existing_manual_button_authority(self): + for stock, ordinary in ((0xE710, 0xC111), (0xE711, 0xC121), (0xE712, 0xC131), (0xE713, 0xC141)): + for word, expected in ((stock, False), (ordinary, True), (stock, False)): + ready(word) + packet(0xC9, bytes((0, 0, 0, 32, 0, 0, 0, 0)), 1_000_001) + libsafety_py.libsafety.set_controls_allowed(True) + self.assertEqual(packet(0x1E1, button_bytes(2, 1), 1_001_000, transmit=True), expected) + + def test_stock_main_cycle_cannot_restore_consumed_or_withdrawn_credit(self): + for word in (0xE710, 0xE711, 0xE712, 0xE713): + ready(word) + packet(0xC9, bytes(8), 1_000_001) + packet(0xC9, bytes((0, 0, 0, 32, 0, 0, 0, 0)), 1_000_002) + self.assertFalse(packet(0x1E1, button_bytes(6, 1), 1_001_000, transmit=True)) + packet(0x1E1, button_bytes(1, 1), 1_002_000) + self.assertTrue(packet(0x1E1, button_bytes(6, 2), 1_003_000, transmit=True)) + + def test_unknown_removed_bolt_words_cannot_borrow_stock_or_active_authority(self): + for word in (0xE6FF, 0xE704, 0xE70F, 0xE714, 0xE7FF): + setup(word) + libsafety_py.libsafety.set_controls_allowed(True) + self.assertFalse(packet(0x180, bytes(4), transmit=True)) + self.assertFalse(packet(0x1E1, button_bytes(6, 1), transmit=True)) + self.assertFalse(packet(0x409, bytes(7), transmit=True))