mawei/StarPilot
1
0
Fork 0
mirror of https://github.com/firestar5683/StarPilot.git synced 2026-11-14 16:33:59 +08:00
Code Issues Packages Projects Releases Wiki Activity

Bolt Cruise

Browse Source 
Implement Bolt cruise control with its own host and native admission gates.
This commit is contained in:
firestar5683
2026-09-25 14:53:18 -05:00
parent 59fb0f9ef9
commit e6aea9f9fa
5 changed files with 1282 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
opendbc_repo/opendbc/car/gm/bolt_cc.py
+263
View File
@@ -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
opendbc_repo/opendbc/car/gm/tests/test_bolt_cc.py
+269
View File
@@ -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()
opendbc_repo/opendbc/safety/modes/gm_bolt_cc.h
+279
View File
@@ -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);
}
opendbc_repo/opendbc/safety/modes/gm_bolt_cc_gate.h
+169
View File
@@ -0,0 +1,169 @@
#pragma once
#include <stdint.h>
#include <stdbool.h>
// 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;
}
opendbc_repo/opendbc/safety/tests/test_gm_bolt_cc.py
+302
View File
@@ -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))