Bolt Pedal

Implement Bolt Pedal command generation and mode selection with dedicated native safety coverage. Keep pedal actuation separate from the factory cruise families.

opendbc_repo/opendbc/car/gm/BOLT_PEDAL.md

@@ -0,0 +1,9 @@
+# Bolt pedal and conventional-cruise configurations
+
+The 2017, 2018–21 and 2022–23 conventional-cruise Bolt configurations and the 2022–23 ACC-pedal configuration have separate owners. Normal CAN/firmware identification remains available. A supported live pedal sensor (`0x201`, bus 0) selects the interceptor branch; an identified Bolt EUV or 2022–23 ACC Bolt with that sensor is mapped to `CHEVROLET_BOLT_ACC_2022_2023_PEDAL`. The current interface does not require a saved `GMPedalLongitudinal` opt-in. `DisableOpenpilotLongitudinal` is a separate startup preference that defaults off; stock restoration depends on the actual selected topology and camera evidence.
+
+The interceptor branch uses paired pedal tracks for acceleration and a regenerative paddle/PRNDL scheduler for braking. Its active pedal and scheduler commands require fresh stock L with ManualMode clear, healthy current sensor/source packets, active longitudinal controls and no driver gas, brake or paddle override. This restriction belongs to the interceptor/scheduler branch. The separate conventional-cruise controller, including the 2018–21 `0xC120`/`0xC121` profiles, accepts Drive or Low and controls speed through stock cruise buttons rather than pedal actuation.
+
+The ACC-pedal branch additionally requires fresh confirmation that stock ACC is disengaged and retains stock cancellation. It can send qualified friction-brake commands on the powertrain bus, including low-speed braking and a bounded zero-brake unwind. The non-ACC interceptor branches use regenerative braking without that friction sender. None of these branches sends ASCM gas/regen commands. A fault, stale sensor, stock takeover or control release withdraws active pedal requests; scheduler release feeds remain conditional on actual L ownership. Driver paddle or gear changes discard pending active scheduler feeds.
+
+Native safety independently checks the selected profile, pedal pairing/CRC/counter, source freshness, scheduler payload and ownership, and applicable numeric command bounds. Existing tests join generated DBC parsing, controller output and native safety. The separate gateway Volt Auto Hold feature does not change these Bolt branches or their regenerative stopping behavior.

opendbc_repo/opendbc/car/gm/bolt_mode.py

@@ -0,0 +1,70 @@
+"""Mode-transition output shaping for admitted Bolt longitudinal owners."""
+
+
+class BoltModeTransition:
+  def __init__(self):
+    self.prev_mode = 'acc'
+    self.current_mode = 'acc'
+    self.timer = 0.0
+    self.duration = 1.0
+    self.transitioning = False
+
+  def update(self, experimental_mode, dt):
+    # Lost mode telemetry cannot create a new transition. A current transition
+    # still advances under the caller's active PID ownership.
+    if type(experimental_mode) is bool:
+      mode = 'blended' if experimental_mode else 'acc'
+      if mode != self.current_mode:
+        self.prev_mode = self.current_mode
+        self.current_mode = mode
+        self.transitioning = True
+        self.timer = 0.0
+    if self.transitioning:
+      self.timer += dt
+      if self.timer >= self.duration:
+        self.transitioning = False
+
+  @property
+  def leaving_experimental(self):
+    return self.transitioning and self.prev_mode == 'blended' and self.current_mode == 'acc'
+
+  def shape(self, output, last_output):
+    if self.leaving_experimental:
+      if output > last_output:
+        progress = min(1.0, self.timer / max(self.duration, 1e-3))
+        output = last_output + (output - last_output) * progress
+    elif self.transitioning and self.prev_mode == 'acc' and self.current_mode == 'blended':
+      if output < 0.0 and output < last_output:
+        progress = min(1.0, self.timer / self.duration)
+        urgency = abs(output / -4.0)
+        urgency_smooth = min(1.0, urgency ** 0.4)
+        blend = 1.0 - (1.0 - progress) * (1.0 - urgency_smooth)
+        output = last_output + (output - last_output) * blend
+    return output
+
+
+def policy_for(cp):
+  from opendbc.car.structs import CarParams
+  from opendbc.car.gm.values import (CAR, GMFlags, GMSafetyFlags, is_bolt_cc_profile, is_bolt_euv_longitudinal,
+                                      is_ordinary_cc_profile, is_ordinary_camera_profile, is_conventional_cc_pedal_profile)
+  from opendbc.car.gm.longitudinal import policy_for as pedal_policy_for
+
+  if not cp.openpilotLongitudinalControl:
+    return None
+  if (is_conventional_cc_pedal_profile(cp) or is_ordinary_camera_profile(cp, longitudinal=True) or
+      is_ordinary_cc_profile(cp) or is_bolt_cc_profile(cp) or is_bolt_euv_longitudinal(cp)):
+    return BoltModeTransition()
+  from opendbc.car.gm.values import gm_control_word
+  base = GMSafetyFlags.HW_CAM | GMSafetyFlags.EV | GMSafetyFlags.PEDAL_LONG | GMSafetyFlags.PADDLE_SCHED
+  words = {
+    CAR.CHEVROLET_BOLT_CC_2017: base | GMSafetyFlags.NO_ACC | GMSafetyFlags.BOLT_2017,
+    CAR.CHEVROLET_BOLT_CC_2018_2021: base | GMSafetyFlags.NO_ACC,
+    CAR.CHEVROLET_BOLT_CC_2022_2023: base | GMSafetyFlags.NO_ACC | GMSafetyFlags.BOLT_GEN2,
+    CAR.CHEVROLET_BOLT_ACC_2022_2023_PEDAL: base | GMSafetyFlags.BOLT_ACC_PEDAL | GMSafetyFlags.BOLT_GEN2,
+  }
+  if (pedal_policy_for(cp) is not None and not cp.flags & GMFlags.CC_LONG.value and
+      len(cp.safetyConfigs) == 1 and
+      cp.networkLocation == CarParams.NetworkLocation.fwdCamera and
+      gm_control_word(cp) == words.get(cp.carFingerprint)):
+    return BoltModeTransition()
+  return None

opendbc_repo/opendbc/car/gm/camera.py

@@ -1,7 +1,10 @@
 """Ordinary camera acceleration owner in normalized GM wire units."""
-from opendbc.car.gm.values import CAR, is_ordinary_camera_profile
-from opendbc.car.gm.longitudinal import GMOrdinaryLongitudinalPolicy, _GMDefaultStopPolicy, AscmStopEvidence
+from opendbc.car.gm.values import CAR, is_ordinary_camera_profile, camera_acc_pedal_profile, volt_cc_pedal_profile
+from opendbc.car.gm.longitudinal import GMOrdinaryLongitudinalPolicy, GMVoltLongitudinalPolicy, _GMDefaultStopPolicy, AscmStopEvidence
+import numpy as np
+
 from opendbc.car.gm.truck_longitudinal import GMTruckLongitudinalPolicy
+from opendbc.car.gm.conventional_pedal import GMConventionalPedalPolicy
 
 
 class CameraLongitudinalPolicy(GMOrdinaryLongitudinalPolicy):
@@ -14,7 +17,25 @@ class TruckCameraLongitudinalPolicy(GMTruckLongitudinalPolicy):
     return _GMDefaultStopPolicy(.25, AscmStopEvidence)
 
 
+class CameraPedalLongitudinalPolicy(CameraLongitudinalPolicy):
+  stopping_decel_rate = float(np.float32(.8))
+  kp = ((0., 5., 15., 35.), tuple(float(np.float32(value)) for value in (.09, .08, .06, .045)))
+
+  def feedforward(self, target, speed, last_output):
+    return target * float(np.float32(.25))
+
+
 def policy_for(cp):
+  cc_profile = volt_cc_pedal_profile(cp)
+  if cc_profile is not None and cc_profile.longitudinal:
+    policy = GMConventionalPedalPolicy(False)
+    policy.starting_speed = .75
+    return policy
+  profile = camera_acc_pedal_profile(cp)
+  if profile is not None and profile.longitudinal:
+    return (GMVoltLongitudinalPolicy(gateway=profile.topology == "gateway")
+            if profile.topology in ("gateway", "ascm", "sdgm") else CameraPedalLongitudinalPolicy())
+
   if not is_ordinary_camera_profile(cp, longitudinal=True):
     return None
   return TruckCameraLongitudinalPolicy() if cp.carFingerprint == CAR.CHEVROLET_SILVERADO else CameraLongitudinalPolicy()

opendbc_repo/opendbc/car/gm/conventional_pedal.py

@@ -0,0 +1,145 @@
+"""Acceleration tuning for conventional-cruise GM pedal configurations."""
+
+import numpy as np
+
+from opendbc.car.gm.values import CAR, is_conventional_cc_pedal_profile, is_silverado_cc_pedal_profile
+from opendbc.car.gm.longitudinal import GMPedalLongitudinalPolicy, _GMDefaultStopPolicy, _moving_stop_target_follow
+from opendbc.car.gm.cc_longitudinal import VoltCcEvidence
+
+
+def stop_start_speed(malibu: bool) -> float:
+  """Default stop/start speed for the conventional-cruise pedal policy."""
+  return .75 if malibu else .5
+
+
+class GMConventionalPedalPolicy(GMPedalLongitudinalPolicy):
+  def __init__(self, malibu: bool):
+    super().__init__(friction_variant=False)
+    # The reached original default reads this rate through Float32 CarParams storage.
+    self.stopping_decel_rate = float(np.float32(.8))
+    self.kp = (((0.0, 5.0, 35.0), (0.06, 0.05, 0.04)) if malibu else
+               ((0.0, 5.0, 15.0, 35.0), (0.09, 0.08, 0.06, 0.045)))
+    self.kp = (self.kp[0], tuple(float(np.float32(value)) for value in self.kp[1]))
+    self.starting_speed = stop_start_speed(malibu)
+    self.feedforward_gain = float(np.float32(0.15 if malibu else 0.25))
+
+  def stop_policy(self):
+    return _GMDefaultStopPolicy(self.starting_speed, VoltCcEvidence)
+
+  def stopping_output(self, output, target, should_stop, cs):
+    return _moving_stop_target_follow(output, target, should_stop, cs, max(1.5, self.starting_speed + 1.))
+
+  def feedforward(self, target: float, speed: float, last_output: float) -> float:
+    return target * self.feedforward_gain
+
+
+def policy_for(cp):
+  if not is_conventional_cc_pedal_profile(cp) or not cp.openpilotLongitudinalControl:
+    return None
+  return GMConventionalPedalPolicy(cp.carFingerprint == CAR.CHEVROLET_MALIBU_CC)
+
+
+class CancelCredit:
+  DRIVER_OVERRIDE_LENGTHS = {0xBD: 7, 0xC9: 8, 0x1C4: 8, 0x1F5: 8}
+
+  """One cancellation per sequential, byte-valid neutral wheel-button packet."""
+
+  def __init__(self, *, neutral_interval_ns=300_000_000, allow_non_acc_neutral=False):
+    self.neutral_interval_ns = neutral_interval_ns
+    self.allow_non_acc_neutral = allow_non_acc_neutral
+    self.counter = None
+    self.packet_ns = self.source_ns = self.credit_ns = 0
+    self.main_ns = self.stock_ns = 0
+    self.gear_ns = self.gear_packet_ns = 0
+    self.main = self.stock_active = False
+    self.override_ns = {}
+
+  def observe(self, can_packets, *, clear_on_main_off=False, clear_on_driver_override=False, observe_gear=False):
+    # Decoded signals omit reserved bits. Observe ordered physical packets so
+    # the sender cannot grant credit for a button frame Panda would reject.
+    for stamp, packets in can_packets:
+      for address, raw, bus in packets:
+        if bus != 0:
+          continue
+        if observe_gear and address == 0x1F5:
+          if len(raw) == 8 and stamp > max(0, self.gear_packet_ns):
+            self.gear_ns = stamp
+          elif len(raw) != 8 or stamp != self.gear_ns or stamp <= 0:
+            # Same-event valid observations retain a qualified clock, never refresh it.
+            self.gear_ns = 0
+          self.gear_packet_ns = max(self.gear_packet_ns, stamp)
+          if not self.gear_ns:
+            self.credit_ns = 0
+        if clear_on_driver_override and stamp > self.override_ns.get(address, 0):
+          if address in self.DRIVER_OVERRIDE_LENGTHS and len(raw) == self.DRIVER_OVERRIDE_LENGTHS[address]:
+            self.override_ns[address] = stamp
+            override = ((address == 0xBD and raw[0] >> 4 != 0) or
+                        (address == 0xC9 and raw[5] & 1 != 0) or
+                        (address == 0x1C4 and raw[5] != 0) or
+                        (address == 0x1F5 and (raw[3] & 15 not in (4, 6) or raw[5] & 2 != 0)))
+            if override:
+              self.credit_ns = 0
+        if address == 0xC9:
+          self.main_ns = stamp if len(raw) == 8 and stamp > 0 else 0
+          self.main = bool(self.main_ns and raw[3] & 0x20)
+          if clear_on_main_off and not self.main:
+            self.credit_ns = 0
+        elif address == 0x3D1:
+          self.stock_ns = stamp if len(raw) == 8 and stamp > 0 else 0
+          self.stock_active = bool(self.stock_ns and raw[4] & 0x80)
+          if not self.stock_active:
+            self.credit_ns = 0
+        elif address == 0x1E1:
+          self._observe_button(raw, stamp)
+
+  def _observe_button(self, raw, stamp):
+    if len(raw) != 7:
+      self.credit_ns = 0
+      return
+    counter = raw[4] & 3
+    if stamp <= 0 or stamp < self.packet_ns:
+      # Panda still observes the physical counter even if its host timestamp is unusable.
+      self.counter = counter
+      self.credit_ns = self.source_ns = 0
+      return
+    checksum = 0xFF + counter * 0x4EF
+    neutral = raw == bytes((0, 0, 0, 1, counter, 0x10 | (checksum >> 8), checksum & 0xFF))
+    if self.allow_non_acc_neutral:
+      non_acc_checksum = 0xF0 + counter * 0x3F0
+      neutral |= raw == bytes((0, 0, 0, 0, counter, 0x10 | (non_acc_checksum >> 8), non_acc_checksum & 0xFF))
+    first = self.counter is None
+    timely = self.source_ns > 0 and 0 <= stamp - self.source_ns <= self.neutral_interval_ns
+    if neutral and (first or timely and counter == (self.counter + 1) % 4):
+      self.credit_ns = stamp
+      self.source_ns = stamp
+    elif not neutral or not timely or counter != self.counter:
+      self.credit_ns = 0
+    if first or counter != self.counter:
+      self.source_ns = stamp
+    self.counter, self.packet_ns = counter, stamp
+
+  def available(self, now_ns, used_ns):
+    return (self.credit_ns > used_ns and self.main and self.stock_active and
+            all(stamp > 0 and 0 <= now_ns - stamp <= 300_000_000
+                for stamp in (self.main_ns, self.stock_ns, self.source_ns)))
+
+
+def cancel_sources_current(cs, now_ns):
+  """Keep the parser's process-health checks alongside raw cancellation credit."""
+  return cs.out.canValid and not cs.out.canTimeout
+
+
+def sources_current(cs, now_ns):
+  if is_silverado_cc_pedal_profile(cs.CP):
+    stamps = cs.silverado_pedal_sources
+    limits = (300_000_000, 100_000_000, 300_000_000, 300_000_000,
+              100_000_000, 100_000_000, 300_000_000, 100_000_000)
+    return (len(stamps) == len(limits) and cs.out.canValid and not cs.out.canTimeout and
+            all(stamp > 0 and 0 <= now_ns - stamp <= limit for stamp, limit in zip(stamps, limits, strict=True)))
+  stamps = cs.conventional_pedal_sources
+  limits = (300_000_000, 300_000_000, 300_000_000, 300_000_000,
+            300_000_000, 100_000_000, 300_000_000, 100_000_000)
+  if len(stamps) == 9:
+    limits += (100_000_000,)
+  return (len(stamps) == len(limits) and cs.out.canValid and not cs.out.canTimeout and
+          all(stamp > 0 and 0 <= now_ns - stamp <= limit for stamp, limit in zip(stamps, limits, strict=True)))

opendbc_repo/opendbc/car/gm/conventional_pedal_command.py

@@ -0,0 +1,75 @@
+"""Conventional-cruise GM pedal command and Malibu physical limits."""
+
+import math
+import numpy as np
+from opendbc.car import ACCELERATION_DUE_TO_GRAVITY
+from opendbc.car.gm.values import CAR
+from opendbc.car.gm.conventional_pedal import stop_start_speed
+from opendbc.car.gm.silverado_cc import pedal_fraction, pedal_slew
+
+# The no-paddle interpolation and slew are shared source arithmetic. Physical
+# gas/brake limits, pitch and launch thresholds remain bound to this identity.
+
+class ConventionalPedalCommand:
+  def __init__(self, cp):
+    self.mass = cp.mass
+    self.wheelbase = cp.wheelbase
+    self.stop_accel = cp.stopAccel
+    malibu = cp.carFingerprint == CAR.CHEVROLET_MALIBU_CC
+    self.stop_speed = self.start_speed = stop_start_speed(malibu)
+    self.max_gas = 8191 if malibu else 7168
+    self.brake_bp = [-4., -.1] if malibu else [-4., 0.]
+    self.reset()
+
+  def reset(self):
+    self.steady = 0.
+    self.active_last = False
+    self.recovering = False
+    self.recovery_emitted = 0.
+
+  def prime_recovery(self):
+    # Invalid input revokes actuation. Recovery resumes through the original slew from zero.
+    self.steady = 0.
+    self.active_last = True
+    self.recovering = True
+    self.recovery_emitted = 0.
+
+  def pause_recovery(self):
+    if self.recovering:
+      self.recovery_emitted = 0.
+
+  def update(self, accel, long_active, cs, *, stopping, resume, orientation=None):
+    # The original outer caller bypasses calc on these branches and retains its memory.
+    if not long_active:
+      self.pause_recovery()
+      return 0., -650., 0
+    if cs.vEgo < .25 and stopping and not resume:
+      self.pause_recovery()
+      return 0., -650., int(min(-100 * self.stop_accel, 400))
+    target = pedal_fraction(accel, cs.vEgo)
+    self.steady = pedal_slew(target, self.steady, accel, cs.vEgo) if self.active_last else target
+    self.active_last = True
+    command = self.steady
+
+    pitch = 0.
+    if orientation is not None and len(orientation) == 3 and cs.vEgo > self.stop_speed and math.isfinite(orientation[1]):
+      pitch = math.sin(orientation[1]) * ACCELERATION_DUE_TO_GRAVITY
+      pitch = 0. if pitch > 0. and accel > 0. else min(pitch, .20)
+    radius = .075 * self.wheelbase + .1453
+    drag = .5 * .30 * (1.05 * self.wheelbase + .0679) * 1.225 * cs.vEgo ** 2
+    scaled = radius * (self.mass * float(np.clip(accel + pitch, -4., 2.)) + drag) + 6150
+    gas = int(round(np.clip(scaled, 5500, self.max_gas)))
+    brake = int(round(np.interp(min((scaled - 6150) / (radius * self.mass), 0.), self.brake_bp, [400., 0.])))
+    if brake > 0 or stopping:
+      gas = 5500
+    if gas > 5500 and cs.cruiseState.standstill and (cs.standstill or cs.vEgo < max(self.start_speed, .3)):
+      command = 18. / 255.
+      gas, brake = 5500, 0
+    if self.recovering:
+      desired = command
+      # Bound the final command, including SNG, after invalid input. Zero and decreases are immediate.
+      command = min(desired, pedal_slew(desired, self.recovery_emitted, accel, cs.vEgo)) if desired > .001 else 0.
+      self.recovery_emitted = command
+      if command == desired:
+        self.recovering = False
+    return command, gas - 6150, brake

opendbc_repo/opendbc/car/gm/tests/test_bolt_acc_pedal_params.py

@@ -0,0 +1,74 @@
+import unittest
+
+from opendbc.can import CANPacker
+from opendbc.car import Bus
+from opendbc.car.gm.gmcan import create_friction_brake_command
+from opendbc.car.gm.interface import CarInterface
+from opendbc.car.gm.tests.test_bolt_pedal import params
+from opendbc.car.gm.values import CAR, DBC, PEDAL_BOLT_CAR
+
+
+class TestBoltAccPedalParams(unittest.TestCase):
+  def test_stop_target_and_delay_follow_final_control_owner(self):
+    for candidate in PEDAL_BOLT_CAR:
+      for alpha in (False, True):
+        for setting in (False, True):
+          for observed in (False, True):
+            with self.subTest(candidate=candidate, alpha=alpha, setting=setting, observed=observed):
+              cp = params(candidate, setting, observed, alpha_long=alpha)
+              self.assertAlmostEqual(cp.stopAccel, -0.25)
+              expected_delay = 0.6 if setting and observed else 1.0
+              self.assertAlmostEqual(cp.longitudinalActuatorDelay, expected_delay)
+    for candidate in (CAR.CHEVROLET_BOLT_EUV, CAR.CHEVROLET_BOLT_ACC_2022_2023):
+      for alpha in (False, True):
+        with self.subTest(candidate=candidate, alpha=alpha):
+          cp = params(candidate, True, True, alpha_long=alpha)
+          expected_stop = -0.25 if candidate == CAR.CHEVROLET_BOLT_ACC_2022_2023 or alpha else -2.0
+          self.assertAlmostEqual(cp.stopAccel, expected_stop)
+          self.assertAlmostEqual(cp.longitudinalActuatorDelay, 0.5)
+
+  def test_final_stop_target_for_all_bolt_cc_camera_profiles(self):
+    for candidate in (CAR.CHEVROLET_BOLT_CC_2017, CAR.CHEVROLET_BOLT_CC_2018_2021,
+                      CAR.CHEVROLET_BOLT_CC_2022_2023, CAR.CHEVROLET_BOLT_ACC_2022_2023_PEDAL):
+      for alpha in (False, True):
+        for camera in (False, True):
+          for setting in (False, True):
+            for observed in (False, True):
+              with self.subTest(candidate=candidate, alpha=alpha, camera=camera, setting=setting, observed=observed):
+                cp = params(candidate, setting, observed, alpha_long=alpha, camera=camera)
+                self.assertEqual(cp.stopAccel, -0.25)
+                self.assertAlmostEqual(cp.longitudinalActuatorDelay, 0.6 if setting and observed else 1.0)
+                self.assertTrue(cp.openpilotLongitudinalControl)
+
+  def test_friction_braking_and_speed_shaped_launch_limits(self):
+    for alpha in (False, True):
+      cp = params(CAR.CHEVROLET_BOLT_ACC_2022_2023_PEDAL, True, True, alpha_long=alpha)
+      for speed, maximum in ((0., .54), (.75, .64), (1.5, .74), (4., 1.03), (8., 1.46), (15., 2.), (40., 2.)):
+        with self.subTest(alpha=alpha, speed=speed):
+          minimum, actual_maximum = CarInterface.get_pid_accel_limits(cp, speed, 30.)
+          self.assertEqual(minimum, -4.)
+          self.assertAlmostEqual(actual_maximum, maximum)
+
+  def test_non_acc_pedal_keeps_regen_only_limits(self):
+    cp = params(CAR.CHEVROLET_BOLT_CC_2022_2023, True, True)
+    for speed, minimum, maximum in ((0., -.93, .54), (1.5, -1.28, .74), (4., -1.98, 1.03),
+                                    (8., -2.58, 1.46), (15., -2.86, 2.), (30., -2.95, 2.)):
+      self.assertEqual(CarInterface.get_pid_accel_limits(cp, speed, 30.), (minimum, maximum))
+
+  def test_friction_payload_modes_counter_and_checksum(self):
+    cp = params(CAR.CHEVROLET_BOLT_ACC_2022_2023_PEDAL, True, True)
+    packer = CANPacker(DBC[cp.carFingerprint][Bus.chassis])
+    for counter in range(4):
+      for brake, enabled, full_stop, mode in ((0, False, False, 1), (0, True, False, 9),
+                                             (100, True, False, 10), (400, True, True, 13)):
+        with self.subTest(counter=counter, brake=brake, enabled=enabled, full_stop=full_stop):
+          message = create_friction_brake_command(packer, 0, brake, counter, enabled, False, full_stop, cp)
+          encoded = (-brake) & 0xfff
+          checksum = (-(mode << 12) - encoded - counter) & 0xffff
+          expected = bytes(((mode << 4) | (encoded >> 8), encoded & 0xff,
+                            checksum >> 8, checksum & 0xff, counter))
+          self.assertEqual(message, (0x315, expected, 0))
+
+
+if __name__ == "__main__":
+  unittest.main()

opendbc_repo/opendbc/car/gm/tests/test_bolt_pedal.py

@@ -0,0 +1,208 @@
+from unittest.mock import patch
+import unittest
+from types import SimpleNamespace
+
+from opendbc.can import CANPacker
+
+from opendbc.car import Bus, gen_empty_fingerprint, structs
+from opendbc.car.gm.carcontroller import CarController
+from opendbc.car.gm.carstate import CarState
+from opendbc.car.gm.gmcan import pedal_crc
+from opendbc.car.gm.interface import CarInterface
+from opendbc.car.gm.values import CAR, DBC, GMFlags, GMSafetyFlags, NO_ACC_BOLT_CAR, PEDAL_BOLT_CAR, BOLT_CC_WORDS
+
+
+class SavedPedalSetting:
+  def __init__(self, enabled=False, missing=False):
+    self.enabled = enabled
+    self.missing = missing
+
+  def get_bool(self, key):
+    assert key == "GMPedalLongitudinal"
+    if self.missing:
+      from openpilot.common.params import UnknownKeyName
+      raise UnknownKeyName(key)
+    return self.enabled
+
+
+def params(candidate, setting=False, pedal=False, alpha_long=False, missing_key=False, camera=False):
+  fingerprint = gen_empty_fingerprint()
+  if camera:
+    fingerprint[2][0x180] = 4
+  if pedal:
+    fingerprint[0][0x201] = 6
+  with patch("opendbc.car.gm.interface.Params", return_value=SavedPedalSetting(setting, missing_key)):
+    return CarInterface.get_params(candidate, fingerprint, [], alpha_long, False, False)
+
+
+class TestBoltPedalIdentity(unittest.TestCase):
+  def test_pedal_requires_platform_observation_and_saved_opt_in(self):
+    for candidate in PEDAL_BOLT_CAR:
+      with self.subTest(candidate=candidate):
+        for setting, observed, missing in ((False, True, False), (True, False, False), (False, True, True)):
+          cp = params(candidate, setting, observed, alpha_long=True, missing_key=missing)
+          self.assertTrue(cp.openpilotLongitudinalControl)
+          self.assertFalse(cp.flags & GMFlags.PEDAL_LONG.value)
+          self.assertFalse(cp.safetyConfigs[0].safetyParam & GMSafetyFlags.PEDAL_LONG.value)
+
+        cp = params(candidate, True, True)
+        self.assertTrue(cp.openpilotLongitudinalControl)
+        self.assertTrue(cp.flags & GMFlags.PEDAL_LONG.value)
+        self.assertTrue(cp.safetyConfigs[0].safetyParam & GMSafetyFlags.PEDAL_LONG.value)
+        self.assertFalse(cp.pcmCruise)
+
+  def test_ordinary_cruise_ownership_is_never_stock_acc(self):
+    for candidate in NO_ACC_BOLT_CAR:
+      for enabled in (False, True):
+        with self.subTest(candidate=candidate, enabled=enabled):
+          cp = params(candidate, enabled, True)
+          self.assertFalse(cp.pcmCruise)
+          if enabled:
+            self.assertTrue(cp.safetyConfigs[0].safetyParam & GMSafetyFlags.NO_ACC.value)
+            self.assertFalse(cp.safetyConfigs[0].safetyParam & GMSafetyFlags.BOLT_ACC_PEDAL.value)
+          else:
+            self.assertEqual(cp.safetyConfigs[0].safetyParam, BOLT_CC_WORDS[candidate][1])
+            self.assertTrue(cp.flags & GMFlags.CC_LONG.value)
+
+  def test_existing_bolt_stock_acc_is_unaffected_by_pedal_setting(self):
+    without = params(CAR.CHEVROLET_BOLT_EUV, False, True)
+    with_setting = params(CAR.CHEVROLET_BOLT_EUV, True, True)
+    self.assertEqual(without.openpilotLongitudinalControl, with_setting.openpilotLongitudinalControl)
+    self.assertEqual(without.pcmCruise, with_setting.pcmCruise)
+    self.assertEqual(without.safetyConfigs[0].safetyParam, with_setting.safetyConfigs[0].safetyParam)
+    self.assertFalse(with_setting.flags & GMFlags.PEDAL_LONG.value)
+
+  def test_2022_pedal_variants_do_not_claim_automatic_fingerprint_identity(self):
+    from opendbc.car.gm.fingerprints import FINGERPRINTS, FW_VERSIONS
+    self.assertNotIn(CAR.CHEVROLET_BOLT_CC_2022_2023, FINGERPRINTS)
+    self.assertNotIn(CAR.CHEVROLET_BOLT_ACC_2022_2023_PEDAL, FINGERPRINTS)
+    self.assertFalse(FW_VERSIONS)
+
+
+class TestBoltPedalMessages(unittest.TestCase):
+  @staticmethod
+  def sensor(packer, gas, counter, state=0, other_track=None, bad_crc=False):
+    msg = packer.make_can_msg("GAS_SENSOR", 0, {
+      "INTERCEPTOR_GAS": gas,
+      "INTERCEPTOR_GAS2": gas if other_track is None else other_track,
+      "STATE": state, "COUNTER_PEDAL": counter,
+    })
+    data = bytearray(msg[1])
+    data[5] = pedal_crc(data) ^ int(bad_crc)
+    return msg[0], bytes(data), msg[2]
+
+  def test_real_parser_sensor_curve_and_fault_recovery(self):
+    for candidate in PEDAL_BOLT_CAR:
+      with self.subTest(candidate=candidate):
+        cp = params(candidate, True, True)
+        state = CarState(cp)
+        parsers = state.get_can_parsers(cp)
+        state.update(parsers)
+        packer = CANPacker(DBC[candidate][Bus.pt])
+        for counter, gas in enumerate((0., 4., 10., 23., 24., 30., 128., 255.), start=1):
+          parsers[Bus.pt].update([(1_000_000_000 + counter * 20_000_000,
+                                   [self.sensor(packer, gas, counter)])])
+          out = state.update(parsers)
+          self.assertTrue(state.pedal_sensor_healthy, gas)
+          self.assertEqual(out.gasPressed, gas > 23., gas)
+
+        for counter, kwargs in ((9, {"state": 1}), (10, {"other_track": 40.}), (11, {"bad_crc": True})):
+          parsers[Bus.pt].update([(1_000_000_000 + counter * 20_000_000,
+                                   [self.sensor(packer, 30., counter, **kwargs)])])
+          state.update(parsers)
+          self.assertFalse(state.pedal_sensor_healthy)
+        parsers[Bus.pt].update([(1_240_000_000, [self.sensor(packer, 30., 12)])])
+        state.update(parsers)
+        self.assertTrue(state.pedal_sensor_healthy)
+        parsers[Bus.pt].update([(1_260_000_000, [self.sensor(packer, 30., 12)])])
+        state.update(parsers)
+        self.assertFalse(state.pedal_sensor_healthy)
+
+  def test_controller_pedal_rate_release_and_stock_handoff(self):
+    for candidate in PEDAL_BOLT_CAR:
+      with self.subTest(candidate=candidate):
+        cp = params(candidate, True, True)
+        controller = CarController(DBC[candidate], cp)
+        controller.frame = 4
+        controller.last_steer_frame = 4
+        control = structs.CarControl()
+        control.enabled = True
+        control.longActive = True
+        control.actuators.accel = 1.
+        state = structs.CarState()
+        state.vEgo = 12.
+        state.aEgo = 0.
+        state.gearShifter = structs.CarState.GearShifter.low
+        state.cruiseState.available = True
+        cs = SimpleNamespace(out=state.as_reader(), pedal_sensor_healthy=True,
+                             pedal_sensor_ts_nanos=950_000_000, stock_acc_status_ts_nanos=950_000_000,
+                             cam_lka_steering_cmd_counter=0, loopback_lka_steering_cmd_updated=False,
+                             loopback_lka_steering_cmd_ts_nanos=1_000_000_000, pt_lka_steering_cmd_counter=0,
+                             buttons_counter=0)
+        _, messages = controller.update(control.as_reader(), cs, 1_000_000_000)
+        expected = [0x200, 0x1F5, 0xBD]
+        if candidate == CAR.CHEVROLET_BOLT_ACC_2022_2023_PEDAL:
+          expected.append(0x315)
+        self.assertEqual([m[0] for m in messages], expected)
+        self.assertGreater(controller.pedal_steady, 0.)
+        first_fraction = controller.pedal_steady
+
+        controller.frame = 8
+        control.actuators.accel = 2.
+        _, messages = controller.update(control.as_reader(), cs, 1_000_000_000)
+        self.assertEqual([m[0] for m in messages], expected)
+        self.assertGreater(controller.pedal_steady, first_fraction)
+        self.assertLess(controller.pedal_steady - first_fraction, 0.06)
+
+        controller.frame = 12
+        cs.pedal_sensor_healthy = False
+        _, messages = controller.update(control.as_reader(), cs, 1_000_000_000)
+        self.assertEqual(messages[0][1][0:4], b"\x00" * 4)
+        self.assertEqual(messages[1][1], b"\x0c\x0c\x00\x06\x00\x00\x01\x00")
+        self.assertEqual(messages[2][1], b"\x00" * 7)
+        self.assertEqual(controller.pedal_steady, 0.)
+
+        if candidate == CAR.CHEVROLET_BOLT_ACC_2022_2023_PEDAL:
+          controller.frame = 16
+          cs.pedal_sensor_healthy = True
+          state.cruiseState.enabled = True
+          cs.out = state.as_reader()
+          _, messages = controller.update(control.as_reader(), cs, 1_000_000_000)
+          self.assertEqual(messages[0][1][0:4], b"\x00" * 4)
+          self.assertIn((0x1E1, 2), [(m[0], m[2]) for m in messages])
+
+  def test_driver_gear_and_paddle_ownership(self):
+    for candidate in PEDAL_BOLT_CAR:
+      with self.subTest(candidate=candidate):
+        cp = params(candidate, True, True)
+        controller = CarController(DBC[candidate], cp)
+        control = structs.CarControl()
+        control.enabled = True
+        control.longActive = True
+        control.actuators.accel = 1.
+        state = structs.CarState()
+        state.vEgo = 12.
+        state.cruiseState.available = True
+        cs = SimpleNamespace(pedal_sensor_healthy=True, pedal_sensor_ts_nanos=950_000_000,
+                             stock_acc_status_ts_nanos=950_000_000, cam_lka_steering_cmd_counter=0,
+                             loopback_lka_steering_cmd_updated=False, loopback_lka_steering_cmd_ts_nanos=1_000_000_000,
+                             pt_lka_steering_cmd_counter=0, buttons_counter=0)
+        for gear, driver_paddle in ((structs.CarState.GearShifter.park, False),
+                                    (structs.CarState.GearShifter.reverse, False),
+                                    (structs.CarState.GearShifter.drive, False),
+                                    (structs.CarState.GearShifter.manumatic, False),
+                                    (structs.CarState.GearShifter.low, True)):
+          controller.frame = 4
+          state.gearShifter = gear
+          state.regenBraking = driver_paddle
+          cs.out = state.as_reader()
+          _, commands = controller.update(control.as_reader(), cs, 1_000_000_000)
+          owned = [msg for msg in commands if msg[0] in (0x200, 0x1F5, 0xBD)]
+          self.assertEqual([msg[0] for msg in owned], [0x200])
+          self.assertEqual(owned[0][1][0:4], b"\x00" * 4)
+        controller.frame = 4
+        state.gearShifter = structs.CarState.GearShifter.low
+        state.regenBraking = False
+        cs.out = state.as_reader()
+        _, commands = controller.update(control.as_reader(), cs, 1_000_000_000)
+        self.assertEqual([msg[0] for msg in commands if msg[0] in (0x200, 0x1F5, 0xBD)], [0x200, 0x1F5, 0xBD])

opendbc_repo/opendbc/car/gm/tests/test_conventional_pedal.py

@@ -0,0 +1,202 @@
+"""Explicit conventional-cruise interceptor startup and physical command owner."""
+import unittest
+from unittest.mock import patch
+
+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.gmcan import pedal_crc
+from opendbc.car.gm.values import CAR, DBC, ORDINARY_CC_CAR, is_conventional_cc_pedal_profile
+from opendbc.car.gm.startup_preferences import prepare_disable_longitudinal
+from opendbc.car.gm.tests.test_cc_gateway_stock import pt_frames
+
+
+class Settings:
+  def __init__(self, enabled):
+    self.enabled = enabled
+
+  def get_bool(self, key):
+    return self.enabled if key == 'GMPedalLongitudinal' else False
+
+
+def params(identity, *, disabled=False, removed=False, release=False, enabled=True, sensor=True, analog=True):
+  fp = gen_empty_fingerprint()
+  fp[0].update({0xF1: 6, 0xBE: 6, 0xC9: 8, 0x3D1: 8, 0x1E1: 7})
+  if not analog:
+    fp[0].pop(0xBE)
+  if sensor:
+    fp[0][0x201] = 6
+  if not removed:
+    fp[2][0x320] = 6
+  with patch('opendbc.car.gm.interface.Params', return_value=Settings(enabled)):
+    cp = CarInterface.get_params(identity, fp, [], False, release, False)
+  prepare_disable_longitudinal(cp, disabled)
+  return cp
+
+
+def pedal_frames(packer, counter, pressed=False, *, removed=False, bad_crc=False):
+  sensor = packer.make_can_msg('GAS_SENSOR', 0, {
+    'INTERCEPTOR_GAS': 30 if pressed else 0, 'INTERCEPTOR_GAS2': 30 if pressed else 0,
+    'STATE': 0, 'COUNTER_PEDAL': counter % 16,
+  })
+  data = bytearray(sensor[1])
+  data[5] = pedal_crc(data) ^ int(bad_crc)
+  frames = [(sensor[0], bytes(data), sensor[2]),
+            packer.make_can_msg("ASCMLKASteeringCmd", 128, {"RollingCounter": counter % 4}),
+            packer.make_can_msg('EBCMBrakePedalPosition', 0, {'BrakePedalPosition': 0})]
+  if not removed:
+    frames += [packer.make_can_msg('ASCMLKASteeringCmd', 2, {'RollingCounter': counter % 4}),
+               packer.make_can_msg('AEBCmd', 2, {})]
+  return frames
+
+
+class TestConventionalPedal(unittest.TestCase):
+  def test_explicit_startup_profile_and_disabled_policy(self):
+    from opendbc.car.gm.conventional_pedal import policy_for
+    for identity in ORDINARY_CC_CAR:
+      for removed in (False, True):
+        for release in (False, True):
+          for disabled in (False, True):
+            cp = params(identity, removed=removed, release=release, disabled=disabled)
+            self.assertTrue(is_conventional_cc_pedal_profile(cp))
+            self.assertFalse(cp.pcmCruise)
+            self.assertEqual(cp.openpilotLongitudinalControl, not disabled)
+            self.assertEqual(cp.safetyConfigs[0].safetyParam, (0xC186 if disabled else 0xC180) + int(removed))
+            self.assertEqual(cp.minEnableSpeed, -1.)
+            self.assertEqual(policy_for(cp) is not None, not disabled)
+            if not disabled:
+              self.assertEqual(policy_for(cp).starting_speed, .75 if identity == CAR.CHEVROLET_MALIBU_CC else .5)
+            self.assertEqual(cp.stopAccel, -1.5 if identity == CAR.CHEVROLET_MALIBU_CC else -2.)
+        self.assertFalse(is_conventional_cc_pedal_profile(params(identity, removed=removed, enabled=False)))
+        self.assertFalse(is_conventional_cc_pedal_profile(params(identity, removed=removed, sensor=False)))
+
+  def test_actual_sensor_threshold_and_stock_source_are_independent(self):
+    for identity in ORDINARY_CC_CAR:
+      cp = params(identity)
+      ci = CarInterface(cp)
+      packer = CANPacker(DBC[identity][Bus.pt])
+      for tick in range(24):
+        now = 1_000_000_000 + tick * 10_000_000
+        frames = pt_frames(packer, cruise=False, acc_cruise=2, counter=tick % 4)
+        frames += pedal_frames(packer, tick, pressed=tick >= 12)
+        out = ci.update([(now, frames)])
+        self.assertTrue(out.canValid)
+        self.assertFalse(out.cruiseState.enabled)
+        self.assertFalse(out.cruiseState.nonAdaptive)
+        self.assertEqual(out.gasPressed, tick >= 12)
+        self.assertTrue(ci.CS.pedal_sensor_healthy)
+
+  def test_actual_pedal_counter_cancel_and_disabled_takeover(self):
+    for identity in ORDINARY_CC_CAR:
+      for removed in (False, True):
+        for disabled in (False, True):
+          cp = params(identity, removed=removed, disabled=disabled)
+          ci = CarInterface(cp)
+          packer = CANPacker(DBC[identity][Bus.pt])
+          seen_cancel = False
+          for tick in range(40):
+            now = 1_000_000_000 + tick * 10_000_000
+            cruise = tick < 25
+            frames = pt_frames(packer, cruise=cruise, counter=tick % 4)
+            frames += pedal_frames(packer, tick, removed=removed)
+            out = ci.update([(now, frames)])
+            self.assertTrue(out.canValid)
+            cc = structs.CarControl(enabled=True, latActive=True, longActive=True)
+            cc.actuators.torque = .1
+            cc.actuators.accel = 1.
+            cc.actuators.longControlState = structs.CarControl.Actuators.LongControlState.pid
+            _, messages = ci.apply(cc.as_reader(), now)
+            self.assertFalse(any(message[0] in (0x2CB, 0x315, 0x370, 0x2CD, 0x3D1, 0xBD, 0x1F5) for message in messages))
+            pedal = [message for message in messages if message[0] == 0x200]
+            cancel = [message for message in messages if message[0] == 0x1E1]
+            if disabled:
+              self.assertFalse(pedal or cancel)
+            else:
+              if pedal:
+                self.assertEqual(pedal[0][2], 0)
+                self.assertEqual(pedal[0][1][4] & 0xF, (tick // 4) % 4)
+              if cancel:
+                seen_cancel = True
+                self.assertTrue(cruise)
+                self.assertEqual(cancel[0][2], 0)
+                if any(message[0] == 0x184 for message in messages):
+                  ids = [message[0] for message in messages]
+                  self.assertLess(ids.index(0x1E1), ids.index(0x184))
+            self.assertEqual(any(message[0] == 0x184 for message in messages), tick % 10 == 0)
+          if not disabled:
+            self.assertTrue(seen_cancel)
+
+  def test_sensor_crc_and_timeout_clear_actual_pedal_output(self):
+    identity = next(iter(ORDINARY_CC_CAR))
+    cp = params(identity, removed=True)
+    ci = CarInterface(cp)
+    packer = CANPacker(DBC[identity][Bus.pt])
+    cc = structs.CarControl(enabled=True, latActive=True, longActive=True)
+    cc.actuators.accel = 1.
+    for tick in range(24):
+      now = 1_000_000_000 + tick * 10_000_000
+      ci.update([(now, pt_frames(packer, cruise=False, counter=tick % 4) + pedal_frames(packer, tick, removed=True))])
+      ci.apply(cc.as_reader(), now)
+    self.assertGreater(ci.CC.pedal_steady, 0)
+    ci.update([(now + 10_000_000, pt_frames(packer, cruise=False) + pedal_frames(packer, 24, removed=True, bad_crc=True))])
+    ci.CC.frame = 24
+    ci.apply(cc.as_reader(), now + 10_000_000)
+    self.assertEqual(ci.CC.pedal_steady, 0)
+
+  def test_actual_packed_commands_match_registered_native_owner(self):
+    from opendbc.car.gm.gmcan import create_buttons
+    from opendbc.car.gm.tests.test_bolt_cc import setup, native
+    from opendbc.safety.tests.libsafety import libsafety_py
+    safety = libsafety_py.libsafety
+    for removed in (False, True):
+      cp = params(CAR.CADILLAC_CT6_CC, removed=removed)
+      ci = CarInterface(cp)
+      packer = CANPacker(DBC[cp.carFingerprint][Bus.pt])
+      setup(cp)
+      for tick in range(24):
+        now = 1_000_000_000 + tick * 20_000_000
+        frames = pt_frames(packer, cruise=tick < 12, counter=tick % 4)
+        frames = [message for message in frames if message[0] != 0x1E1]
+        frames += pedal_frames(packer, tick, removed=removed)
+        frames.append(create_buttons(packer, 0, tick % 4, 1))
+        out = ci.update([(now, frames)])
+        self.assertTrue(out.canValid)
+        for message in frames:
+          if message[2] != 128:  # Host loopback is not a physical CAN bus.
+            native('rx', message, now // 1000)
+        if tick == 1:
+          native('rx', create_buttons(packer, 0, tick % 4, 2), now // 1000)
+        if tick == 2:
+          native('rx', create_buttons(packer, 0, tick % 4, 1), now // 1000)
+        safety.safety_tick_current_safety_config()
+        self.assertTrue(safety.safety_config_valid())
+        cc = structs.CarControl(enabled=True, latActive=True, longActive=tick >= 3)
+        cc.actuators.accel = 1.
+        cc.actuators.torque = .01
+        cc.actuators.longControlState = structs.CarControl.Actuators.LongControlState.pid
+        _, emitted = ci.apply(cc.as_reader(), now)
+        for message in emitted:
+          self.assertTrue(native('tx', message, now // 1000), hex(message[0]))
+
+  def test_malibu_selected_analog_brake_source_and_alternate_threshold(self):
+    from opendbc.car.gm.values import GMFlags
+    for analog in (False, True):
+      for removed in (False, True):
+        cp = params(CAR.CHEVROLET_MALIBU_CC, analog=analog, removed=removed)
+        self.assertTrue(is_conventional_cc_pedal_profile(cp))
+        self.assertEqual(bool(cp.flags & GMFlags.NO_ACCELERATOR_POS_MSG), not analog)
+        ci = CarInterface(cp)
+        packer = CANPacker(DBC[cp.carFingerprint][Bus.pt])
+        for tick in range(32):
+          pressed = tick >= 16
+          frames = [m for m in pt_frames(packer, cruise=False, brake=not pressed, counter=tick % 4)
+                    if m[0] not in (0xBE, 0xF1)]
+          frames += pedal_frames(packer, tick, removed=removed)
+          frames = [m for m in frames if m[0] != 0xF1]
+          frames += [packer.make_can_msg('EBCMBrakePedalPosition', 0,
+                     {'BrakePedalPosition': 21 if pressed else 20})]
+          if analog:
+            frames += [packer.make_can_msg('ECMAcceleratorPos', 0, {'BrakePedalPos': 8 if pressed else 7})]
+          out = ci.update([(1_000_000_000 + tick * 10_000_000, frames)])
+          self.assertTrue(out.canValid)
+          self.assertEqual(out.brakePressed, pressed)

opendbc_repo/opendbc/car/gm/tests/test_conventional_pedal_transitions.py

@@ -0,0 +1,124 @@
+"""GM pedal state and caller transition boundaries."""
+
+import unittest
+from types import SimpleNamespace
+
+from opendbc.can import CANPacker
+from opendbc.car import Bus, structs
+from opendbc.car.gm.gmcan import create_pedal_command, pedal_crc
+from opendbc.car.gm.interface import CarInterface
+from opendbc.car.gm.conventional_pedal import policy_for
+from opendbc.car.gm.values import CAR, DBC, ORDINARY_CC_CAR
+from opendbc.car.gm.tests.test_cc_gateway_stock import pt_frames
+from opendbc.car.gm.tests.test_conventional_pedal import params, pedal_frames
+
+
+class TestConventionalPedalTransitions(unittest.TestCase):
+  def test_actual_decoded_sensor_threshold_and_cruise_standstill(self):
+    for identity in ORDINARY_CC_CAR:
+      for removed in (False, True):
+        for analog in (False, True):
+          cp = params(identity, removed=removed, analog=analog)
+          ci = CarInterface(cp)
+          packer = CANPacker(DBC[identity][Bus.pt])
+          for tick in range(24):
+            level = (22., 23., 24.)[tick // 8]
+            cruise = tick % 2 == 0
+            frames = pt_frames(packer, cruise=cruise, acc_cruise=4, counter=tick % 4)
+            if not analog:
+              frames = [f for f in frames if f[0] != 0xBE]
+            frames += pedal_frames(packer, tick, removed=removed)
+            frames = [f for f in frames if f[0] != 0x201]
+            sensor = packer.make_can_msg('GAS_SENSOR', 0, {'INTERCEPTOR_GAS': level,
+                     'INTERCEPTOR_GAS2': level, 'STATE': 0, 'COUNTER_PEDAL': tick % 16})
+            raw = bytearray(sensor[1])
+            raw[-1] = pedal_crc(raw)
+            frames.append((sensor[0], bytes(raw), sensor[2]))
+            out = ci.update([(1_000_000_000 + tick * 10_000_000, frames)])
+            self.assertTrue(out.canValid)
+            self.assertTrue(ci.CS.pedal_sensor_healthy)
+            decoded = ci.can_parsers[Bus.pt].vl['GAS_SENSOR']
+            mean = (decoded['INTERCEPTOR_GAS'] + decoded['INTERCEPTOR_GAS2']) / 2.
+            self.assertEqual(out.gasPressed, mean > 23.)
+            self.assertEqual(mean > 23., level == 24.)
+            self.assertEqual(out.cruiseState.enabled, cruise)
+            self.assertTrue(out.cruiseState.standstill)
+
+  @staticmethod
+  def caller(identity, removed):
+    cp = params(identity, removed=removed)
+    ci = CarInterface(cp)
+    packer = CANPacker(DBC[identity][Bus.pt])
+    counter = [0]
+    def step(accel=1.8, *, speed=0., active=True, stopping=False, resume=False, main=True, bad_crc=False, enabled=None):
+      tick = counter[0]
+      counter[0] += 1
+      frames = pt_frames(packer, cruise=True, acc_cruise=4, counter=tick % 4, main=main)
+      frames += pedal_frames(packer, tick, removed=removed, bad_crc=bad_crc)
+      now = 1_000_000_000 + tick * 40_000_000
+      out = ci.update([(now, frames)]).as_reader().as_builder()
+      # Caller boundary fixture preserves actual parser freshness and cruise inputs.
+      out.vEgo = speed
+      out.standstill = speed == 0.
+      ci.CS.out = out.as_reader()
+      cc = structs.CarControl(enabled=active if enabled is None else enabled, latActive=True, longActive=active)
+      cc.actuators.accel = accel
+      cc.actuators.longControlState = (structs.CarControl.Actuators.LongControlState.stopping if stopping else
+                                      structs.CarControl.Actuators.LongControlState.pid)
+      cc.cruiseControl.resume = resume
+      ci.CC.frame = tick * 4
+      _, messages = ci.apply(cc.as_reader(), now)
+      return next(m for m in messages if m[0] == 0x200), tick % 4
+    for _ in range(24):
+      step()
+    return SimpleNamespace(cp=cp, ci=ci, packer=packer, step=step)
+
+  def test_reached_stored_stop_limits_and_inactive_memory(self):
+    for identity in ORDINARY_CC_CAR:
+      for removed in (False, True):
+        case = self.caller(identity, removed)
+        malibu = identity == CAR.CHEVROLET_MALIBU_CC
+        self.assertEqual((case.cp.stopAccel, case.ci.CC.conventional_pedal_command.start_speed,
+                          case.ci.CC.conventional_pedal_command.stop_speed),
+                         (-1.5, .75, .75) if malibu else (-2., .5, .5))
+        self.assertEqual(case.ci.CC.conventional_pedal_command.start_speed, policy_for(case.cp).starting_speed)
+        steady, active = case.ci.CC.pedal_steady, case.ci.CC.pedal_active_last
+        emitted, counter = case.step(active=True, enabled=False)
+        self.assertEqual(emitted, create_pedal_command(case.packer, 0., counter))
+        self.assertEqual((case.ci.CC.pedal_steady, case.ci.CC.pedal_active_last), (steady, active))
+        emitted, counter = case.step(active=False)
+        self.assertEqual(emitted, create_pedal_command(case.packer, 0., counter))
+        self.assertEqual((case.ci.CC.pedal_steady, case.ci.CC.pedal_active_last), (steady, active))
+        emitted, counter = case.step(speed=.24, accel=-1., stopping=True)
+        self.assertEqual(emitted, create_pedal_command(case.packer, 0., counter))
+        self.assertEqual(case.ci.CC.apply_brake, 150 if malibu else 200)
+        self.assertEqual((case.ci.CC.pedal_steady, case.ci.CC.pedal_active_last), (steady, active))
+        emitted, counter = case.step(speed=.25, accel=-1., stopping=True)
+        self.assertNotEqual(emitted, create_pedal_command(case.packer, 0., counter))
+        self.assertLess(case.ci.CC.pedal_steady, steady)
+        emitted, counter = case.step(speed=.24, accel=1.8, stopping=True, resume=True)
+        self.assertNotEqual(emitted, create_pedal_command(case.packer, 0., counter))
+        self.assertGreater(case.ci.CC.pedal_steady, steady)
+        emitted, counter = case.step(speed=.24, accel=1.8)
+        self.assertEqual(emitted, create_pedal_command(case.packer, 18. / 255., counter))
+        emitted, counter = case.step(speed=.6, accel=1.8)
+        if malibu:
+          self.assertEqual(emitted, create_pedal_command(case.packer, 18. / 255., counter))
+        else:
+          self.assertNotEqual(emitted, create_pedal_command(case.packer, 18. / 255., counter))
+
+  def test_invalid_sensor_recovery_bounds_final_sng_and_veto_zero(self):
+    for identity in ORDINARY_CC_CAR:
+      for removed in (False, True):
+        case = self.caller(identity, removed)
+        emitted, counter = case.step(bad_crc=True)
+        self.assertEqual(emitted, create_pedal_command(case.packer, 0., counter))
+        self.assertEqual((case.ci.CC.pedal_steady, case.ci.CC.pedal_active_last), (0., True))
+        emitted, counter = case.step()
+        self.assertEqual(emitted, create_pedal_command(case.packer, .0229, counter))
+        emitted, counter = case.step(active=False)
+        self.assertEqual(emitted, create_pedal_command(case.packer, 0., counter))
+        emitted, counter = case.step()
+        self.assertEqual(emitted, create_pedal_command(case.packer, .0229, counter))
+        emitted, counter = case.step(active=False, main=False)
+        self.assertEqual(emitted, create_pedal_command(case.packer, 0., counter))

opendbc_repo/opendbc/safety/modes/gm_cc_pedal.h

@@ -0,0 +1,266 @@
+#pragma once
+
+static uint8_t gm_pedal_crc(const CANPacket_t *msg);
+
+static bool gm_cc_pedal = false;
+static bool gm_cc_pedal_volt;
+static bool gm_cc_pedal_volt_f1;
+static bool gm_cc_pedal_volt_removed;
+static bool gm_cc_pedal_volt_stock;
+static bool gm_cc_pedal_volt_rejected;
+static uint16_t gm_cc_pedal_volt_seen;
+static uint32_t gm_cc_pedal_volt_us[10] = {0U, 0U, 0U, 0U, 0U, 0U, 0U, 0U, 0U, 0U};
+static bool gm_cc_pedal_volt_regen;
+
+static uint16_t gm_cc_pedal_volt_decode(uint16_t raw) {
+  gm_cc_pedal_volt = false;
+  gm_cc_pedal_volt_f1 = false;
+  gm_cc_pedal_volt_removed = false;
+  gm_cc_pedal_volt_stock = false;
+  gm_cc_pedal_volt_rejected = false;
+  uint16_t canonical = raw;
+  switch (raw) {
+    case 0xE600U: gm_cc_pedal_volt = true; gm_cc_pedal_volt_f1 = false; gm_cc_pedal_volt_removed = false; gm_cc_pedal_volt_stock = false; canonical = 5U; break;
+    case 0xE601U: gm_cc_pedal_volt = true; gm_cc_pedal_volt_f1 = true; gm_cc_pedal_volt_removed = false; gm_cc_pedal_volt_stock = false; canonical = 5U; break;
+    case 0xE602U: gm_cc_pedal_volt = true; gm_cc_pedal_volt_f1 = false; gm_cc_pedal_volt_removed = true; gm_cc_pedal_volt_stock = false; canonical = 5U; break;
+    case 0xE603U: gm_cc_pedal_volt = true; gm_cc_pedal_volt_f1 = true; gm_cc_pedal_volt_removed = true; gm_cc_pedal_volt_stock = false; canonical = 5U; break;
+    case 0xE604U: gm_cc_pedal_volt = true; gm_cc_pedal_volt_f1 = false; gm_cc_pedal_volt_removed = false; gm_cc_pedal_volt_stock = false; canonical = 5U; break;
+    case 0xE605U: gm_cc_pedal_volt = true; gm_cc_pedal_volt_f1 = true; gm_cc_pedal_volt_removed = false; gm_cc_pedal_volt_stock = false; canonical = 5U; break;
+    case 0xE606U: gm_cc_pedal_volt = true; gm_cc_pedal_volt_f1 = false; gm_cc_pedal_volt_removed = true; gm_cc_pedal_volt_stock = false; canonical = 5U; break;
+    case 0xE607U: gm_cc_pedal_volt = true; gm_cc_pedal_volt_f1 = true; gm_cc_pedal_volt_removed = true; gm_cc_pedal_volt_stock = false; canonical = 5U; break;
+    case 0xE610U: gm_cc_pedal_volt = true; gm_cc_pedal_volt_f1 = false; gm_cc_pedal_volt_removed = false; gm_cc_pedal_volt_stock = true; canonical = 5U; break;
+    case 0xE611U: gm_cc_pedal_volt = true; gm_cc_pedal_volt_f1 = true; gm_cc_pedal_volt_removed = false; gm_cc_pedal_volt_stock = true; canonical = 5U; break;
+    case 0xE612U: gm_cc_pedal_volt = true; gm_cc_pedal_volt_f1 = false; gm_cc_pedal_volt_removed = true; gm_cc_pedal_volt_stock = true; canonical = 5U; break;
+    case 0xE613U: gm_cc_pedal_volt = true; gm_cc_pedal_volt_f1 = true; gm_cc_pedal_volt_removed = true; gm_cc_pedal_volt_stock = true; canonical = 5U; break;
+    case 0xE614U: gm_cc_pedal_volt = true; gm_cc_pedal_volt_f1 = false; gm_cc_pedal_volt_removed = false; gm_cc_pedal_volt_stock = true; canonical = 5U; break;
+    case 0xE615U: gm_cc_pedal_volt = true; gm_cc_pedal_volt_f1 = true; gm_cc_pedal_volt_removed = false; gm_cc_pedal_volt_stock = true; canonical = 5U; break;
+    case 0xE616U: gm_cc_pedal_volt = true; gm_cc_pedal_volt_f1 = false; gm_cc_pedal_volt_removed = true; gm_cc_pedal_volt_stock = true; canonical = 5U; break;
+    case 0xE617U: gm_cc_pedal_volt = true; gm_cc_pedal_volt_f1 = true; gm_cc_pedal_volt_removed = true; gm_cc_pedal_volt_stock = true; canonical = 5U; break;
+    default:
+      if ((raw >= 0xE600U) && (raw <= 0xE6FFU)) { gm_cc_pedal_volt_rejected = true; canonical = 5U; }
+      break;
+  }
+#ifndef ALLOW_DEBUG
+  if (gm_cc_pedal_volt && !gm_cc_pedal_volt_stock) { gm_cc_pedal_volt_rejected = true; }
+#endif
+  return canonical;
+}
+
+static bool gm_cc_pedal_volt_current(void) {
+  bool current = true;
+  const uint32_t now = microsecond_timer_get();
+  const uint32_t limits[10] = {300000U, 100000U, 300000U, 300000U, 300000U, 300000U, 100000U, 300000U, 1000000U, 100000U};
+  for (uint8_t i = 0U; i < 10U; i++) {
+    if (!gm_cc_pedal_volt_stock || ((i != 6U) && (i != 8U))) {
+      current &= ((gm_cc_pedal_volt_seen & (1U << i)) != 0U) && (safety_get_ts_elapsed(now, gm_cc_pedal_volt_us[i]) <= limits[i]);
+    }
+  }
+  return current && !safety_rx_checks_invalid && !relay_malfunction;
+}
+// Silverado owns four complete selectors; it shares only the checked pedal codec.
+static bool gm_cc_pedal_silverado = false;
+static bool gm_cc_pedal_stock_only = false;
+static bool gm_cc_pedal_ordinary_stock = false;
+static uint8_t gm_cc_pedal_silverado_seen = 0U;
+static uint32_t gm_cc_pedal_silverado_us[8] = {0U, 0U, 0U, 0U, 0U, 0U, 0U, 0U};
+static bool gm_cc_pedal_main = false;
+static bool gm_cc_pedal_stock_active = false;
+static bool gm_cc_pedal_forward = false;
+static bool gm_cc_pedal_sensor_valid = false;
+static bool gm_cc_pedal_sensor_gas = false;
+static bool gm_cc_pedal_sensor_seen = false;
+static uint8_t gm_cc_pedal_sensor_counter = 0U;
+static uint32_t gm_cc_pedal_sensor_us = 0U;
+static uint32_t gm_cc_pedal_main_us = 0U;
+static uint32_t gm_cc_pedal_stock_us = 0U;
+static uint32_t gm_cc_pedal_gear_us = 0U;
+static bool gm_cc_pedal_button_seen = false;
+static bool gm_cc_pedal_cancel_credit = false;
+static uint8_t gm_cc_pedal_button_counter = 0U;
+static uint32_t gm_cc_pedal_button_us = 0U;
+static bool gm_cc_pedal_cancel_seen = false;
+static uint32_t gm_cc_pedal_cancel_us = 0U;
+static bool gm_cc_pedal_tx_seen = false;
+static uint8_t gm_cc_pedal_tx_counter = 0U;
+
+static void gm_cc_pedal_reset(void) {
+  gm_cc_pedal_volt_seen = 0U; gm_cc_pedal_volt_regen = false;
+  for (uint8_t i = 0U; i < 10U; i++) { gm_cc_pedal_volt_us[i] = 0U; }
+  gm_cc_pedal_silverado_seen = 0U;
+  for (uint8_t i = 0U; i < 8U; i++) { gm_cc_pedal_silverado_us[i] = 0U; }
+  gm_cc_pedal_main = false;
+  gm_cc_pedal_stock_active = false;
+  gm_cc_pedal_forward = false;
+  gm_cc_pedal_sensor_valid = false;
+  gm_cc_pedal_sensor_gas = false;
+  gm_cc_pedal_sensor_seen = false;
+  gm_cc_pedal_sensor_counter = 0U;
+  gm_cc_pedal_sensor_us = 0U;
+  gm_cc_pedal_main_us = 0U;
+  gm_cc_pedal_stock_us = 0U;
+  gm_cc_pedal_gear_us = 0U;
+  gm_cc_pedal_button_seen = false;
+  gm_cc_pedal_cancel_credit = false;
+  gm_cc_pedal_button_counter = 0U;
+  gm_cc_pedal_button_us = 0U;
+  gm_cc_pedal_cancel_seen = false;
+  gm_cc_pedal_cancel_us = 0U;
+  gm_cc_pedal_tx_seen = false;
+  gm_cc_pedal_tx_counter = 0U;
+}
+
+static bool gm_cc_pedal_silverado_current(void) {
+  bool current = gm_cc_pedal_silverado_seen == 0xFFU;
+  const uint32_t now = microsecond_timer_get();
+  const uint32_t limits[8] = {300000U, 300000U, 300000U, 300000U, 100000U, 100000U, 300000U, 100000U};
+  for (uint8_t i = 0U; i < 8U; i++) {
+    current &= safety_get_ts_elapsed(now, gm_cc_pedal_silverado_us[i]) <= limits[i];
+  }
+  return current && gm_cc_pedal_sensor_valid && !safety_rx_checks_invalid && !relay_malfunction;
+}
+
+static bool gm_cc_pedal_current(void) {
+  const uint32_t now = microsecond_timer_get();
+  return (!gm_cc_pedal_volt || gm_cc_pedal_volt_current()) && gm_cc_pedal_sensor_valid && gm_cc_pedal_main && gm_cc_pedal_forward &&
+         (safety_get_ts_elapsed(now, gm_cc_pedal_sensor_us) <= 100000U) &&
+         (safety_get_ts_elapsed(now, gm_cc_pedal_main_us) <= 300000U) &&
+         (safety_get_ts_elapsed(now, gm_cc_pedal_gear_us) <= (gm_cc_pedal_volt ? 1000000U : (gm_cc_pedal_silverado ? 300000U : 100000U))) &&
+         (!gm_cc_pedal_silverado || gm_cc_pedal_silverado_current()) &&
+         !safety_rx_checks_invalid && !relay_malfunction;
+}
+
+static void gm_cc_pedal_rx(const CANPacket_t *msg) {
+  if (gm_cc_pedal && (msg->bus == 0U)) {
+    const uint32_t now = microsecond_timer_get();
+    if (gm_cc_pedal_volt) {
+      const uint16_t source_bits[10] = {1U, 2U, 4U, 8U, 16U, 32U, 64U, 128U, 256U, 512U};
+      const uint32_t addresses[10] = {0x184U, 0x34AU, 0x1E1U, gm_cc_pedal_volt_f1 ? 0xF1U : 0xBEU, 0x1C4U, 0xC9U, 0x201U, 0x3D1U, 0x1F5U, 0xBDU};
+      const uint8_t lengths[10] = {8U, 5U, 7U, 6U, 8U, 8U, 6U, 8U, 8U, 7U};
+      for (uint8_t i = 0U; i < 10U; i++) {
+        if ((msg->addr == addresses[i]) && ((GET_LEN(msg) == lengths[i]) || ((i == 3U) && !gm_cc_pedal_volt_f1 && ((GET_LEN(msg) == 7U) || (GET_LEN(msg) == 8U))))) {
+          gm_cc_pedal_volt_seen |= source_bits[i]; gm_cc_pedal_volt_us[i] = now;
+        }
+      }
+      if ((msg->addr == 0xBDU) && (GET_LEN(msg) == 7U)) { gm_cc_pedal_volt_regen = (msg->data[0] >> 4) != 0U; }
+    }
+    if (gm_cc_pedal_silverado) {
+      const uint32_t addresses[8] = {0x184U, 0x34AU, 0xC9U, 0x3D1U, 0x1E1U, 0x1C4U, 0x1F5U, 0x201U};
+      const uint8_t lengths[8] = {8U, 5U, 8U, 8U, 7U, 8U, 8U, 6U};
+      for (uint8_t i = 0U; i < 8U; i++) {
+        if ((msg->addr == addresses[i]) && (GET_LEN(msg) == lengths[i])) {
+          gm_cc_pedal_silverado_seen |= (uint8_t)(1U << i);
+          gm_cc_pedal_silverado_us[i] = now;
+        }
+      }
+    }
+    if ((msg->addr == 0x201U) && (GET_LEN(msg) == 6U)) {
+      // Incoming tracks are independent 12-bit ADC samples, not synthesized command pairs.
+      const int track1 = (msg->data[0] << 8) | msg->data[1];
+      const int track2 = (msg->data[2] << 8) | msg->data[3];
+      const uint8_t counter = msg->data[4] & 0xFU;
+      gm_cc_pedal_sensor_valid = ((msg->data[4] >> 4) == 0U) && (gm_pedal_crc(msg) == msg->data[5]) &&
+        (!gm_cc_pedal_sensor_seen || (counter != gm_cc_pedal_sensor_counter)) &&
+        (track1 <= 4095) && (track2 <= 4095);
+      gm_cc_pedal_sensor_seen = true;
+      gm_cc_pedal_sensor_counter = counter;
+      gm_cc_pedal_sensor_us = now;
+      if (gm_cc_pedal_volt) {
+        gm_cc_pedal_sensor_gas = ((125677U * (uint32_t)track1) + (251976U * (uint32_t)track2)) > (uint32_t)198510000U;
+      } else {
+        gm_cc_pedal_sensor_gas = ((track1 + track2) / 2) > 595;
+      }
+      if (!gm_cc_pedal_sensor_valid && !gm_cc_pedal_volt_stock) { controls_allowed = false; }
+    } else if ((msg->addr == 0x3D1U) && (GET_LEN(msg) == 8U)) {
+      gm_cc_pedal_stock_active = GET_BIT(msg, 39U);
+      gm_cc_pedal_stock_us = now;
+      cruise_engaged_prev = gm_cc_pedal_stock_active;
+      if (!gm_cc_pedal_stock_active && (!gm_cc_pedal_stock_only || gm_cc_pedal_ordinary_stock)) {
+        gm_cc_pedal_cancel_credit = false;
+      }
+    } else if ((msg->addr == 0xC9U) && (GET_LEN(msg) == 8U)) {
+      gm_cc_pedal_main = GET_BIT(msg, 29U);
+      gm_cc_pedal_main_us = now;
+      if (!gm_cc_pedal_main) { controls_allowed = false; }
+    } else if ((msg->addr == 0x1F5U) && (GET_LEN(msg) == 8U)) {
+      const uint8_t gear = msg->data[3] & 0xFU;
+      gm_cc_pedal_forward = ((gear == 4U) || (gear == 6U)) && !GET_BIT(msg, 41U);
+      gm_cc_pedal_gear_us = now;
+      if (!gm_cc_pedal_forward && !gm_cc_pedal_volt_stock) { controls_allowed = false; }
+    } else if ((msg->addr == 0x1E1U) && (GET_LEN(msg) == 7U)) {
+      const uint8_t counter = msg->data[4] & 0x3U;
+      const uint16_t checksum = 0xFFU + (counter * 0x4EFU);
+      const bool neutral = (msg->data[0] == 0U) && (msg->data[1] == 0U) && (msg->data[2] == 0U) &&
+        (msg->data[3] == 1U) && (msg->data[4] == counter) &&
+        (msg->data[5] == (uint8_t)(0x10U | (checksum >> 8))) && (msg->data[6] == (uint8_t)checksum);
+      const bool timely = safety_get_ts_elapsed(now, gm_cc_pedal_button_us) <= 300000U;
+      if (neutral && (!gm_cc_pedal_button_seen ||
+          (timely && (counter == ((gm_cc_pedal_button_counter + 1U) % 4U))))) {
+        gm_cc_pedal_cancel_credit = true;
+        gm_cc_pedal_button_us = now;
+      } else if (!neutral || !timely || (counter != gm_cc_pedal_button_counter)) {
+        gm_cc_pedal_cancel_credit = false;
+      } else {
+        // A duplicate neutral frame cannot replenish a consumed cancellation slot.
+      }
+      if (!gm_cc_pedal_button_seen || (counter != gm_cc_pedal_button_counter)) {
+        gm_cc_pedal_button_us = now;
+      }
+      gm_cc_pedal_button_seen = true;
+      gm_cc_pedal_button_counter = counter;
+    } else {
+      // Only exact selected PT sources change this owner state.
+    }
+    if (!gm_cc_pedal_volt_stock) { gas_pressed = gm_cc_pedal_sensor_gas; }
+    if (!gm_cc_pedal_main || (!gm_cc_pedal_volt_stock && (!gm_cc_pedal_forward || !gm_cc_pedal_sensor_valid))) { controls_allowed = false; }
+  }
+}
+
+static bool gm_cc_pedal_tx(const CANPacket_t *msg) {
+  bool allowed = true;
+  if (msg->addr == 0x200U) {
+    const int track1 = (msg->data[0] << 8) | msg->data[1];
+    const int track2 = (msg->data[2] << 8) | msg->data[3];
+    const int pair_delta = track1 - (2 * track2);
+    const uint8_t counter = msg->data[4] & 0xFU;
+    const bool enabled = GET_BIT(msg, 39U);
+    const bool inactive = !enabled && (track1 == 0) && (track2 == 0);
+    const bool active = enabled && gm_cc_pedal_current() && get_longitudinal_allowed() && !brake_pressed_prev && (!gm_cc_pedal_volt || (!gm_cc_pedal_volt_regen && !gm_cc_pedal_sensor_gas)) &&
+      (track1 >= 604) && (track1 <= 2633) && (track2 >= 304) && (track2 <= 1316) &&
+      (pair_delta >= -16) && (pair_delta <= 16);
+    allowed = !gm_cc_pedal_stock_only && (inactive || active) && ((msg->data[4] & 0x70U) == 0U) && (counter < 4U) &&
+      (gm_pedal_crc(msg) == msg->data[5]) &&
+      (!gm_cc_pedal_tx_seen || (gm_cc_pedal_volt ? (counter == ((gm_cc_pedal_tx_counter + 1U) & 3U)) : (counter != gm_cc_pedal_tx_counter)));
+    if (allowed) {
+      gm_cc_pedal_tx_seen = true;
+      gm_cc_pedal_tx_counter = counter;
+    }
+  } else if (msg->addr == 0x1E1U) {
+    const uint32_t now = microsecond_timer_get();
+    const uint8_t counter = gm_cc_pedal_stock_only ? gm_cc_pedal_button_counter :
+      (gm_cc_pedal_button_counter + 1U) % 4U;
+    const uint16_t checksum = 0xFFU + (counter * 0x4EFU) - (5U << 4);
+    allowed = gm_cc_pedal_main && ((gm_cc_pedal_stock_only && !gm_cc_pedal_ordinary_stock && !gm_cc_pedal_volt_stock) || gm_cc_pedal_stock_active) &&
+      gm_cc_pedal_cancel_credit && (!gm_cc_pedal_ordinary_stock || !safety_rx_checks_invalid) &&
+      (!gm_cc_pedal_silverado || gm_cc_pedal_silverado_current()) &&
+      (!gm_cc_pedal_volt || gm_cc_pedal_volt_current()) &&
+      (safety_get_ts_elapsed(now, gm_cc_pedal_main_us) <= 300000U) &&
+      (safety_get_ts_elapsed(now, gm_cc_pedal_stock_us) <= 300000U) &&
+      (safety_get_ts_elapsed(now, gm_cc_pedal_button_us) <= ((gm_cc_pedal_silverado || gm_cc_pedal_volt) ? 100000U : 300000U)) &&
+      (!gm_cc_pedal_cancel_seen || (safety_get_ts_elapsed(now, gm_cc_pedal_cancel_us) > 40000U)) &&
+      (msg->data[0] == 0U) && (msg->data[1] == 0U) && (msg->data[2] == 0U) &&
+      (msg->data[3] == 1U) && (msg->data[4] == counter) &&
+      (msg->data[5] == (uint8_t)(0x60U | (checksum >> 8))) && (msg->data[6] == (uint8_t)checksum);
+    if (allowed) {
+      gm_cc_pedal_cancel_credit = false;
+      gm_cc_pedal_cancel_seen = true;
+      gm_cc_pedal_cancel_us = now;
+    }
+  } else if ((msg->addr == 0x409U) || (msg->addr == 0x40AU)) {
+    for (uint8_t i = 0U; i < 7U; i++) {
+      allowed &= msg->data[i] == 0U;
+    }
+  } else {
+    // Other admitted messages retain the common GM checks.
+  }
+  return allowed;
+}

opendbc_repo/opendbc/safety/tests/test_gm_bolt_pedal.py

@@ -0,0 +1,402 @@
+import unittest
+from types import SimpleNamespace
+
+from opendbc.can import CANPacker
+from opendbc.car import Bus, structs
+from opendbc.car.gm.carcontroller import CarController
+from opendbc.car.gm.carstate import CarState
+from opendbc.car.gm.gmcan import create_bolt_regen_gear, create_bolt_regen_paddle, create_pedal_command, create_steering_control, pedal_crc
+from opendbc.car.gm.tests.test_bolt_pedal import params
+from opendbc.car.gm.values import CAR, DBC, GMSafetyFlags, PEDAL_BOLT_CAR
+from opendbc.car.structs import CarParams
+from opendbc.safety.tests.libsafety import libsafety_py
+from opendbc.safety.tests.libsafety.libsafety_py import ffi, new_CANPacket
+
+
+class TestGmBoltPedalSafety(unittest.TestCase):
+  def setUp(self):
+    self.packer = CANPacker(DBC[CAR.CHEVROLET_BOLT_CC_2017]["pt"])
+    self.safety = libsafety_py.libsafety
+    self.init_mode(GMSafetyFlags.NO_ACC | GMSafetyFlags.BOLT_2017)
+
+  def init_mode(self, variant, pedal=True):
+    param = int(GMSafetyFlags.HW_CAM | GMSafetyFlags.EV | variant)
+    if pedal:
+      param |= int(GMSafetyFlags.PEDAL_LONG | GMSafetyFlags.PADDLE_SCHED)
+    self.assertEqual(self.safety.set_safety_hooks(CarParams.SafetyModel.gm, param), 0)
+    self.safety.init_tests()
+    self.safety.reset_recorded_can()
+
+  @staticmethod
+  def packet(msg):
+    return libsafety_py.make_CANPacket(msg[0], msg[2], msg[1])
+
+  def sensor(self, counter, state=0, gas=0., pair_error=False, bad_crc=False):
+    msg = self.packer.make_can_msg("GAS_SENSOR", 0, {
+      "INTERCEPTOR_GAS": gas, "INTERCEPTOR_GAS2": gas,
+      "STATE": state, "COUNTER_PEDAL": counter,
+    })
+    data = bytearray(msg[1])
+    if pair_error:
+      data[2] += 8
+    data[5] = pedal_crc(data) ^ int(bad_crc)
+    return self.packet((msg[0], bytes(data), 0))
+
+  def stock(self, name, values):
+    return self.packet(self.packer.make_can_msg(name, 0, values))
+
+  def low_gear(self):
+    return self.stock("ECMPRDNL2", {"PRNDL2": 6, "ManualMode": 0})
+
+  def recorded(self):
+    result = []
+    for index in range(self.safety.get_recorded_can_count()):
+      packet = new_CANPacket()
+      self.assertTrue(self.safety.get_recorded_can(index, packet))
+      result.append((packet.addr, packet.bus, bytes(packet.data[0:7 if packet.addr == 0xBD else 8])))
+    return result
+
+  def test_paired_sensor_fault_checksum_counter_and_timeout(self):
+    self.safety.safety_rx_hook(self.low_gear())
+    self.safety.set_controls_allowed(True)
+    self.assertTrue(self.safety.safety_rx_hook(self.sensor(1)))
+    self.assertTrue(self.safety.safety_tx_hook(self.packet(create_pedal_command(self.packer, .2, 1))))
+    self.assertFalse(self.safety.safety_tx_hook(self.packet(create_pedal_command(self.packer, .2, 1))))
+
+    for counter, bad in enumerate((self.sensor(2, state=1), self.sensor(3, pair_error=True),
+                                   self.sensor(4, bad_crc=True), self.sensor(4)), start=2):
+      self.safety.set_controls_allowed(True)
+      self.safety.safety_rx_hook(bad)
+      self.assertFalse(self.safety.safety_tx_hook(self.packet(create_pedal_command(self.packer, .2, counter))))
+      self.assertTrue(self.safety.safety_tx_hook(self.packet(create_pedal_command(self.packer, 0., counter))))
+
+    self.safety.safety_rx_hook(self.sensor(5))
+    self.safety.set_controls_allowed(True)
+    self.safety.set_timer(100001)
+    self.assertFalse(self.safety.safety_tx_hook(self.packet(create_pedal_command(self.packer, .2, 6))))
+    self.assertTrue(self.safety.safety_tx_hook(self.packet(create_pedal_command(self.packer, 0., 6))))
+
+  def test_sensor_physical_curve_and_driver_override(self):
+    self.safety.safety_rx_hook(self.low_gear())
+    for counter, gas in enumerate((0., 4., 10., 23., 24., 30., 128., 255.), start=1):
+      with self.subTest(gas=gas):
+        self.assertTrue(self.safety.safety_rx_hook(self.sensor(counter, gas=gas)))
+        self.assertEqual(self.safety.get_gas_pressed_prev(), gas > 23.)
+        self.safety.set_controls_allowed(True)
+        self.assertEqual(self.safety.safety_tx_hook(self.packet(create_pedal_command(self.packer, .2, counter))),
+                         gas <= 23.)
+
+  def test_paddle_waits_for_stock_phase_and_expires_without_feed(self):
+    self.safety.safety_rx_hook(self.low_gear())
+    self.safety.safety_rx_hook(self.sensor(1))
+    self.safety.set_controls_allowed(True)
+    paddle = create_bolt_regen_paddle(self.packer, True)
+    gear = create_bolt_regen_gear(self.packer, True, False)
+    self.assertFalse(self.safety.safety_tx_hook(self.packet(paddle)))
+    self.assertFalse(self.safety.safety_tx_hook(self.packet(gear)))
+    self.assertEqual(self.recorded(), [])
+
+    stock_paddle = self.stock("EBCMRegenPaddle", {"RegenPaddle": 0})
+    stock_gear = self.stock("ECMPRDNL2", {"PRNDL2": 6})
+    self.safety.safety_rx_hook(stock_paddle)
+    self.safety.safety_rx_hook(stock_gear)
+    self.assertEqual([(addr, bus) for addr, bus, _ in self.recorded()], [(0xBD, 0), (0x1F5, 0)])
+    self.assertEqual(self.recorded()[0][2], paddle[1])
+    self.assertEqual(self.recorded()[1][2], gear[1])
+
+    self.safety.reset_recorded_can()
+    self.safety.set_timer(150000)
+    self.safety.safety_rx_hook(stock_paddle)
+    self.safety.safety_rx_hook(stock_gear)
+    self.assertEqual(self.recorded(), [])
+    self.safety.safety_rx_hook(self.sensor(2))
+    self.safety.set_controls_allowed(True)
+    self.safety.safety_rx_hook(self.low_gear())
+    self.assertFalse(self.safety.safety_tx_hook(self.packet(paddle)))
+    self.assertFalse(self.safety.safety_tx_hook(self.packet(gear)))
+    self.safety.safety_rx_hook(stock_paddle)
+    self.safety.safety_rx_hook(stock_gear)
+    self.assertEqual([(addr, bus) for addr, bus, _ in self.recorded()], [(0xBD, 0), (0x1F5, 0)])
+
+  def test_fault_and_driver_source_changes_consume_pending_spoofs(self):
+    self.safety.safety_rx_hook(self.sensor(1))
+    self.safety.safety_rx_hook(self.low_gear())
+    self.safety.set_controls_allowed(True)
+    applied_gear = self.packet(create_bolt_regen_gear(self.packer, True, False))
+    applied_paddle = self.packet(create_bolt_regen_paddle(self.packer, True))
+    stock_paddle = self.stock("EBCMRegenPaddle", {"RegenPaddle": 0})
+    self.assertFalse(self.safety.safety_tx_hook(applied_gear))
+    self.assertFalse(self.safety.safety_tx_hook(applied_paddle))
+    # A real brake edge clears the pending active request; regaining controls alone cannot replay it.
+    self.safety.safety_rx_hook(self.packet((0xC9, b"\x00" * 5 + b"\x01\x00\x00", 0)))
+    self.safety.safety_rx_hook(self.packet((0xC9, b"\x00" * 8, 0)))
+    self.safety.set_controls_allowed(True)
+    self.safety.safety_rx_hook(self.low_gear())
+    self.safety.safety_rx_hook(stock_paddle)
+    self.assertEqual(self.recorded(), [])
+    self.assertFalse(self.safety.safety_tx_hook(applied_gear))
+    self.assertFalse(self.safety.safety_tx_hook(applied_paddle))
+    self.safety.safety_rx_hook(self.low_gear())
+    self.safety.safety_rx_hook(stock_paddle)
+    self.assertEqual([addr for addr, _, _ in self.recorded()], [0x1F5, 0xBD])
+
+    for prndl, manual in ((1, 0), (2, 0), (4, 0), (6, 1)):
+      with self.subTest(prndl=prndl, manual=manual):
+        self.safety.reset_recorded_can()
+        self.safety.safety_rx_hook(self.low_gear())
+        self.safety.set_controls_allowed(True)
+        self.assertFalse(self.safety.safety_tx_hook(applied_gear))
+        self.safety.safety_rx_hook(self.stock("ECMPRDNL2", {"PRNDL2": prndl, "ManualMode": manual}))
+        self.assertEqual(self.recorded(), [])
+        self.assertFalse(self.safety.safety_tx_hook(self.packet(create_pedal_command(self.packer, .2, prndl))))
+        self.safety.safety_rx_hook(self.low_gear())
+        self.assertEqual(self.recorded(), [])
+
+    self.safety.reset_recorded_can()
+    self.safety.safety_rx_hook(self.low_gear())
+    self.safety.set_controls_allowed(True)
+    self.assertFalse(self.safety.safety_tx_hook(applied_paddle))
+    self.safety.safety_rx_hook(self.stock("EBCMRegenPaddle", {"RegenPaddle": 2}))
+    self.assertEqual(self.recorded(), [])
+    self.safety.safety_rx_hook(stock_paddle)
+    self.assertEqual(self.recorded(), [])
+    self.safety.set_controls_allowed(True)
+    self.assertFalse(self.safety.safety_tx_hook(applied_paddle))
+    self.safety.safety_rx_hook(stock_paddle)
+    self.assertEqual([addr for addr, _, _ in self.recorded()], [0xBD])
+
+  def test_owned_tx_only_and_stock_acc_sibling(self):
+    self.safety.safety_rx_hook(self.sensor(1))
+    self.safety.set_controls_allowed(True)
+    self.assertFalse(self.safety.safety_tx_hook(self.stock("ASCMGasRegenCmd", {"GasRegenCmd": 0})))
+    self.assertFalse(self.safety.safety_tx_hook(self.packet(self.packer.make_can_msg("ASCMSteeringButton", 2, {"ACCButtons": 6}))))
+
+    self.init_mode(GMSafetyFlags.BOLT_ACC_PEDAL)
+    self.safety.safety_rx_hook(self.sensor(1))
+    self.safety.safety_rx_hook(self.stock("AcceleratorPedal2", {"CruiseState": 1}))
+    self.assertTrue(self.safety.safety_tx_hook(self.packet(self.packer.make_can_msg("ASCMSteeringButton", 2, {"ACCButtons": 6}))))
+    self.assertFalse(self.safety.safety_tx_hook(self.packet(self.packer.make_can_msg("ASCMSteeringButton", 2, {"ACCButtons": 2}))))
+    self.assertFalse(self.safety.safety_tx_hook(self.stock("ASCMGasRegenCmd", {"GasRegenCmd": 0})))
+
+    self.assertEqual(self.safety.set_safety_hooks(CarParams.SafetyModel.gm, int(GMSafetyFlags.HW_CAM | GMSafetyFlags.EV)), 0)
+    self.safety.init_tests()
+    self.assertFalse(self.safety.safety_tx_hook(self.packet(create_pedal_command(self.packer, .2, 1))))
+
+  def test_acc_pedal_friction_requires_exclusive_fresh_owner(self):
+    chassis = CANPacker("gm_global_a_chassis")
+
+    def brake(value, counter, mode=None, bus=0):
+      mode = (1 if value == 0 else 0xA) if mode is None else mode
+      raw = (0x1000 - value) & 0xFFF
+      checksum = (0x10000 - (mode << 12) - raw - counter) & 0xFFFF
+      return self.packet(chassis.make_can_msg("EBCMFrictionBrakeCmd", bus, {
+        "FrictionBrakeCmd": -value, "FrictionBrakeMode": mode,
+        "RollingCounter": counter, "FrictionBrakeChecksum": checksum,
+      }))
+
+    self.init_mode(GMSafetyFlags.BOLT_ACC_PEDAL | GMSafetyFlags.BOLT_GEN2)
+    self.safety.set_controls_allowed(True)
+    self.assertFalse(self.safety.safety_tx_hook(brake(100, 0)))
+    self.assertFalse(self.safety.safety_tx_hook(brake(0, 0)))
+    self.assertEqual(self.safety.safety_fwd_hook(2, 0x315), 0)
+    self.assertEqual(self.safety.safety_fwd_hook(2, 0x2CB), 0)
+
+    self.safety.safety_rx_hook(self.low_gear())
+    self.safety.safety_rx_hook(self.sensor(1))
+    self.safety.safety_rx_hook(self.stock("AcceleratorPedal2", {"CruiseState": 0}))
+    self.assertEqual(self.safety.safety_fwd_hook(2, 0x315), -1)
+    self.safety.set_controls_allowed(True)
+    self.assertFalse(self.safety.safety_tx_hook(brake(100, 0)))
+    self.assertTrue(self.safety.safety_tx_hook(brake(0, 0)))
+    self.safety.safety_rx_hook(self.stock("ECMEngineStatus", {"CruiseMainOn": 1}))
+    self.assertTrue(self.safety.safety_tx_hook(brake(100, 1)))
+    self.assertFalse(self.safety.safety_tx_hook(brake(100, 1)))
+    self.assertFalse(self.safety.safety_tx_hook(brake(401, 2)))
+    self.assertTrue(self.safety.safety_tx_hook(brake(0, 2, mode=9)))
+    self.assertFalse(self.safety.safety_tx_hook(brake(100, 3, mode=1)))
+    self.assertFalse(self.safety.safety_tx_hook(brake(100, 3, bus=2)))
+    invalid = bytearray(chassis.make_can_msg("EBCMFrictionBrakeCmd", 0, {
+      "FrictionBrakeCmd": -100, "FrictionBrakeMode": 0xA,
+      "RollingCounter": 3, "FrictionBrakeChecksum": (0x10000 - (0xA << 12) - (0x1000 - 100) - 3) & 0xFFFF,
+    })[1])
+    invalid[3] ^= 1
+    self.assertFalse(self.safety.safety_tx_hook(self.packet((0x315, bytes(invalid), 0))))
+
+    self.safety.safety_rx_hook(self.stock("AcceleratorPedal2", {"CruiseState": 1}))
+    self.assertFalse(self.safety.safety_tx_hook(brake(100, 3)))
+    self.assertFalse(self.safety.safety_tx_hook(brake(0, 3, mode=9)))
+    self.assertFalse(self.safety.safety_tx_hook(brake(0, 3)))
+    self.assertEqual(self.safety.safety_fwd_hook(2, 0x315), 0)
+    self.safety.safety_rx_hook(self.stock("AcceleratorPedal2", {"CruiseState": 0}))
+    self.safety.safety_rx_hook(self.stock("ECMEngineStatus", {"CruiseMainOn": 0, "BrakePressed": 1}))
+    self.assertFalse(self.safety.safety_tx_hook(brake(100, 3)))
+    self.assertTrue(self.safety.safety_tx_hook(brake(0, 3)))
+    self.safety.set_timer(300001)
+    self.assertFalse(self.safety.safety_tx_hook(brake(100, 0)))
+    self.assertFalse(self.safety.safety_tx_hook(brake(0, 0)))
+    self.assertEqual(self.safety.safety_fwd_hook(2, 0x315), 0)
+
+    self.init_mode(GMSafetyFlags.NO_ACC | GMSafetyFlags.BOLT_GEN2)
+    self.assertFalse(self.safety.safety_tx_hook(brake(0, 0)))
+    self.assertEqual(self.safety.safety_fwd_hook(2, 0x315), 0)
+
+  def test_2017_steering_limit_is_conditional(self):
+    steer = self.packet(create_steering_control(self.packer, 0, 400, 0, True))
+    for variant, pedal, allowed in ((GMSafetyFlags.NO_ACC | GMSafetyFlags.BOLT_2017, True, True),
+                                    (GMSafetyFlags.NO_ACC | GMSafetyFlags.BOLT_2017, False, True),
+                                    (GMSafetyFlags.NO_ACC, True, False)):
+      with self.subTest(variant=variant, pedal=pedal):
+        self.init_mode(variant, pedal=pedal)
+        self.safety.set_controls_allowed(True)
+        self.safety.set_torque_driver(0, 0)
+        self.safety.set_desired_torque_last(400)
+        self.safety.set_rt_torque_last(400)
+        self.assertEqual(self.safety.safety_tx_hook(steer), allowed)
+
+  def test_stock_acc_ownership_requires_fresh_confirmed_disengagement(self):
+    self.init_mode(GMSafetyFlags.BOLT_ACC_PEDAL | GMSafetyFlags.BOLT_GEN2)
+    self.safety.safety_rx_hook(self.low_gear())
+
+    def active(counter):
+      return self.packet(create_pedal_command(self.packer, .2, counter))
+
+    def disabled(counter):
+      return self.packet(create_pedal_command(self.packer, 0., counter))
+    self.safety.safety_rx_hook(self.sensor(1))
+    self.safety.set_controls_allowed(True)
+    self.assertFalse(self.safety.safety_tx_hook(active(1)))  # startup status unknown
+    self.assertTrue(self.safety.safety_tx_hook(disabled(1)))
+
+    self.safety.safety_rx_hook(self.stock("AcceleratorPedal2", {"CruiseState": 0}))
+    self.safety.set_controls_allowed(True)
+    self.assertFalse(self.safety.safety_tx_hook(active(2)))  # main state unknown
+    self.safety.safety_rx_hook(self.stock("ECMEngineStatus", {"CruiseMainOn": 1}))
+    self.assertTrue(self.safety.safety_tx_hook(active(2)))
+    self.safety.safety_rx_hook(self.stock("AcceleratorPedal2", {"CruiseState": 1}))
+    self.assertFalse(self.safety.safety_tx_hook(active(3)))
+    self.assertFalse(self.safety.safety_tx_hook(self.packet(create_bolt_regen_paddle(self.packer, True))))
+    self.assertFalse(self.safety.safety_tx_hook(self.packet(create_bolt_regen_gear(self.packer, True, True))))
+    self.assertTrue(self.safety.safety_tx_hook(disabled(3)))
+    self.assertTrue(self.safety.safety_tx_hook(self.packet(self.packer.make_can_msg("ASCMSteeringButton", 2, {"ACCButtons": 6}))))
+
+    self.safety.safety_rx_hook(self.stock("AcceleratorPedal2", {"CruiseState": 0}))
+    self.safety.safety_rx_hook(self.sensor(2))
+    self.safety.set_controls_allowed(True)
+    self.assertTrue(self.safety.safety_tx_hook(active(4)))
+    self.safety.safety_rx_hook(self.stock("ECMEngineStatus", {"CruiseMainOn": 0}))
+    self.assertFalse(self.safety.get_controls_allowed())
+    self.safety.set_controls_allowed(True)
+    self.assertFalse(self.safety.safety_tx_hook(active(5)))
+    self.assertTrue(self.safety.safety_tx_hook(disabled(5)))
+    self.safety.safety_rx_hook(self.stock("ECMEngineStatus", {"CruiseMainOn": 1}))
+    self.safety.set_timer(300001)
+    self.safety.safety_rx_hook(self.sensor(3))
+    self.safety.set_controls_allowed(True)
+    self.assertFalse(self.safety.safety_tx_hook(active(6)))
+    self.assertTrue(self.safety.safety_tx_hook(disabled(6)))
+
+  def test_scheduler_rejects_all_noncanonical_payloads(self):
+    for variant in (GMSafetyFlags.NO_ACC | GMSafetyFlags.BOLT_2017,
+                    GMSafetyFlags.NO_ACC | GMSafetyFlags.BOLT_GEN2):
+      with self.subTest(variant=variant):
+        self.init_mode(variant)
+        self.safety.safety_rx_hook(self.low_gear())
+        self.safety.safety_rx_hook(self.sensor(1))
+        self.safety.set_controls_allowed(True)
+        paddle = create_bolt_regen_paddle(self.packer, True)
+        gear = create_bolt_regen_gear(self.packer, True, bool(variant & GMSafetyFlags.BOLT_GEN2))
+        bad = bytearray(paddle[1])
+        bad[1] = 1
+        self.assertFalse(self.safety.safety_tx_hook(self.packet((paddle[0], bytes(bad), paddle[2]))))
+        bad = bytearray(paddle[1])
+        bad[0] = 0x30
+        self.assertFalse(self.safety.safety_tx_hook(self.packet((paddle[0], bytes(bad), paddle[2]))))
+        for index, replacement in ((0, 0x0D), (3, 0x0F), (5, 3), (6, 2)):
+          bad = bytearray(gear[1])
+          bad[index] = replacement
+          self.assertFalse(self.safety.safety_tx_hook(self.packet((gear[0], bytes(bad), gear[2]))))
+        wrong_generation = create_bolt_regen_gear(self.packer, True, not bool(variant & GMSafetyFlags.BOLT_GEN2))
+        self.assertFalse(self.safety.safety_tx_hook(self.packet(wrong_generation)))
+        self.assertEqual(self.recorded(), [])
+
+  def test_real_parser_controller_and_native_safety_four_variants(self):
+    for candidate in PEDAL_BOLT_CAR:
+      with self.subTest(candidate=candidate):
+        cp = params(candidate, True, True)
+        self.assertTrue(cp.openpilotLongitudinalControl)
+        self.assertEqual(cp.safetyConfigs[0].safetyModel, CarParams.SafetyModel.gm)
+        self.assertEqual(self.safety.set_safety_hooks(CarParams.SafetyModel.gm,
+                                                      cp.safetyConfigs[0].safetyParam), 0)
+        self.safety.init_tests()
+        self.safety.reset_recorded_can()
+        self.safety.set_timer(950000)
+        packer = CANPacker(DBC[candidate][Bus.pt])
+        car_state = CarState(cp)
+        parsers = car_state.get_can_parsers(cp)
+        car_state.update(parsers)
+        sensor = packer.make_can_msg("GAS_SENSOR", 0, {"INTERCEPTOR_GAS": 0,
+                                                      "INTERCEPTOR_GAS2": 0,
+                                                      "COUNTER_PEDAL": 1, "STATE": 0})
+        sensor_data = bytearray(sensor[1])
+        sensor_data[5] = pedal_crc(sensor_data)
+        source = [(sensor[0], bytes(sensor_data), sensor[2]),
+                  packer.make_can_msg("ECMPRDNL2", 0, {"PRNDL2": 6, "ManualMode": 0})]
+        if candidate == CAR.CHEVROLET_BOLT_ACC_2022_2023_PEDAL:
+          source.append(packer.make_can_msg("AcceleratorPedal2", 0, {"CruiseState": 0}))
+          source.append(packer.make_can_msg("ECMEngineStatus", 0, {"CruiseMainOn": 1}))
+        parsers[Bus.pt].update([(950_000_000, source)])
+        out = car_state.update(parsers)
+        self.assertTrue(car_state.pedal_sensor_healthy)
+        self.assertTrue(self.safety.safety_rx_hook(self.packet(source[0])))
+        self.assertTrue(self.safety.safety_rx_hook(self.packet(source[1])))
+        if len(source) > 2:
+          self.assertTrue(self.safety.safety_rx_hook(self.packet(source[2])))
+          self.assertTrue(self.safety.safety_rx_hook(self.packet(source[3])))
+          self.assertFalse(out.cruiseState.enabled)
+          self.assertEqual(car_state.stock_acc_status_ts_nanos, 950_000_000)
+        self.safety.set_controls_allowed(True)
+        self.assertEqual(out.gearShifter, structs.CarState.GearShifter.low)
+
+        controller = CarController(DBC[candidate], cp)
+        controller.frame = 4
+        controller.last_steer_frame = 4
+        control = structs.CarControl()
+        control.enabled = True
+        control.longActive = True
+        control.actuators.accel = 1.
+        out.vEgo = 12.
+        cs = SimpleNamespace(out=out.as_reader(), pedal_sensor_healthy=car_state.pedal_sensor_healthy,
+                             pedal_sensor_ts_nanos=car_state.pedal_sensor_ts_nanos,
+                             stock_acc_status_ts_nanos=car_state.stock_acc_status_ts_nanos,
+                             cam_lka_steering_cmd_counter=0, loopback_lka_steering_cmd_updated=False,
+                             loopback_lka_steering_cmd_ts_nanos=1_000_000_000, pt_lka_steering_cmd_counter=0,
+                             buttons_counter=0)
+        _, commands = controller.update(control.as_reader(), cs, 1_000_000_000)
+        acc_pedal = candidate == CAR.CHEVROLET_BOLT_ACC_2022_2023_PEDAL
+        self.assertEqual([msg[0] for msg in commands], [0x200, 0x1F5, 0xBD] + ([0x315] if acc_pedal else []))
+        self.safety.set_timer(1_000_000)
+        self.assertTrue(self.safety.safety_tx_hook(self.packet(commands[0])))
+        self.assertFalse(self.safety.safety_tx_hook(self.packet(commands[1])))
+        self.assertFalse(self.safety.safety_tx_hook(self.packet(commands[2])))
+        if acc_pedal:
+          self.assertTrue(self.safety.safety_tx_hook(self.packet(commands[3])))
+        self.assertEqual(self.recorded(), [])
+        self.safety.safety_rx_hook(self.packet(packer.make_can_msg("ECMPRDNL2", 0, {"PRNDL2": 6})))
+        self.safety.safety_rx_hook(self.packet(packer.make_can_msg("EBCMRegenPaddle", 0, {"RegenPaddle": 0})))
+        self.assertEqual([item[0] for item in self.recorded()], [0x1F5, 0xBD])
+
+        controller.frame = 8
+        self.safety.reset_recorded_can()
+        _, released = controller.update(control.as_reader(), cs, 1_200_000_000)
+        released = [msg for msg in released if msg[0] in (0x200, 0x1F5, 0xBD)]
+        self.assertEqual([msg[0] for msg in released], [0x200, 0x1F5, 0xBD])
+        self.assertEqual(released[0][1][0:4], b"\x00" * 4)
+        self.assertEqual(released[1][1], b"\x0c\x0c\x00\x06\x00\x00\x01\x00")
+        self.assertEqual(released[2][1], b"\x00" * 7)
+        self.safety.set_timer(1_200_000)
+        self.assertTrue(self.safety.safety_tx_hook(self.packet(released[0])))
+        self.assertFalse(self.safety.safety_tx_hook(self.packet(released[1])))
+        self.assertFalse(self.safety.safety_tx_hook(self.packet(released[2])))
+        self.safety.safety_rx_hook(self.packet(packer.make_can_msg("ECMPRDNL2", 0, {"PRNDL2": 6})))
+        self.safety.safety_rx_hook(self.packet(packer.make_can_msg("EBCMRegenPaddle", 0, {"RegenPaddle": 0})))
+        self.assertEqual(self.recorded(), [(0x1F5, 0, released[1][1]), (0xBD, 0, released[2][1])])

opendbc_repo/opendbc/safety/tests/test_gm_cc_pedal.py

@@ -0,0 +1,299 @@
+import unittest
+
+from opendbc.can import CANPacker
+from opendbc.car.gm.gmcan import create_buttons, create_pedal_command, create_steering_control
+from opendbc.car.structs import CarParams
+from opendbc.safety.tests.libsafety import libsafety_py
+
+
+def crc(data):
+  value = 255
+  for byte in reversed(data[:5]):
+    value ^= byte
+    for _ in range(8):
+      value = ((value << 1) ^ (0xD5 if value & 128 else 0)) & 255
+  return value
+
+
+class TestGmCcPedal(unittest.TestCase):
+  def setUp(self):
+    self.safety = libsafety_py.libsafety
+    self.packer = CANPacker('gm_global_a_powertrain_generated')
+    self.init(0xC180)
+
+  def init(self, word):
+    self.assertEqual(self.safety.set_safety_hooks(CarParams.SafetyModel.gm, word), 0)
+    self.safety.init_tests()
+    self.safety.set_timer(0)
+
+  @staticmethod
+  def packet(msg):
+    return libsafety_py.make_CANPacket(msg[0], msg[2], msg[1])
+
+  def rx(self, addr, data, bus=0):
+    return self.safety.safety_rx_hook(libsafety_py.make_CANPacket(addr, bus, bytes(data)))
+
+  def buttons(self, counter, button=1):
+    return self.packet(create_buttons(self.packer, 0, counter, button))
+
+  def sensor(self, counter, track1=604, track2=304, bad_crc=False):
+    data = bytearray([track1 >> 8, track1 & 255, track2 >> 8, track2 & 255, counter, 0])
+    data[5] = crc(data) ^ int(bad_crc)
+    return data
+
+  def feed(self, counter=0, active=False, main=True, gear=4, brake=False):
+    self.rx(0x184, bytes(8))
+    self.rx(0x34A, bytes(5))
+    self.rx(0xF1, bytes(6))
+    self.rx(0x1C4, bytes(8))
+    c9 = bytearray(8)
+    c9[3] = 32 if main else 0
+    c9[5] = int(brake)
+    self.rx(0xC9, c9)
+    prndl = bytearray(8)
+    prndl[3] = gear
+    self.rx(0x1F5, prndl)
+    self.rx(0x201, self.sensor(counter))
+    stock = bytearray(8)
+    stock[4] = 128 if active else 0
+    self.rx(0x3D1, stock)
+    self.safety.safety_rx_hook(self.buttons(counter % 4))
+
+  def command(self, counter, fraction=.2):
+    return self.packet(create_pedal_command(self.packer, fraction, counter))
+
+  def engage(self):
+    self.safety.safety_rx_hook(self.buttons(0, 2))
+
+  def test_physical_latch_independent_from_stock_and_sensor_drop_recovery(self):
+    for word in (0xC180, 0xC181):
+      self.init(word)
+      self.feed(active=False)
+      self.assertFalse(self.safety.get_controls_allowed())
+      self.engage()
+      self.assertTrue(self.safety.get_controls_allowed())
+      self.assertTrue(self.safety.safety_tx_hook(self.command(0)))
+      self.assertFalse(self.safety.safety_tx_hook(self.command(0)))
+      self.rx(0x201, self.sensor(3))  # Missed sensor counters do not revoke the independent latch.
+      self.assertTrue(self.safety.get_controls_allowed())
+      self.assertTrue(self.safety.safety_tx_hook(self.command(2)))
+      self.rx(0x3D1, bytes(8))
+      self.assertTrue(self.safety.get_controls_allowed())
+      self.assertTrue(self.safety.safety_tx_hook(self.command(3)))
+      self.assertFalse(self.safety.safety_tx_hook(self.command(4)))
+      low = bytearray(8)
+      low[3] = 6
+      self.rx(0x1F5, low)
+      self.assertTrue(self.safety.get_controls_allowed())
+      self.assertTrue(self.safety.safety_tx_hook(self.command(0)))
+      low[3] = 4
+      self.rx(0x1F5, low)
+      self.assertTrue(self.safety.get_controls_allowed())
+      self.assertTrue(self.safety.safety_tx_hook(self.command(1)))
+
+  def test_sensor_crc_override_and_stale_sources(self):
+    self.feed()
+    self.engage()
+    self.rx(0x201, self.sensor(1, bad_crc=True))
+    self.assertFalse(self.safety.get_controls_allowed())
+    self.assertFalse(self.safety.safety_tx_hook(self.command(0)))
+    self.assertTrue(self.safety.safety_tx_hook(self.command(0, 0.)))
+    self.rx(0x201, self.sensor(4))
+    self.engage()
+    # Release then press is a genuine physical rearm edge.
+    self.safety.safety_rx_hook(self.buttons(1))
+    self.engage()
+    self.assertTrue(self.safety.safety_tx_hook(self.command(1)))
+    self.safety.set_timer(100001)
+    self.assertFalse(self.safety.safety_tx_hook(self.command(2)))
+    self.assertTrue(self.safety.safety_tx_hook(self.command(2, 0.)))
+    self.rx(0x201, self.sensor(8, track1=900, track2=450))
+    self.assertFalse(self.safety.safety_tx_hook(self.command(3)))
+
+  def test_cancel_has_physical_credit_and_never_acceleration_buttons(self):
+    self.feed(active=True)
+    self.assertFalse(self.safety.get_controls_allowed())
+    self.assertTrue(self.safety.safety_tx_hook(self.buttons(1, 6)))
+    self.assertFalse(self.safety.safety_tx_hook(self.buttons(1, 6)))
+    self.assertFalse(self.safety.safety_tx_hook(self.buttons(1, 2)))
+    self.assertFalse(self.safety.safety_tx_hook(self.buttons(1, 3)))
+    self.safety.set_timer(50000)
+    self.safety.safety_rx_hook(self.buttons(1))
+    self.assertTrue(self.safety.safety_tx_hook(self.buttons(2, 6)))
+    self.safety.set_timer(100000)
+    self.safety.safety_rx_hook(self.buttons(1))
+    self.assertFalse(self.safety.safety_tx_hook(self.buttons(2, 6)))
+    self.rx(0x3D1, bytes(8))
+    self.safety.safety_rx_hook(self.buttons(2))
+    self.assertFalse(self.safety.safety_tx_hook(self.buttons(3, 6)))
+
+  def test_main_brake_gear_and_wrong_sensor_provenance(self):
+    for main, gear, brake in ((False, 4, False), (True, 2, False), (True, 0, False), (True, 4, True)):
+      self.init(0xC180)
+      self.feed(main=main, gear=gear, brake=brake)
+      self.engage()
+      self.assertFalse(self.safety.safety_tx_hook(self.command(0)))
+    self.init(0xC180)
+    self.feed()
+    self.engage()
+    self.safety.set_timer(100001)
+    self.rx(0x201, self.sensor(1), bus=2)
+    self.rx(0x201, self.sensor(1)[:-1])
+    self.assertFalse(self.safety.safety_tx_hook(self.command(0)))
+
+  def test_topology_and_unowned_messages(self):
+    for word in (0xC180, 0xC181):
+      self.init(word)
+      for addr, length in ((0x315, 8), (0x2CB, 5), (0x370, 6), (0x3D1, 8), (0xBD, 7), (0x1F5, 8)):
+        self.assertFalse(self.safety.safety_tx_hook(libsafety_py.make_CANPacket(addr, 0, bytes(length))))
+      for addr in (0x409, 0x40A):
+        packet = libsafety_py.make_CANPacket(addr, 0, bytes(7))
+        self.assertEqual(self.safety.safety_tx_hook(packet), word == 0xC181)
+        self.assertFalse(self.safety.safety_tx_hook(libsafety_py.make_CANPacket(addr, 0, bytes([1] + [0] * 6))))
+
+  def test_cancel_jitter_skipped_slot_and_idle_recovery(self):
+    self.feed(active=True)
+    self.assertTrue(self.safety.safety_tx_hook(self.buttons(1, 6)))
+    self.safety.set_timer(110000)
+    self.safety.safety_rx_hook(self.buttons(1))
+    self.assertTrue(self.safety.safety_tx_hook(self.buttons(2, 6)))
+    self.safety.set_timer(220000)
+    self.safety.safety_rx_hook(self.buttons(3))  # A dropped physical neutral cannot grant fresh credit.
+    self.assertFalse(self.safety.safety_tx_hook(self.buttons(0, 6)))
+    self.safety.set_timer(250000)
+    self.safety.safety_rx_hook(self.buttons(0))
+    self.assertTrue(self.safety.safety_tx_hook(self.buttons(1, 6)))
+    self.safety.set_timer(600000)
+    self.feed(counter=1, active=True)
+    self.assertFalse(self.safety.safety_tx_hook(self.buttons(2, 6)))
+    self.safety.set_timer(650000)
+    self.safety.safety_rx_hook(self.buttons(2))
+    self.assertTrue(self.safety.safety_tx_hook(self.buttons(3, 6)))
+
+  def test_aol_lateral_never_grants_pedal_longitudinal(self):
+    for word in (0xC180, 0xC181):
+      self.safety.set_alternative_experience(32)
+      self.init(word)
+      self.safety.set_aol_test_heartbeat(True)
+      self.feed(active=False)
+      self.safety.aol_set_host_request(1)
+      self.assertEqual(self.safety.aol_get_permission_mask(), 1)
+      steer = self.packet(create_steering_control(self.packer, 0, 5, 0, True))
+      self.assertTrue(self.safety.safety_tx_hook(steer))
+      self.assertFalse(self.safety.safety_tx_hook(self.command(0)))
+      self.engage()
+      self.assertFalse(self.safety.safety_tx_hook(self.command(0)))
+      self.safety.aol_set_host_request(3)
+      self.assertTrue(self.safety.safety_tx_hook(self.command(0)))
+    self.safety.set_alternative_experience(0)
+
+  def test_recorded_adc_tracks_and_overflow_keep_command_pair_contract(self):
+    for sample in ('053502ba0164', '04e30286048f', '046c024108e3', '0279012a06f6'):
+      with self.subTest(sample=sample):
+        self.init(0xC180)
+        self.feed()
+        self.engage()
+        self.assertTrue(self.safety.get_controls_allowed())
+        data = bytes.fromhex(sample)
+        self.assertEqual(crc(data), data[5])
+        self.assertTrue(self.rx(0x201, data))
+        self.assertTrue(self.safety.get_controls_allowed())
+        self.assertEqual(self.safety.safety_tx_hook(self.command(1)), sample == '0279012a06f6')
+    self.init(0xC180)
+    self.feed()
+    self.engage()
+    command = create_pedal_command(self.packer, .2, 1)
+    bad_pair = bytearray(command[1])
+    bad_pair[2:4] = (int.from_bytes(bad_pair[2:4], "big") + 12).to_bytes(2, "big")
+    bad_pair[5] = crc(bad_pair)
+    self.assertFalse(self.safety.safety_tx_hook(self.packet((command[0], bytes(bad_pair), command[2]))))
+    self.assertTrue(self.safety.safety_tx_hook(self.packet(command)))
+    for track1, track2 in ((4096, 304), (604, 4096)):
+      self.feed(counter=8 if track1 == 4096 else 9)
+      self.safety.safety_rx_hook(self.buttons(1))
+      self.engage()
+      self.assertTrue(self.safety.get_controls_allowed())
+      self.rx(0x201, self.sensor(2 if track1 == 4096 else 3, track1=track1, track2=track2))
+      self.assertFalse(self.safety.get_controls_allowed())
+
+  def volt_feed(self, word, counter=0, *, active=True, regen=False):
+    self.feed(counter=counter, active=active)
+    if not (word & 1):
+      self.rx(0xBE, bytes(6))
+    self.rx(0xBD, bytes([16 if regen else 0] + [0] * 6))
+
+  def test_volt_cc_exact_records_and_release_denial(self):
+    release = self.safety.set_safety_hooks(CarParams.SafetyModel.allOutput, 0) != 0
+    for index in range(8):
+      word = 0xE600 + index
+      with self.subTest(word=word):
+        self.init(word)
+        self.volt_feed(word)
+        self.engage()
+        self.assertEqual(self.safety.safety_tx_hook(self.command(0)), not release)
+        for address, length in ((0x315, 5), (0x2CB, 8), (0x370, 6), (0x2CD, 8), (0x306, 8)):
+          self.assertFalse(self.safety.safety_tx_hook(libsafety_py.make_CANPacket(address, 0, bytes(length))))
+        for address in (0x409, 0x40A):
+          self.assertEqual(self.safety.safety_tx_hook(libsafety_py.make_CANPacket(address, 0, bytes(7))),
+                           bool(index & 2) and not release)
+    for word in (0xE5FF, 0xE608, 0xE60F, 0xE618, 0xE61F, 0xE6FF):
+      self.init(word)
+      self.feed()
+      self.engage()
+      self.assertFalse(self.safety.safety_tx_hook(self.command(0)))
+
+  def test_volt_cc_independent_adc_gas_threshold_and_regen(self):
+    if self.safety.set_safety_hooks(CarParams.SafetyModel.allOutput, 0) != 0:
+      self.skipTest("Active Volt CC interceptor profiles are DEBUG-only")
+    for raw1, raw2, gas in ((0, 787, False), (0, 788, True), (604, 486, False), (604, 487, True)):
+      self.init(0xE600)
+      self.volt_feed(0xE600)
+      self.engage()
+      self.assertTrue(self.rx(0x201, self.sensor(1, raw1, raw2)))
+      self.assertEqual(self.safety.safety_tx_hook(self.command(0)), not gas)
+    self.init(0xE601)
+    self.volt_feed(0xE601, regen=True)
+    self.engage()
+    self.assertFalse(self.safety.safety_tx_hook(self.command(0)))
+    self.assertTrue(self.safety.safety_tx_hook(self.command(0, 0.)))
+
+  def test_volt_cc_stock_optional_pedal_and_gear_cannot_withdraw_cruise(self):
+    for index in range(8):
+      word = 0xE610 + index
+      self.init(word)
+      # Feed every stock-owned source, deliberately never supplying 201 or gear.
+      for address, length in ((0x184, 8), (0x34A, 5), (0x1C4, 8), (0xBD, 7)):
+        self.rx(address, bytes(length))
+      self.rx(0xF1 if index & 1 else 0xBE, bytes(6))
+      main = bytearray(8)
+      main[3] = 32
+      self.rx(0xC9, main)
+      cruise = bytearray(8)
+      cruise[4] = 128
+      self.rx(0x3D1, cruise)
+      self.safety.safety_rx_hook(self.buttons(0))
+      self.engage()
+      self.assertTrue(self.safety.get_controls_allowed())
+      self.rx(0x201, self.sensor(1, 4096, 0))
+      self.rx(0x1F5, bytes(8))
+      self.assertTrue(self.safety.get_controls_allowed())
+      self.assertFalse(self.safety.safety_tx_hook(self.command(0)))
+      self.safety.safety_rx_hook(self.buttons(1))
+      cancel = create_buttons(self.packer, 2, 1, 6)
+      self.assertTrue(self.safety.safety_tx_hook(self.packet(cancel)))
+      self.assertFalse(self.safety.safety_tx_hook(self.packet(cancel)))
+      self.assertFalse(self.safety.safety_tx_hook(self.buttons(1, 6)))
+
+  def test_volt_cc_sequential_command_counter_and_source_expiry(self):
+    if self.safety.set_safety_hooks(CarParams.SafetyModel.allOutput, 0) != 0:
+      self.skipTest("Active Volt CC interceptor profiles are DEBUG-only")
+    self.init(0xE600)
+    self.volt_feed(0xE600)
+    self.engage()
+    self.assertTrue(self.safety.safety_tx_hook(self.command(3)))
+    self.assertTrue(self.safety.safety_tx_hook(self.command(0)))
+    self.assertFalse(self.safety.safety_tx_hook(self.command(2)))
+    self.assertTrue(self.safety.safety_tx_hook(self.command(1)))
+    self.safety.set_timer(100001)
+    self.assertFalse(self.safety.safety_tx_hook(self.command(2)))
+    self.assertTrue(self.safety.safety_tx_hook(self.command(2, 0.)))

opendbc_repo/opendbc/safety/tests/test_gm_cc_pedal_disabled.py

@@ -0,0 +1,204 @@
+import unittest
+
+from opendbc.can import CANPacker
+from opendbc.car.gm.gmcan import create_buttons, create_pedal_command, create_steering_control
+from opendbc.car.structs import CarParams
+from opendbc.safety.tests.libsafety import libsafety_py
+
+
+def crc(data):
+  value = 255
+  for byte in reversed(data[:5]):
+    value ^= byte
+    for _ in range(8):
+      value = ((value << 1) ^ (0xD5 if value & 128 else 0)) & 255
+  return value
+
+
+class TestGmCcPedalDisabled(unittest.TestCase):
+  WORDS = (0xC186, 0xC187)
+
+  def setUp(self):
+    self.safety = libsafety_py.libsafety
+    self.packer = CANPacker('gm_global_a_powertrain_generated')
+    self.safety.set_alternative_experience(0)
+    self.init(self.WORDS[0])
+
+  def tearDown(self):
+    self.safety.set_alternative_experience(0)
+
+  def init(self, word):
+    self.assertEqual(self.safety.set_safety_hooks(CarParams.SafetyModel.gm, word), 0)
+    self.safety.init_tests()
+    self.safety.set_timer(0)
+
+  @staticmethod
+  def packet(msg):
+    return libsafety_py.make_CANPacket(msg[0], msg[2], msg[1])
+
+  def rx(self, addr, data, bus=0):
+    return self.safety.safety_rx_hook(libsafety_py.make_CANPacket(addr, bus, bytes(data)))
+
+  def buttons(self, counter, button=1, bus=0):
+    return self.packet(create_buttons(self.packer, bus, counter, button))
+
+  def sensor(self, counter, track1=604, track2=304, bad_crc=False):
+    data = bytearray([track1 >> 8, track1 & 255, track2 >> 8, track2 & 255, counter, 0])
+    data[5] = crc(data) ^ int(bad_crc)
+    return data
+
+  def sources(self, counter=0, active=True, main=True, brake=False, gear=4, skip=None):
+    c9, stock, prndl = bytearray(8), bytearray(8), bytearray(8)
+    c9[3], c9[5], stock[4], prndl[3] = int(main) << 5, int(brake), int(active) << 7, gear
+    sources = {0x184: bytes(8), 0x34A: bytes(5), 0xF1: bytes(6), 0x1C4: bytes(8),
+               0xC9: c9, 0x1F5: prndl, 0x201: self.sensor(counter), 0x3D1: stock}
+    for addr, data in sources.items():
+      if addr != skip:
+        self.rx(addr, data)
+    if skip != 0x1E1:
+      self.safety.safety_rx_hook(self.buttons(counter % 4))
+
+  def cancel(self, counter, bus=2):
+    return self.buttons(counter, 6, bus)
+
+  def test_observed_camera_cancel_and_no_stock_inactive_exception(self):
+    for word in self.WORDS:
+      for active in (False, True):
+        self.init(word)
+        self.sources(active=active)
+        self.assertFalse(self.safety.get_controls_allowed())
+        self.assertEqual(self.safety.safety_tx_hook(self.cancel(0)), active)
+      # Switching owners must not leak the disabled bus/counter codec into active words.
+      self.init(0xC181 if word == 0xC187 else 0xC180)
+      self.sources()
+      self.assertFalse(self.safety.safety_tx_hook(self.cancel(0)))
+      self.assertFalse(self.safety.safety_tx_hook(self.cancel(0, 0)))
+      self.assertTrue(self.safety.safety_tx_hook(self.cancel(1, 0)))
+
+  def test_main_stock_button_leases(self):
+    for word in self.WORDS:
+      self.init(word)
+      self.sources(main=False)
+      self.assertFalse(self.safety.safety_tx_hook(self.cancel(0)))
+      self.init(word)
+      self.sources()
+      self.safety.set_timer(300000)
+      self.assertTrue(self.safety.safety_tx_hook(self.cancel(0)))
+      for missing in (0xC9, 0x3D1, 0x1E1):
+        with self.subTest(word=word, missing=missing):
+          self.init(word)
+          self.sources()
+          self.safety.set_timer(300001)
+          self.sources(counter=1, skip=missing)
+          if missing != 0x1E1:
+            self.safety.safety_rx_hook(self.buttons(2))
+          self.assertFalse(self.safety.safety_tx_hook(self.cancel(0 if missing == 0x1E1 else 2)))
+
+  def test_full_template_wrong_bus_counter_and_buttons(self):
+    for word in self.WORDS:
+      self.init(word)
+      self.sources()
+      self.assertFalse(self.safety.safety_tx_hook(self.cancel(0, 0)))
+      self.assertFalse(self.safety.safety_tx_hook(self.cancel(1)))
+      for button in (0, 1, 2, 3, 4, 5, 7):
+        self.assertFalse(self.safety.safety_tx_hook(self.buttons(0, button, 2)))
+      canonical = create_buttons(self.packer, 2, 0, 6)[1]
+      for bit in range(56):
+        data = bytearray(canonical)
+        data[bit // 8] ^= 1 << (bit % 8)
+        self.assertFalse(self.safety.safety_tx_hook(libsafety_py.make_CANPacket(0x1E1, 2, bytes(data))))
+      self.assertFalse(self.safety.safety_tx_hook(libsafety_py.make_CANPacket(0x1E1, 2, canonical[:-1])))
+      self.assertTrue(self.safety.safety_tx_hook(self.cancel(0)))
+
+  def test_neutral_credit_cadence_and_interleaved_stock(self):
+    for word in self.WORDS:
+      self.init(word)
+      self.sources()
+      self.assertTrue(self.safety.safety_tx_hook(self.cancel(0)))
+      self.assertFalse(self.safety.safety_tx_hook(self.cancel(0)))
+      self.safety.set_timer(40000)
+      self.safety.safety_rx_hook(self.buttons(1))
+      self.assertFalse(self.safety.safety_tx_hook(self.cancel(1)))
+      self.safety.set_timer(40001)
+      self.assertTrue(self.safety.safety_tx_hook(self.cancel(1)))
+      self.safety.set_timer(90000)
+      self.safety.safety_rx_hook(self.buttons(1))
+      self.assertFalse(self.safety.safety_tx_hook(self.cancel(1)))
+      self.safety.safety_rx_hook(self.buttons(3))
+      self.assertFalse(self.safety.safety_tx_hook(self.cancel(3)))
+      self.safety.safety_rx_hook(self.buttons(0))
+      self.rx(0x3D1, bytes(8))
+      active = bytearray(8)
+      active[4] = 128
+      self.rx(0x3D1, active)
+      self.assertFalse(self.safety.safety_tx_hook(self.cancel(0)))
+      self.safety.safety_rx_hook(self.buttons(1))
+      self.assertTrue(self.safety.safety_tx_hook(self.cancel(1)))
+
+  def test_cancel_while_braking_or_driver_override_and_sensor_fault(self):
+    for word in self.WORDS:
+      for condition in ('brake', 'gas', 'reverse', 'sensor_crc', 'sensor_replay'):
+        with self.subTest(word=word, condition=condition):
+          self.init(word)
+          self.sources(brake=condition == 'brake', gear=2 if condition == 'reverse' else 4)
+          if condition == 'gas':
+            self.rx(0x201, self.sensor(1, 900, 450))
+          elif condition == 'sensor_crc':
+            self.rx(0x201, self.sensor(1, bad_crc=True))
+          elif condition == 'sensor_replay':
+            self.rx(0x201, self.sensor(0))
+          self.assertTrue(self.safety.safety_tx_hook(self.cancel(0)))
+          self.safety.set_controls_allowed(True)
+          for fraction in (0., .2):
+            self.assertFalse(self.safety.safety_tx_hook(self.packet(create_pedal_command(self.packer, fraction, 0))))
+
+  def test_common_rx_and_relay_health(self):
+    for word in self.WORDS:
+      self.init(word)
+      self.sources()
+      self.safety.set_relay_malfunction(True)
+      self.assertFalse(self.safety.safety_tx_hook(self.cancel(0)))
+      self.safety.set_relay_malfunction(False)
+      self.safety.set_timer(2000000)
+      self.safety.safety_tick()
+      self.sources(counter=1)
+      self.assertFalse(self.safety.safety_tx_hook(self.cancel(1)))
+      self.safety.safety_tick()
+      self.safety.safety_rx_hook(self.buttons(2))
+      self.assertTrue(self.safety.safety_tx_hook(self.cancel(2)))
+
+  def test_camera_acc_status_does_not_replace_pt_cruise(self):
+    for word in self.WORDS:
+      for active in (False, True):
+        self.init(word)
+        self.sources(active=active)
+        status = bytearray(6)
+        status[2] = 0 if active else 128
+        self.rx(0x370, status, 2)
+        self.assertEqual(self.safety.safety_tx_hook(self.cancel(0)), active)
+      self.init(word)
+      self.sources(skip=0x3D1)
+      status[2] = 128
+      self.rx(0x370, status, 2)
+      self.assertFalse(self.safety.safety_tx_hook(self.cancel(0)))
+
+  def test_no_longitudinal_authority_including_aol(self):
+    for word in self.WORDS:
+      for alternative in (0, 32):
+        self.safety.set_alternative_experience(alternative)
+        self.init(word)
+        self.sources()
+        if alternative:
+          self.safety.set_aol_test_heartbeat(True)
+          self.safety.aol_set_host_request(1)
+          self.assertEqual(self.safety.aol_get_permission_mask(), 1)
+          steer = self.packet(create_steering_control(self.packer, 0, 5, 0, True))
+          self.assertTrue(self.safety.safety_tx_hook(steer))
+        self.safety.set_controls_allowed(True)
+        if alternative:
+          self.safety.aol_set_host_request(3)
+        for fraction in (0., .2, 1.):
+          self.assertFalse(self.safety.safety_tx_hook(self.packet(create_pedal_command(self.packer, fraction, 0))))
+        for addr, length in ((0x409, 7), (0x40A, 7), (0x315, 5), (0x2CB, 8), (0x370, 6), (0x3D1, 8), (0xBD, 7), (0x1F5, 8)):
+          self.assertFalse(self.safety.safety_tx_hook(libsafety_py.make_CANPacket(addr, 0, bytes(length))))
+        self.assertTrue(self.safety.safety_tx_hook(self.cancel(0)))

opendbc_repo/opendbc/safety/tests/test_gm_silverado_pedal.py

@@ -0,0 +1,277 @@
+import unittest
+
+from opendbc.can import CANPacker
+from opendbc.car.gm.gmcan import create_buttons, create_pedal_command, create_steering_control
+from opendbc.car.structs import CarParams
+from opendbc.safety.tests.libsafety import libsafety_py
+
+
+def crc(data):
+  value = 255
+  for byte in reversed(data[:5]):
+    value ^= byte
+    for _ in range(8):
+      value = ((value << 1) ^ (0xD5 if value & 128 else 0)) & 255
+  return value
+
+
+class TestGmSilveradoPedal(unittest.TestCase):
+  WORDS = (0xC182, 0xC183, 0xC184, 0xC185)
+  LENGTHS = {0x184: 8, 0x34A: 5, 0xC9: 8, 0x3D1: 8, 0x1E1: 7, 0x1C4: 8, 0x1F5: 8, 0x201: 6}
+
+  def setUp(self):
+    self.safety = libsafety_py.libsafety
+    self.packer = CANPacker('gm_global_a_powertrain_generated')
+    self.safety.set_alternative_experience(0)
+    self.init(0xC182)
+
+  def tearDown(self):
+    self.safety.set_alternative_experience(0)
+
+  def init(self, word):
+    self.assertEqual(self.safety.set_safety_hooks(CarParams.SafetyModel.gm, word), 0)
+    self.safety.init_tests()
+    self.safety.set_timer(0)
+
+  @staticmethod
+  def packet(msg):
+    return libsafety_py.make_CANPacket(msg[0], msg[2], msg[1])
+
+  def rx(self, addr, data, bus=0):
+    return self.safety.safety_rx_hook(libsafety_py.make_CANPacket(addr, bus, bytes(data)))
+
+  def buttons(self, counter, button=1, bus=0):
+    return self.packet(create_buttons(self.packer, bus, counter, button))
+
+  def sensor(self, counter, track1=604, track2=304, state=0, bad_crc=False):
+    data = bytearray([track1 >> 8, track1 & 255, track2 >> 8, track2 & 255, (state << 4) | counter, 0])
+    data[5] = crc(data) ^ int(bad_crc)
+    return data
+
+  def sources(self, counter=0, active=False, main=True, gear=4, brake=False, skip=None, bus=0):
+    c9, stock, prndl = bytearray(8), bytearray(8), bytearray(8)
+    c9[3], c9[5], stock[4], prndl[3] = (32 if main else 0), int(brake), (128 if active else 0), gear
+    values = {0x184: bytes(8), 0x34A: bytes(5), 0xC9: c9, 0x3D1: stock,
+              0x1C4: bytes(8), 0x1F5: prndl, 0x201: self.sensor(counter)}
+    for addr, data in values.items():
+      if addr != skip:
+        self.rx(addr, data, bus=bus)
+    if skip != 0x1E1:
+      self.safety.safety_rx_hook(self.buttons(counter % 4, bus=bus))
+
+  def command(self, counter, fraction=.2, bus=0):
+    msg = create_pedal_command(self.packer, fraction, counter)
+    return libsafety_py.make_CANPacket(msg[0], bus, msg[1])
+
+  def engage(self):
+    self.safety.safety_rx_hook(self.buttons(0, 2))
+
+  def test_active_pedal_latch_and_stock_cancel_pending(self):
+    for word in self.WORDS[:2]:
+      for active in (False, True):
+        self.init(word)
+        self.sources(active=active)
+        self.assertFalse(self.safety.get_controls_allowed())
+        self.engage()
+        self.assertTrue(self.safety.safety_tx_hook(self.command(0)))
+        self.assertFalse(self.safety.safety_tx_hook(self.command(0)))
+        self.assertTrue(self.safety.safety_tx_hook(self.command(2)))
+        self.assertFalse(self.safety.safety_tx_hook(self.command(4)))
+        self.assertFalse(self.safety.safety_tx_hook(self.command(3, bus=2)))
+
+  def test_cancel_bus_counter_and_disabled_stock_state(self):
+    for word in self.WORDS:
+      disabled = word >= 0xC184
+      for active in (False, True):
+        self.init(word)
+        self.sources(counter=2, active=active)
+        counter, bus = (2, 2) if disabled else (3, 0)
+        # An interleaved stock-off update must not erase valid disabled credit.
+        stock = bytearray(8)
+        stock[4] = 128 if active else 0
+        self.rx(0x3D1, stock)
+        self.assertFalse(self.safety.safety_tx_hook(self.buttons(counter, 6, 2 - bus)))
+        self.assertFalse(self.safety.safety_tx_hook(self.buttons((counter + 1) % 4, 6, bus)))
+        self.assertEqual(self.safety.safety_tx_hook(self.buttons(counter, 6, bus)), disabled or active)
+        self.assertFalse(self.safety.safety_tx_hook(self.buttons(counter, 6, bus)))
+        for button in (1, 2, 3):
+          self.assertFalse(self.safety.safety_tx_hook(self.buttons(counter, button, bus)))
+        self.assertFalse(self.safety.safety_tx_hook(self.command(0)))
+
+  def test_cancel_credit_cadence_replay_and_noncanonical_packet(self):
+    for word in self.WORDS:
+      disabled, bus = word >= 0xC184, 2 if word >= 0xC184 else 0
+      self.init(word)
+      self.sources(active=True)
+      self.assertTrue(self.safety.safety_tx_hook(self.buttons(0 if disabled else 1, 6, bus)))
+      self.safety.set_timer(40000)
+      self.sources(counter=1, active=True)
+      self.assertFalse(self.safety.safety_tx_hook(self.buttons(1 if disabled else 2, 6, bus)))
+      self.safety.set_timer(40001)
+      self.assertTrue(self.safety.safety_tx_hook(self.buttons(1 if disabled else 2, 6, bus)))
+      self.safety.set_timer(90000)
+      self.safety.safety_rx_hook(self.buttons(1))
+      self.assertFalse(self.safety.safety_tx_hook(self.buttons(1 if disabled else 2, 6, bus)))
+      self.sources(counter=2, active=True)
+      msg = create_buttons(self.packer, bus, 2 if disabled else 3, 6)
+      for index in range(7):
+        mutated = bytearray(msg[1])
+        mutated[index] ^= 1
+        self.assertFalse(self.safety.safety_tx_hook(libsafety_py.make_CANPacket(0x1E1, bus, mutated)))
+      self.assertTrue(self.safety.safety_tx_hook(self.buttons(2 if disabled else 3, 6, bus)))
+
+  def test_every_selected_source_is_required_and_bus_length_bound(self):
+    for word in self.WORDS:
+      for addr, length in self.LENGTHS.items():
+        self.init(word)
+        self.sources(active=True, skip=addr)
+        self.rx(addr, bytes(length), bus=2)
+        self.rx(addr, bytes(length - 1))
+        if word >= 0xC184:
+          self.assertFalse(self.safety.safety_tx_hook(self.buttons(0, 6, 2)))
+        else:
+          self.safety.set_controls_allowed(True)  # Isolate RX freshness from the engagement latch.
+          self.assertFalse(self.safety.safety_tx_hook(self.command(0)))
+
+  def test_each_source_staleness_and_recovery(self):
+    for word in self.WORDS:
+      for addr in self.LENGTHS:
+        self.init(word)
+        self.sources(active=True)
+        self.safety.set_timer(300001)
+        self.sources(counter=1, active=True, skip=addr)
+        self.safety.set_controls_allowed(True)
+        if word >= 0xC184:
+          self.assertFalse(self.safety.safety_tx_hook(self.buttons(1, 6, 2)))
+        else:
+          self.assertFalse(self.safety.safety_tx_hook(self.command(0)))
+        self.safety.set_timer(350001)
+        self.sources(counter=2, active=True)
+        self.safety.safety_rx_hook(self.buttons(3))
+        self.safety.set_controls_allowed(True)
+        if word >= 0xC184:
+          self.assertTrue(self.safety.safety_tx_hook(self.buttons(3, 6, 2)))
+        else:
+          self.assertTrue(self.safety.safety_tx_hook(self.command(0)))
+
+  def test_main_off_and_disabled_never_pedal(self):
+    for word in self.WORDS:
+      self.init(word)
+      self.sources(active=True, main=False)
+      self.engage()
+      self.assertFalse(self.safety.safety_tx_hook(self.command(0)))
+      self.assertFalse(self.safety.safety_tx_hook(self.buttons(0 if word >= 0xC184 else 1, 6, 2 if word >= 0xC184 else 0)))
+      if word >= 0xC184:
+        self.init(word)
+        self.sources(active=True)
+        self.engage()
+        self.assertFalse(self.safety.safety_tx_hook(self.command(0)))
+        self.assertFalse(self.safety.safety_tx_hook(self.command(0, 0.)))
+
+  def test_digital_brake_driver_gas_sensor_integrity_and_forward_gear(self):
+    for word in self.WORDS[:2]:
+      for source in ('brake', 'gas', 'crc', 'pair', 'state', 'replay', 'reverse', 'manual'):
+        self.init(word)
+        self.sources(brake=source == 'brake', gear=2 if source == 'reverse' else 4)
+        if source == 'gas':
+          self.rx(0x201, self.sensor(1, 900, 450))
+        elif source == 'crc':
+          self.rx(0x201, self.sensor(1, bad_crc=True))
+        elif source == 'pair':
+          self.rx(0x201, self.sensor(1, 900, 304))
+        elif source == 'state':
+          self.rx(0x201, self.sensor(1, state=1))
+        elif source == 'replay':
+          self.rx(0x201, self.sensor(0))
+        elif source == 'manual':
+          prndl = bytearray(8)
+          prndl[3], prndl[5] = 4, 2
+          self.rx(0x1F5, prndl)
+        self.engage()
+        self.assertFalse(self.safety.safety_tx_hook(self.command(0)))
+      self.init(word)
+      self.sources()
+      self.engage()
+      self.rx(0xF1, bytes([0, 255, 0, 0, 0, 0]))
+      self.rx(0xBE, bytes([0, 255, 0, 0, 0, 0]))
+      self.assertTrue(self.safety.safety_tx_hook(self.command(0)))
+      low = bytearray(8)
+      low[3] = 6
+      self.rx(0x1F5, low)
+      self.assertTrue(self.safety.safety_tx_hook(self.command(1)))
+
+  def test_malformed_sensor_cannot_refund_disabled_cancel_and_pedal_packet_masks(self):
+    for word in self.WORDS[2:]:
+      for replay in (False, True):
+        self.init(word)
+        self.sources(active=False)
+        self.rx(0x201, self.sensor(0 if replay else 1, bad_crc=not replay))
+        self.assertFalse(self.safety.safety_tx_hook(self.buttons(0, 6, 2)))
+    for word in self.WORDS[:2]:
+      self.init(word)
+      self.sources()
+      self.engage()
+      self.assertTrue(self.safety.safety_tx_hook(self.command(0, 1.)))
+      msg = create_pedal_command(self.packer, .2, 1)
+      for index in range(6):
+        changed = bytearray(msg[1])
+        changed[index] ^= 1
+        self.assertFalse(self.safety.safety_tx_hook(libsafety_py.make_CANPacket(0x200, 0, changed)))
+      changed = bytearray(msg[1])
+      changed[4] |= 0x10
+      changed[5] = crc(changed)
+      self.assertFalse(self.safety.safety_tx_hook(libsafety_py.make_CANPacket(0x200, 0, changed)))
+      changed[4] &= ~0x10
+      changed[0], changed[1] = 2634 >> 8, 2634 & 255
+      changed[5] = crc(changed)
+      self.assertFalse(self.safety.safety_tx_hook(libsafety_py.make_CANPacket(0x200, 0, changed)))
+      self.assertTrue(self.safety.safety_tx_hook(self.command(1, 0.)))
+
+  def test_topology_no_unfed_scheduler_or_unowned_speed_packets(self):
+    for word in self.WORDS:
+      self.init(word)
+      self.sources()
+      for addr, length in ((0x315, 5), (0x2CB, 8), (0x2CD, 5), (0x370, 6), (0x3D1, 8), (0xBD, 7),
+                           (0x1F5, 8), (0xA1, 7), (0x306, 8), (0x308, 7), (0x310, 2)):
+        for bus in range(3):
+          self.assertFalse(self.safety.safety_tx_hook(libsafety_py.make_CANPacket(addr, bus, bytes(length))))
+      for addr in (0x409, 0x40A):
+        self.assertEqual(self.safety.safety_tx_hook(libsafety_py.make_CANPacket(addr, 0, bytes(7))), word == 0xC183)
+        self.assertFalse(self.safety.safety_tx_hook(libsafety_py.make_CANPacket(addr, 0, bytes([1] + [0] * 6))))
+      for bus, addr in ((0, 0x184), (0, 0x3D1), (2, 0x180), (2, 0x370)):
+        self.assertEqual(self.safety.safety_fwd_hook(bus, addr), -1)
+      self.assertEqual(self.safety.safety_fwd_hook(2, 0x315), 0)
+      self.assertEqual(self.safety.safety_fwd_hook(2, 0x2CB), 0)
+
+  def test_reset_removes_cancel_and_pedal_authority(self):
+    for word in self.WORDS:
+      self.init(word)
+      self.sources(active=True)
+      self.engage()
+      self.init(word)
+      self.assertFalse(self.safety.safety_tx_hook(self.command(0)))
+      self.assertFalse(self.safety.safety_tx_hook(self.buttons(0 if word >= 0xC184 else 1, 6, 2 if word >= 0xC184 else 0)))
+
+  def test_aol_does_not_grant_longitudinal_and_stock_only_stays_denied(self):
+    for word in self.WORDS:
+      self.safety.set_alternative_experience(32)
+      self.init(word)
+      self.safety.set_aol_test_heartbeat(True)
+      self.sources()
+      self.safety.aol_set_host_request(1)
+      self.assertEqual(self.safety.aol_get_permission_mask(), 1)
+      self.assertTrue(self.safety.safety_tx_hook(self.packet(create_steering_control(self.packer, 0, 5, 0, True))))
+      self.assertFalse(self.safety.safety_tx_hook(self.command(0)))
+      self.engage()
+      self.assertFalse(self.safety.safety_tx_hook(self.command(0)))
+      self.safety.aol_set_host_request(3)
+      self.assertEqual(self.safety.safety_tx_hook(self.command(0)), word < 0xC184)
+
+  def test_alternative_zero_has_no_aol_axis_permission(self):
+    for word in self.WORDS:
+      self.init(word)
+      self.safety.set_aol_test_heartbeat(True)
+      self.sources()
+      self.safety.aol_set_host_request(3)
+      self.assertEqual(self.safety.aol_get_permission_mask(), 0)
+      self.assertFalse(self.safety.safety_tx_hook(self.command(0)))