Restore Jeep Grand Cherokee stock ACC brake hold

This commit is contained in:
firestar5683
2026-10-20 21:12:26 -05:00
parent e26202eae2
commit 2ad95d4176
20 changed files with 776 additions and 11 deletions
@@ -20,11 +20,14 @@ def qualified(cp, *, marked_only=False):
cp.alternativeExperience not in ((CHRYSLER_AOL_EXPERIENCE,) if marked_only else (0, CHRYSLER_AOL_EXPERIENCE))):
return False
flags = ChryslerFlags.HIGHER_MIN_STEERING_SPEED | ChryslerFlags.HAS_BSM | ChryslerFlags.STEERING_SPEED_BYPASS
jeep = cp.carFingerprint in (CAR.JEEP_GRAND_CHEROKEE, CAR.JEEP_GRAND_CHEROKEE_2019)
if jeep:
flags |= ChryslerFlags.JEEP_BRAKE_HOLD_AVAILABLE
expected = (1 if cp.carFingerprint == CAR.RAM_1500_5TH_GEN else
8 if cp.carFingerprint == CAR.CHRYSLER_PACIFICA_2019_HYBRID and cp.alternativeExperience == 32 else 0)
return bool(not int(cp.flags) & ~int(flags) and
cp.safetyConfigs[0].safetyModel == structs.CarParams.SafetyModel.chrysler and
int(cp.safetyConfigs[0].safetyParam) == expected and
int(cp.safetyConfigs[0].safetyParam) in ((0, 4) if jeep else (expected,)) and
cp.lateralTuning.which() in (('torque',) if cp.carFingerprint == CAR.RAM_1500_5TH_GEN else ('pid', 'torque')))
@@ -0,0 +1,25 @@
"""Optional stock-ACC stop retention for exactly the two Grand Cherokee ports."""
def brake_hold_supported(cp):
from opendbc.car import structs
from opendbc.car.chrysler.values import CAR, ChryslerFlags
return bool(cp is not None and cp.brand == "chrysler" and
cp.carFingerprint in (CAR.JEEP_GRAND_CHEROKEE, CAR.JEEP_GRAND_CHEROKEE_2019) and
cp.flags & ChryslerFlags.JEEP_BRAKE_HOLD_AVAILABLE and cp.pcmCruise and
not cp.openpilotLongitudinalControl and not cp.passive and not cp.dashcamOnly and not cp.notCar and
len(cp.safetyConfigs) == 1 and cp.safetyConfigs[0].safetyModel == structs.CarParams.SafetyModel.chrysler and
cp.safetyConfigs[0].safetyParam in (0, 4))
def apply_brake_hold(cp, requested):
if brake_hold_supported(cp):
cp.safetyConfigs[0].safetyParam = 4 if requested else 0
def create_das_3_command(packer, counter_offset, brake_decel, das_3):
values = das_3.copy()
values.update(ACC_AVAILABLE=1, ACC_ACTIVE=1, ACC_GO=0, ACC_STANDSTILL=0, ACC_DECEL_REQ=1,
ACC_DECEL=brake_decel, ACC_BRK_PREP=0, ENGINE_TORQUE_REQUEST_MAX=0, GR_MAX_REQ=2,
COUNTER=(int(das_3["COUNTER"]) + counter_offset) % 16)
return packer.make_can_msg("DAS_3", 0, values)
@@ -1,8 +1,12 @@
import os
import time
from opendbc.can import CANPacker
from opendbc.car import Bus, DT_CTRL
from opendbc.car.lateral import apply_meas_steer_torque_limits
from opendbc.car.chrysler import chryslercan
from opendbc.car.chrysler.aol_contract import qualified as aol_qualified
from opendbc.car.chrysler.brake_hold import brake_hold_supported, create_das_3_command
from opendbc.car.chrysler.values import CUSW_CARS, RAM_CARS, RAM_DT, CarControllerParams, ChryslerFlags
from opendbc.car.structs import CarState
from opendbc.car.interfaces import CarControllerBase
@@ -17,6 +21,11 @@ class CarController(CarControllerBase):
self.last_lkas_falling_edge = 0
self.lkas_control_bit_prev = False
self.last_button_frame = 0
self.jeep_hold_replay = "REPLAY" in os.environ
self.jeep_decel = -2.0
self.last_das_3_counter = -1
self.jeep_hold_cancelled = False
self.jeep_cruise_was_active = False
self.packer = CANPacker(dbc_names[Bus.pt])
self.params = CarControllerParams(CP)
@@ -100,6 +109,14 @@ class CarController(CarControllerBase):
if _projection is not None:
_projection.emit('chrysler.LKAS_COMMAND', apply_torque, lkas_control_bit, can_sends[-1])
if brake_hold_supported(self.CP) and self.CP.safetyConfigs[0].safetyParam == 4:
withdraw = bool(aol_qualified(self.CP, marked_only=True) and CS.aol_pause_longitudinal and
0 < CS.aol_pause_longitudinal_stamp_ns <= now_nanos and
now_nanos - CS.aol_pause_longitudinal_stamp_ns <= 30_000_000)
self.update_jeep_brake_hold(CC, CS, now_nanos, can_sends, withdraw=withdraw)
else:
CS.brake_hold = False
self.frame += 1
new_actuators = CC.actuators.as_builder()
@@ -109,3 +126,43 @@ class CarController(CarControllerBase):
if _projection is not None:
_projection.finish(new_actuators.torque)
return new_actuators, can_sends
def update_jeep_brake_hold(self, CC, CS, now_nanos, can_sends, *, withdraw=False):
# CAN Event/parser timestamps are BOOTTIME. Card passes MONOTONIC for
# ordinary live Chrysler controllers, and original CAN time during replay.
# Select the established replay contract explicitly; never infer an offset
# or renew the parser's original source timestamp at command time.
now_boot_ns = now_nanos if self.jeep_hold_replay else time.clock_gettime_ns(time.CLOCK_BOOTTIME)
if (not CS.das_3 or not CS.out.canValid or CS.out.canTimeout or
not 0 <= now_boot_ns - CS.das_3_timestamp <= 100_000_000):
CS.brake_hold = False
return
counter = int(CS.das_3["COUNTER"])
changed = counter != self.last_das_3_counter
self.last_das_3_counter = counter
active = CS.out.cruiseState.enabled
if active and not self.jeep_cruise_was_active:
self.jeep_hold_cancelled = False
self.jeep_cruise_was_active = active
if CC.cruiseControl.cancel or CS.jeep_driver_cancel or withdraw:
self.jeep_hold_cancelled = True
if self.jeep_hold_cancelled:
CS.brake_hold = False
return
decel = float(CS.das_3["ACC_DECEL"])
if not CS.brake_hold and CS.out.cruiseState.enabled and decel < -0.5 and CS.out.standstill:
CS.brake_hold = True
self.jeep_decel = -2.0
if (CC.cruiseControl.cancel or CS.jeep_driver_cancel or withdraw or CS.out.gasPressed or CS.out.brakePressed or
CS.out.gearShifter != CarState.GearShifter.drive or not CS.out.standstill or
not CS.out.cruiseState.available or CS.out.accFaulted):
CS.brake_hold = False
return
if not CS.brake_hold:
return
if CS.out.cruiseState.enabled:
self.jeep_decel = max(-3.0, min(-0.5, min(self.jeep_decel, decel)))
return
can_sends.append(create_das_3_command(self.packer, 2 if changed else 3, self.jeep_decel, CS.das_3))
if self.frame % 10 == 0:
can_sends.append(chryslercan.create_cruise_buttons(self.packer, CS.button_counter + 1, 0, resume=True))
@@ -27,6 +27,10 @@ class CarState(CarStateBase):
self.aol_pause_longitudinal = False
self.aol_pause_longitudinal_stamp_ns = 0
self.aol_buttons = {ButtonType.mainCruise: 0, ButtonType.cancel: 0}
self.brake_hold = False
self.das_3 = {}
self.das_3_timestamp = 0
self.jeep_driver_cancel = False
def update(self, can_parsers) -> structs.CarState:
cp = can_parsers[Bus.pt]
@@ -84,6 +88,11 @@ class CarState(CarStateBase):
ret.cruiseState.nonAdaptive = cp_cruise.vl["DAS_4"]["ACC_STATE"] in (1, 2) # 1 NormalCCOn and 2 NormalCCSet
ret.cruiseState.standstill = cp_cruise.vl["DAS_3"]["ACC_STANDSTILL"] == 1
ret.accFaulted = cp_cruise.vl["DAS_3"]["ACC_FAULTED"] != 0
if self.CP.flags & ChryslerFlags.JEEP_BRAKE_HOLD_AVAILABLE:
self.das_3 = dict(cp_cruise.vl["DAS_3"])
self.jeep_driver_cancel = bool(cp.vl["CRUISE_BUTTONS"]["ACC_Cancel"])
self.das_3_timestamp = cp_cruise.ts_nanos["DAS_3"]["COUNTER"]
ret.brakeHoldActive = self.brake_hold
if self.CP.carFingerprint in RAM_CARS:
# Auto High Beam isn't Located in this message on chrysler or jeep currently located in 729 message
@@ -57,6 +57,7 @@ class CarInterface(CarInterfaceBase):
elif candidate in (CAR.JEEP_GRAND_CHEROKEE, CAR.JEEP_GRAND_CHEROKEE_2019):
ret.steerActuatorDelay = 0.2
ret.flags |= ChryslerFlags.JEEP_BRAKE_HOLD_AVAILABLE.value
ret.lateralTuning.init('pid')
ret.lateralTuning.pid.kpBP, ret.lateralTuning.pid.kiBP = [[9., 20.], [9., 20.]]
@@ -0,0 +1,235 @@
"""Real Jeep parser/controller stock-stop retention; no system speed authority."""
import os
import unittest
from unittest.mock import patch
from opendbc.can import CANPacker, CANParser
from opendbc.car import Bus, gen_empty_fingerprint, structs
from opendbc.car.chrysler.brake_hold import apply_brake_hold, brake_hold_supported
from opendbc.car.chrysler.interface import CarInterface
from opendbc.car.chrysler.values import CAR, DBC
def params(candidate=CAR.JEEP_GRAND_CHEROKEE, *, enabled=True):
cp = CarInterface.get_params(candidate, gen_empty_fingerprint(), [], False, False, False)
apply_brake_hold(cp, enabled)
return cp
class PhysicalCan:
def __init__(self, ci):
ci.update([])
self.ci = ci
self.packers = {bus: CANPacker(parser.dbc.name) for bus, parser in ci.can_parsers.items()}
def frames(self, tick, *, active=True, decel=-2.0, main=True, gas=0.0, brake=0, speed=0.0, gear=4, fault=0, omit=()):
values = {
'DAS_3': {
'ACC_ACTIVE': int(active),
'ACC_AVAILABLE': int(main),
'ACC_DECEL': decel,
'ACC_STANDSTILL': 1,
'ACC_FAULTED': fault,
'ENGINE_TORQUE_REQUEST': 13.25,
'DISABLE_FUEL_SHUTOFF': 1,
'COLLISION_BRK_PREP': 1,
'COUNTER': (14 + tick // 2) % 16,
},
'DAS_4': {'ACC_SET_SPEED_KPH': 100, 'ACC_STATE': 3},
'GEAR': {'PRNDL': gear},
'ECM_5': {'Accelerator_Position': gas},
'ESP_1': {'Brake_Pedal_State': brake},
'SPEED_1': {'SPEED_LEFT': speed, 'SPEED_RIGHT': speed},
}
return [
self.packers[bus].make_can_msg(parser.dbc.addr_to_msg[address].name, parser.bus, values.get(parser.dbc.addr_to_msg[address].name, {}))
for bus, parser in self.ci.can_parsers.items()
for address in sorted(parser.addresses)
if parser.dbc.addr_to_msg[address].name not in omit
]
def apply_at_boot(ci, command, now):
with patch('opendbc.car.chrysler.carcontroller.time.clock_gettime_ns', return_value=now):
return ci.apply(command.as_reader(), now)
class TestJeepBrakeHold(unittest.TestCase):
def setUp(self):
prior = os.environ.pop('REPLAY', None)
def restore():
if prior is None:
os.environ.pop('REPLAY', None)
else:
os.environ['REPLAY'] = prior
self.addCleanup(restore)
def ready(self, candidate=CAR.JEEP_GRAND_CHEROKEE):
ci = CarInterface(params(candidate))
stream = PhysicalCan(ci)
cc = structs.CarControl(enabled=True)
for tick in range(20):
ci.update([(100_000_000_000 + tick * 10_000_000, stream.frames(tick))])
apply_at_boot(ci, cc, 100_000_000_000 + tick * 10_000_000)
self.assertTrue(ci.CS.out.canValid and ci.CS.brake_hold)
return ci, stream, cc
def test_two_exact_platforms_stock_ownership_default_off_and_siblings(self):
for candidate in CAR:
for release in (False, True):
cp = CarInterface.get_params(candidate, gen_empty_fingerprint(), [], True, release, False)
expected = candidate in (CAR.JEEP_GRAND_CHEROKEE, CAR.JEEP_GRAND_CHEROKEE_2019)
self.assertEqual(brake_hold_supported(cp), expected)
before = cp.to_dict()
apply_brake_hold(cp, True)
self.assertEqual(cp.safetyConfigs[0].safetyParam, 4 if expected else before['safetyConfigs'][0]['safetyParam'])
self.assertFalse(cp.openpilotLongitudinalControl)
self.assertTrue(cp.pcmCruise)
apply_brake_hold(cp, False)
self.assertEqual(cp.to_dict(), before)
def test_real_parser_copied_fields_decel_counter_wrap_and_repeat(self):
for candidate in (CAR.JEEP_GRAND_CHEROKEE, CAR.JEEP_GRAND_CHEROKEE_2019):
ci, stream, cc = self.ready(candidate)
decoder = CANParser(DBC[candidate][Bus.pt], [('DAS_3', 0)], 0)
for tick in range(20, 55):
now = 100_000_000_000 + tick * 10_000_000
previous = ci.CC.last_das_3_counter
ci.update([(now, stream.frames(tick, active=False))])
_, frames = apply_at_boot(ci, cc, now)
hold = [frame for frame in frames if frame[0] == 0x1F4]
self.assertEqual(len(hold), 1)
decoder.update([(now, hold)])
parsed = decoder.vl['DAS_3']
stock = ci.CS.das_3
self.assertEqual(parsed['COUNTER'], (int(stock['COUNTER']) + (2 if int(stock['COUNTER']) != previous else 3)) % 16)
for name in ('ENGINE_TORQUE_REQUEST', 'DISABLE_FUEL_SHUTOFF', 'COLLISION_BRK_PREP', 'ACC_FAULTED'):
self.assertEqual(parsed[name], stock[name])
for name, expected in {
'ACC_ACTIVE': 1,
'ACC_AVAILABLE': 1,
'ACC_GO': 0,
'ACC_STANDSTILL': 0,
'ACC_DECEL_REQ': 1,
'GR_MAX_REQ': 2,
'ACC_BRK_PREP': 0,
'ENGINE_TORQUE_REQUEST_MAX': 0,
}.items():
self.assertEqual(parsed[name], expected)
self.assertAlmostEqual(parsed['ACC_DECEL'], -2.0, delta=0.004885)
resumes = [frame for frame in frames if frame[0] == 0x23B]
self.assertEqual(bool(resumes), tick % 10 == 0)
def test_driver_main_fault_and_source_withdrawal(self):
for stimulus in ('gas', 'brake', 'speed', 'gear', 'main', 'fault', 'cancel', 'stale'):
ci, stream, cc = self.ready()
args = (
{stimulus: {'gas': 1.0, 'brake': 1, 'speed': 1.0, 'gear': 1, 'main': False, 'fault': 1}.get(stimulus)} if stimulus not in ('cancel', 'stale') else {}
)
now = 100_200_000_000
if stimulus == 'cancel':
cc.cruiseControl.cancel = True
ci.update([(now, stream.frames(20, active=False, **args))])
if stimulus == 'stale':
now += 100_000_001
_, frames = apply_at_boot(ci, cc, now)
self.assertFalse(ci.CS.brake_hold, stimulus)
self.assertFalse(any(frame[0] == 0x1F4 for frame in frames), stimulus)
def test_stock_braking_strength_clamped_to_original_envelope(self):
ci, stream, cc = self.ready()
for tick in range(20, 30):
now = 100_000_000_000 + tick * 10_000_000
ci.update([(now, stream.frames(tick, decel=-4.0))])
apply_at_boot(ci, cc, now)
self.assertEqual(ci.CC.jeep_decel, -3.0)
def test_real_parser_crc_fault_withdraws_hold(self):
ci, stream, cc = self.ready()
for tick in range(20, 32):
now = 100_000_000_000 + tick * 10_000_000
frames = stream.frames(tick, active=False)
for index, (address, data, bus) in enumerate(frames):
if address == 0x1F4:
invalid = bytearray(data)
invalid[-1] ^= 1
frames[index] = (address, bytes(invalid), bus)
ci.update([(now, frames)])
_, emitted = apply_at_boot(ci, cc, now)
self.assertGreater(now - ci.CS.das_3_timestamp, 100_000_000)
self.assertFalse(ci.CS.brake_hold)
self.assertFalse(any(frame[0] == 0x1F4 for frame in emitted))
def test_commanded_cancel_requires_fresh_stock_acc_engagement(self):
ci, stream, cc = self.ready()
cc.cruiseControl.cancel = True
ci.update([(100_200_000_000, stream.frames(20))])
apply_at_boot(ci, cc, 100_200_000_000)
cc.cruiseControl.cancel = False
for tick in range(21, 25):
now = 100_000_000_000 + tick * 10_000_000
ci.update([(now, stream.frames(tick))])
apply_at_boot(ci, cc, now)
self.assertFalse(ci.CS.brake_hold)
ci.update([(100_250_000_000, stream.frames(25, active=False))])
apply_at_boot(ci, cc, 100_250_000_000)
ci.update([(100_260_000_000, stream.frames(26))])
apply_at_boot(ci, cc, 100_260_000_000)
self.assertTrue(ci.CS.brake_hold)
def test_actual_factory_default_off_and_siblings_emit_no_hold(self):
for candidate in CAR:
for requested in (False, True):
if requested and candidate in (CAR.JEEP_GRAND_CHEROKEE, CAR.JEEP_GRAND_CHEROKEE_2019):
continue
ci = CarInterface(params(candidate, enabled=requested))
stream = PhysicalCan(ci)
cc = structs.CarControl()
for tick in range(20):
now = 100_000_000_000 + tick * 10_000_000
ci.update([(now, stream.frames(tick, active=False))])
_, frames = apply_at_boot(ci, cc, now)
self.assertFalse(ci.CS.brake_hold)
self.assertFalse(any(frame[0] == 0x1F4 for frame in frames))
def test_actual_parser_reverse_then_drive_does_not_restore_disabled_stock_long(self):
ci, stream, cc = self.ready()
ci.update([(100_200_000_000, stream.frames(20, active=False, gear=2))])
_, frames = apply_at_boot(ci, cc, 100_200_000_000)
self.assertEqual(ci.CS.out.gearShifter, structs.CarState.GearShifter.reverse)
self.assertFalse(ci.CS.brake_hold)
self.assertFalse(any(frame[0] == 0x1F4 for frame in frames))
ci.update([(100_210_000_000, stream.frames(21, active=False, gear=4))])
_, frames = apply_at_boot(ci, cc, 100_210_000_000)
self.assertFalse(ci.CS.brake_hold)
self.assertFalse(any(frame[0] == 0x1F4 for frame in frames))
def test_original_can_boot_age_is_independent_of_controller_monotonic_time(self):
ci, stream, cc = self.ready()
boot_ns = 100_200_000_000
ci.update([(boot_ns, stream.frames(20, active=False))])
with patch('opendbc.car.chrysler.carcontroller.time.clock_gettime_ns', return_value=boot_ns):
_, frames = ci.apply(cc.as_reader(), boot_ns - 9_000_000_000)
self.assertTrue(ci.CS.brake_hold)
self.assertTrue(any(frame[0] == 0x1F4 for frame in frames))
def test_original_boot_source_future_and_stale_are_not_restamped(self):
for offset in (-1, 100_000_001):
ci, stream, cc = self.ready()
source = 100_200_000_000
ci.update([(source, stream.frames(20, active=False))])
with patch('opendbc.car.chrysler.carcontroller.time.clock_gettime_ns', return_value=source + offset):
_, frames = ci.apply(cc.as_reader(), source)
self.assertFalse(ci.CS.brake_hold)
self.assertFalse(any(frame[0] == 0x1F4 for frame in frames))
def test_replay_uses_original_can_time_in_the_explicit_replay_domain(self):
with patch.dict(os.environ, {'REPLAY': '1'}):
ci, stream, cc = self.ready()
source = 100_200_000_000
ci.update([(source, stream.frames(20, active=False))])
with patch('opendbc.car.chrysler.carcontroller.time.clock_gettime_ns', side_effect=AssertionError('no host clock in original-time replay')):
_, frames = ci.apply(cc.as_reader(), source)
self.assertTrue(any(frame[0] == 0x1F4 for frame in frames))
@@ -13,6 +13,7 @@ class ChryslerSafetyFlags(IntFlag):
RAM_DT = 1
RAM_HD = 2
AOL_PACIFICA_HYBRID_SET_GUARD = 8
JEEP_BRAKE_HOLD = 4
class ChryslerFlags(IntFlag):
@@ -20,6 +21,7 @@ class ChryslerFlags(IntFlag):
HIGHER_MIN_STEERING_SPEED = 1
HAS_BSM = 2 # blind spot monitoring
STEERING_SPEED_BYPASS = 4
JEEP_BRAKE_HOLD_AVAILABLE = 8
@dataclass
+103 -2
View File
@@ -39,6 +39,31 @@ typedef enum {
CHRYSLER_PACIFICA, // plus Jeep
} ChryslerPlatform;
static ChryslerPlatform chrysler_platform;
static bool chrysler_jeep_hold;
static bool chrysler_hold_armed;
static bool chrysler_hold_cancelled;
static uint8_t chrysler_hold_stock[8];
static bool chrysler_hold_drive;
static bool chrysler_hold_gear_seen;
static bool chrysler_hold_gear_initialized;
static uint8_t chrysler_hold_gear_counter;
static uint32_t chrysler_hold_gear_timestamp;
static uint32_t chrysler_hold_stock_timestamp;
static bool chrysler_hold_ready(void) {
const uint32_t now = microsecond_timer_get();
bool healthy = !safety_rx_checks_invalid;
for (int i = 0; i < current_safety_config.rx_checks_len; i++) {
const RxStatus *status = &current_safety_config.rx_checks[i].status;
healthy &= status->msg_seen && status->valid_checksum && status->valid_quality_flag &&
status->wrong_counters < MAX_WRONG_COUNTERS && !status->lagging &&
safety_get_ts_elapsed(now, status->last_timestamp) <= 1000000U;
}
return chrysler_jeep_hold && healthy && chrysler_hold_gear_seen && chrysler_hold_drive &&
safety_get_ts_elapsed(now, chrysler_hold_gear_timestamp) <= 100000U &&
safety_get_ts_elapsed(now, chrysler_hold_stock_timestamp) <= 100000U && acc_main_on && !vehicle_moving && !gas_pressed && !brake_pressed &&
((chrysler_hold_stock[5] & 0xC0U) == 0U);
}
#define CHRYSLER_ADDR(name) ((uint32_t)((chrysler_platform == CHRYSLER_RAM_DT) ? CHRYSLER_RAM_DT_##name : \
((chrysler_platform == CHRYSLER_RAM_HD) ? CHRYSLER_RAM_HD_##name : CHRYSLER_##name)))
@@ -62,6 +87,17 @@ static void chrysler_rx_hook(const CANPacket_t *msg) {
if (msg_matches(msg, CHRYSLER_ADDR(DAS_3), das_3_bus)) {
bool cruise_engaged = GET_BIT(msg, 21U);
pcm_cruise_check(cruise_engaged);
if (chrysler_jeep_hold) {
chrysler_hold_stock_timestamp = microsecond_timer_get();
const bool previous_active = (chrysler_hold_stock[2] & 0x20U) != 0U;
if (cruise_engaged && !previous_active) { chrysler_hold_cancelled = false; }
acc_main_on = GET_BIT(msg, 20U);
for (int i = 0; i < 8; i++) { chrysler_hold_stock[i] = msg->data[i]; }
const unsigned int raw_decel = ((msg->data[2] & 0xFU) << 8) | msg->data[3];
if (chrysler_hold_ready() && !chrysler_hold_cancelled && cruise_engaged && (raw_decel < 3173U)) {
chrysler_hold_armed = true;
}
}
}
// TODO: use the same message for both
@@ -84,6 +120,29 @@ static void chrysler_rx_hook(const CANPacket_t *msg) {
if (msg_matches(msg, CHRYSLER_ADDR(ESP_1), 0U)) {
brake_pressed = ((msg->data[0U] & 0xFU) >> 2U) == 1U;
}
if (chrysler_jeep_hold && !chrysler_hold_ready()) { chrysler_hold_armed = false; }
}
static void chrysler_hold_cancel_rx(const CANPacket_t *msg) {
if (chrysler_jeep_hold && msg_matches(msg, 746U, 0U)) {
// GEAR_CHECKSUM has no checksum algorithm in the Chrysler DBC. Require
// its known size and advancing counter without inventing a CRC contract.
const uint8_t counter = (uint8_t)(msg->data[3] >> 4);
const bool valid = (GET_LEN(msg) == 5U) &&
(!chrysler_hold_gear_initialized || counter == ((chrysler_hold_gear_counter + 1U) & 0xFU));
chrysler_hold_gear_initialized = true;
chrysler_hold_gear_counter = counter;
chrysler_hold_gear_seen = valid;
chrysler_hold_drive = valid && ((msg->data[0] & 7U) == 4U);
if (valid) { chrysler_hold_gear_timestamp = microsecond_timer_get(); }
if (!chrysler_hold_ready()) { chrysler_hold_armed = false; }
}
if (chrysler_jeep_hold && msg_matches(msg, CHRYSLER_CRUISE_BUTTONS, 0U)) {
if ((GET_LEN(msg) != 3U) || (chrysler_get_checksum(msg) != chrysler_compute_checksum(msg)) || GET_BIT(msg, 0U)) {
chrysler_hold_armed = false;
chrysler_hold_cancelled = true;
}
}
}
static bool chrysler_tx_hook(const CANPacket_t *msg) {
@@ -135,17 +194,46 @@ static bool chrysler_tx_hook(const CANPacket_t *msg) {
if (msg->addr == CHRYSLER_ADDR(CRUISE_BUTTONS)) {
const bool is_cancel = msg->data[0] == 1U;
const bool is_resume = msg->data[0] == 0x10U;
const bool allowed = is_cancel || (is_resume && controls_allowed);
if (is_cancel && chrysler_jeep_hold) {
chrysler_hold_armed = false;
chrysler_hold_cancelled = true;
}
const bool hold_resume = chrysler_hold_armed && chrysler_hold_ready() && ((chrysler_hold_stock[2] & 0x20U) == 0U) &&
(chrysler_get_checksum(msg) == chrysler_compute_checksum(msg));
const bool allowed = is_cancel || (is_resume && (controls_allowed || hold_resume));
if (!allowed) {
tx = false;
}
}
if (msg->addr == CHRYSLER_DAS_3) {
const unsigned int raw_decel = ((msg->data[2] & 0xFU) << 8) | msg->data[3];
const unsigned int stock_counter = chrysler_hold_stock[6] >> 4;
const unsigned int counter = msg->data[6] >> 4;
tx &= chrysler_hold_armed && chrysler_hold_ready() && ((chrysler_hold_stock[2] & 0x20U) == 0U) &&
(raw_decel >= 2661U) && (raw_decel <= 3173U) &&
((counter == ((stock_counter + 2U) & 0xFU)) || (counter == ((stock_counter + 3U) & 0xFU))) &&
(msg->data[0] == (chrysler_hold_stock[0] & 0x1FU)) && (msg->data[1] == chrysler_hold_stock[1]) &&
((msg->data[2] & 0xF0U) == ((chrysler_hold_stock[2] & 0x80U) | 0x30U)) &&
(msg->data[4] == 0x12U) && (msg->data[5] == (chrysler_hold_stock[5] & 0xC0U)) &&
((msg->data[6] & 0xFU) == (chrysler_hold_stock[6] & 1U)) &&
(chrysler_get_checksum(msg) == chrysler_compute_checksum(msg));
}
return tx;
}
static safety_config chrysler_init(uint16_t param) {
const uint32_t CHRYSLER_PARAM_RAM_DT = 1U; // set for Ram DT platform
chrysler_jeep_hold = param == 4U;
chrysler_hold_armed = false;
chrysler_hold_cancelled = false;
chrysler_hold_drive = false;
chrysler_hold_gear_seen = false;
chrysler_hold_gear_initialized = false;
chrysler_hold_gear_counter = 0U;
chrysler_hold_gear_timestamp = 0U;
chrysler_hold_stock_timestamp = 0U;
for (int i = 0; i < 8; i++) { chrysler_hold_stock[i] = 0U; }
static RxCheck chrysler_ram_dt_rx_checks[] = {
{.msg = {{CHRYSLER_RAM_DT_EPS_2, 0, 8, 100U, .max_counter = 15U, .ignore_quality_flag = true}, { 0 }, { 0 }}},
@@ -169,6 +257,13 @@ static safety_config chrysler_init(uint16_t param) {
{CHRYSLER_DAS_6, 0, 8, .check_relay = true},
};
static const CanMsg CHRYSLER_JEEP_HOLD_TX_MSGS[] = {
{CHRYSLER_CRUISE_BUTTONS, 0, 3, .check_relay = false},
{CHRYSLER_LKAS_COMMAND, 0, 6, .check_relay = true},
{CHRYSLER_DAS_6, 0, 8, .check_relay = true},
{CHRYSLER_DAS_3, 0, 8, .check_relay = false},
};
static const CanMsg CHRYSLER_RAM_DT_TX_MSGS[] = {
{CHRYSLER_RAM_DT_CRUISE_BUTTONS, 2, 3, .check_relay = false},
{CHRYSLER_RAM_DT_LKAS_COMMAND, 0, 8, .check_relay = true},
@@ -209,6 +304,7 @@ static safety_config chrysler_init(uint16_t param) {
} else {
chrysler_platform = CHRYSLER_PACIFICA;
ret = BUILD_SAFETY_CFG(chrysler_rx_checks, CHRYSLER_TX_MSGS);
if (chrysler_jeep_hold) { SET_TX_MSGS(CHRYSLER_JEEP_HOLD_TX_MSGS, ret); }
}
chrysler_aol_configure(param);
// The allocated Hybrid guard never falls back to ordinary or Ram output.
@@ -219,10 +315,15 @@ static safety_config chrysler_init(uint16_t param) {
return ret;
}
static void chrysler_optional_rx(const CANPacket_t *msg) {
chrysler_aol_optional_rx(msg);
chrysler_hold_cancel_rx(msg);
}
const safety_hooks chrysler_hooks = {
.init = chrysler_init,
.rx = chrysler_rx_hook,
.optional_rx = chrysler_aol_optional_rx,
.optional_rx = chrysler_optional_rx,
.tx = chrysler_tx_hook,
.get_counter = chrysler_get_counter,
.get_checksum = chrysler_get_checksum,
@@ -173,7 +173,7 @@ static void chrysler_aol_optional_rx(const CANPacket_t *msg) {
static void chrysler_aol_configure(uint16_t param) {
chrysler_aol_reset();
chrysler_aol_set_guard = param == 8U;
chrysler_aol_enabled = (alternative_experience == 32) && ((param == 0U) || (param == 1U) || (param == 8U));
chrysler_aol_enabled = (alternative_experience == 32) && ((param == 0U) || (param == 1U) || (param == 4U) || (param == 8U));
if (chrysler_aol_enabled) {
static const AolSafetyPolicy policy = {
.reset = chrysler_aol_reset,
@@ -0,0 +1,205 @@
"""Exact opt-in Jeep native retention, physical permission and copied payload."""
import unittest
from opendbc.can import CANPacker
from opendbc.car import structs
from opendbc.car.chrysler.brake_hold import create_das_3_command
from opendbc.safety.tests.libsafety import libsafety_py
class TestJeepBrakeHoldSafety(unittest.TestCase):
def setUp(self):
self.safety = libsafety_py.libsafety
def select(self, word=4):
self.safety.init_tests()
self.safety.set_alternative_experience(0)
self.assertEqual(self.safety.set_safety_hooks(structs.CarParams.SafetyModel.chrysler, word), 0)
self.packer = CANPacker('chrysler_pacifica_2017_hybrid_generated')
self.stock = {
'ACC_AVAILABLE': 1,
'ACC_ACTIVE': 1,
'ACC_STANDSTILL': 1,
'ACC_DECEL': -2.0,
'ENGINE_TORQUE_REQUEST': 13.25,
'DISABLE_FUEL_SHUTOFF': 1,
'COLLISION_BRK_PREP': 1,
'COUNTER': 0,
}
def packet(self, frame):
return libsafety_py.make_CANPacket(frame[0], frame[2], frame[1])
def gear(self, value):
return self.packer.make_can_msg('GEAR', 0, {'PRNDL': value})
def feed(self, tick, *, active=True, main=True, gas=0.0, brake=0, speed=0.0, fault=0, gear=4, omit_gear=False):
self.safety.set_timer(tick * 10_000)
self.stock.update(ACC_ACTIVE=int(active), ACC_AVAILABLE=int(main), ACC_FAULTED=fault, COUNTER=tick % 16)
if not omit_gear:
self.safety.safety_rx_hook(self.packet(self.gear(gear)))
for name, values in [
('EPS_2', {}),
('ESP_1', {'Brake_Pedal_State': brake}),
('SPEED_1', {'SPEED_LEFT': speed, 'SPEED_RIGHT': speed}),
('ECM_5', {'Accelerator_Position': gas}),
('DAS_3', self.stock),
]:
self.assertTrue(self.safety.safety_rx_hook(self.packet(self.packer.make_can_msg(name, 0, values))))
self.safety.safety_tick()
def ready(self):
self.select()
for tick in range(20):
self.feed(tick)
self.assertTrue(self.safety.get_controls_allowed())
self.feed(20, active=False)
self.assertTrue(self.safety.safety_config_valid())
self.assertFalse(self.safety.get_controls_allowed())
def hold(self, offset=2, decel=-2.0):
return create_das_3_command(self.packer, offset, decel, self.stock)
def test_physical_acc_braking_arms_retention_and_disengaged_resume(self):
self.ready()
for offset in (2, 3):
self.assertTrue(self.safety.safety_tx_hook(self.packet(self.hold(offset))))
button = self.packer.make_can_msg('CRUISE_BUTTONS', 0, {'ACC_Resume': 1})
self.assertTrue(self.safety.safety_tx_hook(self.packet(button)))
for name in ('ACC_Cancel', 'ACC_Resume'):
self.assertFalse(self.safety.safety_tx_hook(self.packet(self.packer.make_can_msg('CRUISE_BUTTONS', 0, {name: 1, 'ACC_Distance_Dec': 1}))))
def test_driver_main_stale_and_unarmed_withdrawal(self):
for change in ('gas', 'brake', 'speed', 'main', 'stale', 'unarmed'):
self.ready()
if change == 'stale':
self.safety.set_timer(1_200_001)
self.safety.safety_tick()
elif change == 'unarmed':
self.select()
self.feed(20, active=False)
else:
self.feed(21, active=False, **{change: {'gas': 1.0, 'brake': 1, 'speed': 1.0, 'main': False}[change]})
self.assertFalse(self.safety.safety_tx_hook(self.packet(self.hold())), change)
self.assertFalse(self.safety.safety_tx_hook(self.packet(self.packer.make_can_msg('CRUISE_BUTTONS', 0, {'ACC_Resume': 1}))), change)
def test_decel_counter_checksum_bus_length_and_copied_field_denials(self):
self.ready()
for decel in (-3.0, -0.5):
self.assertTrue(self.safety.safety_tx_hook(self.packet(self.hold(decel=decel))))
for decel in (-3.01, -0.49, 0.0, 1.0):
self.assertFalse(self.safety.safety_tx_hook(self.packet(self.hold(decel=decel))))
for offset in (0, 1, 4):
self.assertFalse(self.safety.safety_tx_hook(self.packet(self.hold(offset))))
address, data, bus = self.hold()
for invalid_bus, invalid_data in ((1, data), (2, data), (0, data[:7]), (0, data + b'0000')):
self.assertFalse(self.safety.safety_tx_hook(self.packet((address, invalid_data, invalid_bus))))
for signal in ('ACC_GO', 'ACC_STANDSTILL', 'ENGINE_TORQUE_REQUEST_MAX', 'ACC_BRK_PREP'):
fields = self.stock.copy()
fields.update(
ACC_AVAILABLE=1,
ACC_ACTIVE=1,
ACC_GO=0,
ACC_STANDSTILL=0,
ACC_DECEL_REQ=1,
ACC_BRK_PREP=0,
ENGINE_TORQUE_REQUEST_MAX=0,
GR_MAX_REQ=2,
COUNTER=(fields['COUNTER'] + 2) % 16,
)
fields[signal] = 1
self.assertFalse(self.safety.safety_tx_hook(self.packet(self.packer.make_can_msg('DAS_3', 0, fields))), signal)
changed = bytearray(data)
changed[-1] ^= 1
self.assertFalse(self.safety.safety_tx_hook(self.packet((address, bytes(changed), bus))))
def test_permission_word_and_lifecycle_isolation(self):
for word in (0, 1, 2, 3, 5, 6, 7, 8, 12):
self.select(word)
self.assertFalse(self.safety.safety_tx_hook(self.packet(self.hold())), word)
self.ready()
self.select()
self.assertFalse(self.safety.safety_tx_hook(self.packet(self.hold())))
def test_physical_cancel_and_crc_health_revoke_retention(self):
self.ready()
button = self.packer.make_can_msg('CRUISE_BUTTONS', 0, {'ACC_Cancel': 1, 'ACC_Distance_Dec': 1})
self.safety.safety_rx_hook(self.packet(button))
self.assertFalse(self.safety.safety_tx_hook(self.packet(self.hold())))
for tick in range(21, 25):
self.feed(tick, active=False)
self.assertFalse(self.safety.safety_tx_hook(self.packet(self.hold())))
for tick in range(25, 28):
self.feed(tick, active=True)
self.feed(28, active=False)
self.assertTrue(self.safety.safety_tx_hook(self.packet(self.hold())))
address, data, bus = self.packer.make_can_msg('DAS_3', 0, self.stock)
invalid = bytearray(data)
invalid[-1] ^= 1
self.assertFalse(self.safety.safety_rx_hook(self.packet((address, bytes(invalid), bus))))
self.safety.safety_tick()
self.assertFalse(self.safety.safety_tx_hook(self.packet(self.hold())))
def test_crc_valid_inactive_acc_fault_releases_hold_until_reengagement(self):
for fault in (1, 2, 3):
with self.subTest(fault=fault):
self.ready()
self.assertTrue(self.safety.safety_tx_hook(self.packet(self.hold())))
self.feed(21, active=False, fault=fault)
self.assertTrue(self.safety.safety_config_valid())
self.assertFalse(self.safety.get_controls_allowed())
resume = self.packer.make_can_msg('CRUISE_BUTTONS', 0, {'ACC_Resume': 1})
self.assertFalse(self.safety.safety_tx_hook(self.packet(self.hold())))
self.assertFalse(self.safety.safety_tx_hook(self.packet(resume)))
self.feed(22, active=False, fault=0)
self.assertTrue(self.safety.safety_config_valid())
self.assertFalse(self.safety.safety_tx_hook(self.packet(self.hold())))
self.assertFalse(self.safety.safety_tx_hook(self.packet(resume)))
self.feed(23, active=True, fault=0)
self.assertTrue(self.safety.get_controls_allowed())
self.feed(24, active=False, fault=0)
self.assertFalse(self.safety.get_controls_allowed())
self.assertTrue(self.safety.safety_tx_hook(self.packet(self.hold())))
self.assertTrue(self.safety.safety_tx_hook(self.packet(resume)))
def test_native_reverse_pause_and_fresh_drive_reengagement(self):
self.ready()
self.feed(21, active=False, gear=2)
self.assertFalse(self.safety.safety_tx_hook(self.packet(self.hold())))
self.feed(22, active=False, gear=4)
self.assertFalse(self.safety.safety_tx_hook(self.packet(self.hold())))
self.feed(23, active=True, gear=4)
self.feed(24, active=False, gear=4)
self.assertTrue(self.safety.safety_tx_hook(self.packet(self.hold())))
def test_native_gear_and_das_source_expire_independently(self):
self.ready()
for tick in range(21, 33):
self.feed(tick, active=False, omit_gear=True)
self.assertFalse(self.safety.safety_tx_hook(self.packet(self.hold())))
self.ready()
self.safety.set_timer(300_001)
self.assertFalse(self.safety.safety_tx_hook(self.packet(self.hold())))
def test_native_bad_gear_counter_wrong_bus_and_commanded_cancel(self):
self.ready()
address, data, bus = self.gear(1)
self.safety.safety_rx_hook(self.packet((address, data, 1)))
self.assertTrue(self.safety.safety_tx_hook(self.packet(self.hold())))
invalid = bytearray(data)
invalid[3] ^= 0x20
self.safety.safety_rx_hook(self.packet((address, bytes(invalid), bus)))
self.assertFalse(self.safety.safety_tx_hook(self.packet(self.hold())))
self.ready()
cancel = self.packer.make_can_msg('CRUISE_BUTTONS', 0, {'ACC_Cancel': 1})
self.assertTrue(self.safety.safety_tx_hook(self.packet(cancel)))
self.assertFalse(self.safety.safety_tx_hook(self.packet(self.hold())))
def test_repeated_invalid_gear_does_not_renew_permission(self):
self.ready()
address, data, bus = self.gear(4)
self.safety.safety_rx_hook(self.packet((address, data, bus)))
# Same counter is invalid, including when repeated after invalidation.
for _ in range(3):
self.safety.safety_rx_hook(self.packet((address, data, bus)))
self.assertFalse(self.safety.safety_tx_hook(self.packet(self.hold())))
Binary file not shown.
Binary file not shown.
+1
View File
@@ -22,6 +22,7 @@ inline static std::unordered_map<std::string, ParamKeyAttributes> keys = {
{"AlwaysOnLateral", {PERSISTENT, BOOL, "1"}},
{"NostalgiaMode", {PERSISTENT, BOOL, "0"}},
{"ToyotaAutoHold", {PERSISTENT, BOOL, "0"}},
{"JeepBrakeHold", {PERSISTENT, BOOL, "0"}},
{"ToyotaRadarMode", {PERSISTENT, INT, "0"}},
{"RemoteStartBootsComma", {PERSISTENT, BOOL, "0"}},
{"HKGRemoteStartBootsComma", {PERSISTENT, BOOL, "0"}},
@@ -91,7 +91,7 @@ int main() {
auto chrysler = status();
chrysler.safety_mode = 9U;
chrysler.safety_param = static_cast<uint16_t>(word);
assert(aol_capable(chrysler, 9U, VEHICLE_REGISTRY) == ((word == 0U) || (word == 1U) || (word == 8U)));
assert(aol_capable(chrysler, 9U, VEHICLE_REGISTRY) == ((word == 0U) || (word == 1U) || (word == 4U) || (word == 8U)));
chrysler.capability_flags = 0U;
assert(!aol_capable(chrysler, 9U, VEHICLE_REGISTRY));
}
+1 -1
View File
@@ -19,7 +19,7 @@ def policy_for(cp):
def native_profile_supported(model, param):
return model == int(structs.CarParams.SafetyModel.chrysler) and param in (0, 1, 8)
return model == int(structs.CarParams.SafetyModel.chrysler) and param in (0, 1, 4, 8)
def native_accepts_cp(cp, model, param):
@@ -3,7 +3,7 @@
#include "selfdrive/pandad/aol_protocol.h"
inline bool chrysler_aol_param(uint16_t param) {
return (param == 0U) || (param == 1U) || (param == 8U);
return (param == 0U) || (param == 1U) || (param == 4U) || (param == 8U);
}
inline constexpr AolSafetyProfile CHRYSLER_AOL_PROFILE{9U, chrysler_aol_param, true};
@@ -5,7 +5,8 @@ from openpilot.starpilot.vehicle_selection import read_selection
STARTUP_KEYS = ('AlphaLongitudinalEnabled', 'IsReleaseBranch', 'OpenpilotEnabledToggle', 'SafeMode',
'DisableOpenpilotLongitudinal', 'ToyotaAutoHold', 'ToyotaRadarMode', 'VoltSNG', 'GMAutoHold', 'VoltOnePedalMode', 'TurnAssist', 'LongPitch',
'DisableOpenpilotLongitudinal', 'ToyotaAutoHold', 'JeepBrakeHold',
'ToyotaRadarMode', 'VoltSNG', 'GMAutoHold', 'VoltOnePedalMode', 'TurnAssist', 'LongPitch',
'HondaBoschARadar', 'NAPPedalEnabled', 'NAPRadarEnabled', 'NAPRadarBehindNosecone')
@@ -74,7 +74,7 @@ from openpilot.starpilot.controllers.toyota_cruise import capability as toyota_c
BOOL_DEFAULTS = {
"GMPedalLongitudinal": False, "ForceStops": False, "AlwaysAllowUploads": False, "TurnAssist": True,
"ReverseCruise": False, "ToyotaAutoHold": False, "VoltSNG": False, "GMAutoHold": False, "VoltOnePedalMode": False,
"ReverseCruise": False, "ToyotaAutoHold": False, "JeepBrakeHold": False, "VoltSNG": False, "GMAutoHold": False, "VoltOnePedalMode": False,
"RemapCancelToDistance": False, "LongPitch": True, "DisableOpenpilotLongitudinal": False,
"SpeedLimitController": False, "ShowSpeedLimits": True,
"SLCConfirmation": False,
@@ -318,6 +318,28 @@ class FeatureSettingsOwner:
return commit_exact(self.params, key=request.key, max_bytes=128, raw=b"1" if request.value == "On" else b"0",
expected=request.expected, authorized=authorized, temp_prefix=".gm-stop-").verified
def _jeep_hold_capability(self):
from opendbc.car.chrysler.brake_hold import brake_hold_supported
cp = self.vehicle_params()
try:
if not brake_hold_supported(cp):
return None
return (str(cp.carFingerprint), int(cp.flags),
tuple((str(config.safetyModel), int(config.safetyParam)) for config in cp.safetyConfigs),
str(cp.carVin) if cp.carVin else None)
except (AttributeError, TypeError, ValueError):
return None
def _apply_jeep_hold(self, request):
if not request.confirmation or request.value not in ("Off", "On") or request.dependencies:
return False
def authorized():
return (bool(request.vehicle_fingerprint) and self.vehicle_fingerprint() == request.vehicle_fingerprint and
request.capability is not None and request.capability == self._jeep_hold_capability() and
self.authority("vehicle") and self.authority("parked_preferences") and request.expected in (None, b"0", b"1"))
return commit_exact(self.params, key="JeepBrakeHold", max_bytes=128, raw=b"1" if request.value == "On" else b"0",
expected=request.expected, authorized=authorized, temp_prefix=".jeep-hold-").verified
def _auto_hold_capability(self) -> tuple | None:
from opendbc.car.toyota.interface import toyota_auto_hold_supported
cp = self.vehicle_params()
@@ -713,7 +735,8 @@ class FeatureSettingsOwner:
FeatureRow("", "Always On Lateral", "Keep steering assistance active independently of cruise control", page=FeaturePage.AOL, available=True),
FeatureRow("", "Wheel Controls", "Assign steering-wheel buttons and cruise behavior", page=FeaturePage.WHEEL, available=True)]
if (self._long_pitch_capability() is not None or self._bolt_disable_capability() is not None or
self._auto_hold_capability() is not None or self._toyota_radar_capability() is not None or self._pedal_setup_capability() is not None or
self._auto_hold_capability() is not None or self._jeep_hold_capability() is not None or
self._toyota_radar_capability() is not None or self._pedal_setup_capability() is not None or
self._gm_stop_capability("VoltSNG") is not None or
self._gm_stop_capability("GMAutoHold") is not None or self.tesla_screen.capability() is not None or
self.gm_tune.capability() is not None or self.gm_truck.capability() is not None or
@@ -740,6 +763,12 @@ class FeatureSettingsOwner:
reason=("Optional RAV4 2022/2023 receiver. Off retains the original radar choice. Applies after startup; " +
"native safety and ECU configuration are unchanged."),
capability=radar_capability, default_value="Off"))
jeep_capability = self._jeep_hold_capability()
if jeep_capability is not None:
row = self._bool_row("JeepBrakeHold", "Jeep Brake Hold",
configurable and parked and self.authority("vehicle") and self.authority("parked_preferences"),
"Retains a cruise-controlled stop until you press a pedal or cancel cruise. Applies after the next startup")
rows.append(replace(row, available=row.available and self._readable("JeepBrakeHold"), capability=jeep_capability))
rows.extend(self.tesla_screen.rows(parked))
rows.extend(self.gm_cancel_remap.rows(parked))
@@ -1200,6 +1229,8 @@ class FeatureSettingsOwner:
return self._apply_gm_stop(request)
if key == "LongPitch":
return self._apply_long_pitch(request)
if key == "JeepBrakeHold":
return self._apply_jeep_hold(request)
if key == "ToyotaAutoHold":
return self._apply_auto_hold(request)
if key == "ToyotaRadarMode":
@@ -0,0 +1,87 @@
"""Saved Jeep setting reaches exact startup permission and settings revision."""
from dataclasses import replace
import tempfile
import json
import unittest
from unittest.mock import patch
from opendbc.car import gen_empty_fingerprint
from opendbc.car.chrysler.interface import CarInterface
from opendbc.car.chrysler.values import CAR
from openpilot.common.params import Params, params_put, params_get, _copy_string
from openpilot.starpilot.ui.feature_settings_owner import FeatureSettingsOwner
from openpilot.starpilot.ui.feature_settings_state import row_change
from openpilot.starpilot.vehicle_preferences import VehicleStartupPreferences
from openpilot.starpilot.galaxy import vehicle_configuration as configuration
from openpilot.starpilot.vehicle_selection import encode
def put_stored(params, key, value):
assert type(value) is bytes
if key == 'VehicleSelection':
params.put(key, json.loads(value), block=True)
elif value in (b'0', b'1'):
params.put_bool(key, value == b'1', block=True)
else:
assert params_put(params.p, params.check_key(key), value, len(value), True) == 0
def get_raw(params, key):
return _copy_string(params_get(params.p, params.check_key(key), False))
class TestJeepBrakeHoldPreference(unittest.TestCase):
def test_real_owner_saved_setting_and_strict_next_startup(self):
with tempfile.TemporaryDirectory() as directory:
saved = Params(directory)
cp = CarInterface.get_params(CAR.JEEP_GRAND_CHEROKEE, gen_empty_fingerprint(), [], False, False, False)
groups = {'vehicle': True, 'parked_preferences': True}
owner = FeatureSettingsOwner(saved, lambda group: groups.get(group, True), vehicle_fingerprint=lambda: cp.carFingerprint, vehicle_params=lambda: cp)
row = next(
row for row in owner.snapshot('vehicle', parked=True, system_long=False, lateral_context=True, metric=False).rows if row.key == 'JeepBrakeHold'
)
request = replace(row_change(row), confirmation=True)
for group in groups:
groups[group] = False
self.assertFalse(owner.apply(request))
self.assertIsNone(saved.get('JeepBrakeHold'))
groups[group] = True
self.assertTrue(owner.apply(request))
self.assertEqual(get_raw(saved, 'JeepBrakeHold'), b'1')
for raw, safe, enabled, word in (
(b'1', b'0', True, 4),
(b'0', b'0', True, 0),
(b'bad', b'0', True, 0),
(b'1', b'1', True, 0),
(b'1', b'bad', True, 0),
(b'1', b'0', False, 0),
):
put_stored(saved, 'JeepBrakeHold', raw)
put_stored(saved, 'SafeMode', safe)
target = CarInterface.get_params(CAR.JEEP_GRAND_CHEROKEE, gen_empty_fingerprint(), [], False, False, False)
VehicleStartupPreferences.read(saved, enabled=enabled).prepare(target)
self.assertEqual(target.safetyConfigs[0].safetyParam, word)
self.assertEqual(get_raw(saved, 'JeepBrakeHold'), raw)
cp.carFingerprint = CAR.DODGE_DURANGO
self.assertFalse(owner.apply(request))
def test_actual_galaxy_startup_interleaving_and_physical_revision(self):
with tempfile.TemporaryDirectory() as directory:
saved = Params(directory)
put_stored(saved, 'VehicleSelection', encode(str(CAR.JEEP_GRAND_CHEROKEE)))
saved.put_bool('OpenpilotEnabledToggle', True, block=True)
saved.put_bool('JeepBrakeHold', False, block=True)
real = configuration._factory_context
def interleaved(*args):
result = real(*args)
saved.put_bool('JeepBrakeHold', True, block=True)
return result
with patch.object(configuration, '_factory_context', side_effect=interleaved):
self.assertEqual(configuration.configuration_context(saved, None, None), (None, None))
cp = CarInterface.get_params(CAR.JEEP_GRAND_CHEROKEE, gen_empty_fingerprint(), [], False, False, False)
_, first = configuration.configuration_context(saved, cp, b'physical')
saved.put_bool('JeepBrakeHold', False, block=True)
_, second = configuration.configuration_context(saved, cp, b'physical')
self.assertNotEqual(first, second)
+8 -1
View File
@@ -19,6 +19,7 @@ def bolt_disable_supported(cp) -> bool:
@dataclass(frozen=True)
class VehicleStartupPreferences:
toyota_auto_hold: bool = False
jeep_brake_hold: bool = False
turn_assist: bool = False
gm_long_pitch: bool = True
disable_bolt_long: bool = False
@@ -65,6 +66,8 @@ class VehicleStartupPreferences:
requested = read_saved(params, "ToyotaAutoHold", 8) == (b"1", True)
safe, readable = read_saved(params, "SafeMode", 8)
toyota = bool(enabled and requested and readable and safe in (None, b"0"))
jeep_hold = bool(enabled and read_saved(params, "JeepBrakeHold", 8) == (b"1", True) and
readable and safe in (None, b"0"))
volt_sng = bool(enabled and read_saved(params, "VoltSNG", 8) == (b"1", True) and
readable and safe in (None, b"0"))
gm_auto_hold = bool(enabled and not disable_bolt and read_saved(params, "GMAutoHold", 8) == (b"1", True) and
@@ -91,7 +94,7 @@ class VehicleStartupPreferences:
screen_brake = read_saved(params, 'TeslaAOLDisengageOnBrake', 8) == (b'1', True)
return cls(toyota_radar_mode=toyota_radar_mode(params, enabled=enabled),
ford_lateral_mode=ford_mode(params, enabled=enabled), tesla_screen=screen,
tesla_screen_brake=screen_brake, toyota_auto_hold=toyota,
tesla_screen_brake=screen_brake, toyota_auto_hold=toyota, jeep_brake_hold=jeep_hold,
gm_longitudinal_tune=selected_tune(params, enabled=enabled),
toyota_lock_doors=bool(enabled and readable and safe in (None, b"0") and
read_saved(params, "LockDoors", 8) == (b"1", True)),
@@ -159,6 +162,8 @@ class VehicleStartupPreferences:
from opendbc.car.tesla.screen_button import apply_screen_button
apply_screen_button(cp, self.tesla_screen, self.tesla_screen_brake)
self._prepare_honda_radar(cp)
from opendbc.car.chrysler.brake_hold import apply_brake_hold
apply_brake_hold(cp, self.jeep_brake_hold)
if not self.toyota_filter:
from opendbc.car.toyota.prius_longitudinal import prepare_stock as prepare_prius_stock
prepare_prius_stock(cp)
@@ -186,6 +191,8 @@ class VehicleStartupPreferences:
from opendbc.car.tesla.screen_button import apply_screen_button
apply_screen_button(cp, self.tesla_screen, self.tesla_screen_brake)
self._prepare_honda_radar(cp)
from opendbc.car.chrysler.brake_hold import apply_brake_hold
apply_brake_hold(cp, self.jeep_brake_hold)
if not self.toyota_filter:
from opendbc.car.toyota.prius_longitudinal import prepare_stock as prepare_prius_stock
prepare_prius_stock(cp)